公平地說，大多數(shù)Hackers 和 Slackers讀者都有一個共同點(diǎn)：我們喜歡用 Python 編寫東西。這并不使我們與眾不同；相反，我們是獨(dú)一無二的。它反映了一個眾所周知且易于解釋的現(xiàn)象，即數(shù)據(jù)科學(xué)家/工程師進(jìn)入（以及最近離開）以前為軟件工程保留的空間：多用途編程語言。盡管這些學(xué)科彼此之間有多么獨(dú)特，但我們有一個共同的特征。引用披頭士樂隊的話，我們需要的是 Python??。
然而，每一次旅程都有一個決定性的時刻，很明顯，這門語言已經(jīng)度過了它的30 歲生日。自 Guido 釋放 Serpent 來解決1991 年的問題以來，已經(jīng)過去了三十年。這是冷戰(zhàn)時代的最后一年：歷史上的一個不同時期，在計算領(lǐng)域更是如此。Python 背后的大多數(shù)設(shè)計決策在當(dāng)時都是明智的，但其中一些決策已成為當(dāng)今的“怪癖”。最具爭議性的怪癖很容易就是并發(fā)話題。
事實上，我指的是全局解釋器鎖（GIL）。我會讓你免去貶低 GIL 的痛苦，因為其他人在這方面做得比我好得多（如果你對細(xì)節(jié)感興趣并且有時間，我強(qiáng)烈推薦一篇題為The GIL 的文章，以及它對 Python 多線程的影響）。GIL 的長處和短處在于它限制了 Python 有效利用多個 CPU 核心，讓您的 8 核筆記本電腦在單個核心上運(yùn)行 Python 腳本，而其他核心則閑置。
并發(fā)性是一個遠(yuǎn)遠(yuǎn)超出 Python 范圍的復(fù)雜問題。大多數(shù)編程語言都有相似的命運(yùn)。但我們來這里并不是為了哀嘆我們的處境；而是為了我們的處境。我們在這里討論異步 I/O。
在 Python 中同時做多件事
并發(fā)是編程中的一個廣泛概念，可以歸結(jié)為“同時做一堆事情”。并發(fā)運(yùn)行的代碼通常采用兩種可能的形式之一：
任務(wù)輪流執(zhí)行，以盡量減少同事任務(wù)的停機(jī)時間。
真正并行的任務(wù)同時并行運(yùn)行。
Python 附帶了兩個模塊，分別處理這兩種方法：
線程（單進(jìn)程）：Python 的線程模塊僅限于在給定時間使用單個處理器。線程模塊可以管理占用給定線程的任務(wù)。任務(wù) X 一直運(yùn)行，直到被外部因素阻止（例如等待對 HTTP 請求的響應(yīng)）。同時，任務(wù) Y 優(yōu)先執(zhí)行它，直到任務(wù) X 準(zhǔn)備好繼續(xù)（因此“阻塞 I/O”）。
多處理（多個處理器）：多處理模塊使代碼能夠并行運(yùn)行。腳本被初始化并在n 個CPU上同時運(yùn)行n次。將單條道路擴(kuò)展為 8 車道高速公路具有明顯的性能優(yōu)勢，直到需要整合每項任務(wù)的結(jié)果為止。多個進(jìn)程無法同時將數(shù)據(jù)寫入同一目標(biāo)（數(shù)據(jù)庫、文件等）而不創(chuàng)建相互阻塞的鎖。對于旨在產(chǎn)生輸出的腳本來說，嘗試解決這個問題幾乎肯定是一項難以克服的努力。
Asyncio 適用于哪里？
Asyncio是上述方法的第三種也是通常首選的替代方法。盡管僅限于單個線程，Asyncio可以比 Python 的本機(jī)單線程執(zhí)行速度快得多地執(zhí)行大量操作。為了說明這是如何可能的，請考慮人類如何傾向于“同時處理多項任務(wù)”。當(dāng)人們聲稱自己是“多任務(wù)處理”時，他們通常是通過在任務(wù)之間切換來完成工作，而不是同時做多件事。單線程程序以同樣的方式異步：通過重疊工作來優(yōu)化輸出。
雖然人類在多任務(wù)處理方面表現(xiàn)不佳是出了名的，但機(jī)器可以從這種做法中看到顯著的性能優(yōu)勢。它們通常更適合在沒有額外開銷的情況下啟動和停止工作。這個概念是FastAPI等框架的“秘密武器” ，F(xiàn)astAPI 是一個異步 Python 框架，它自稱性能“與NodeJS和Go相當(dāng) ” （我保證下次會在 FastAPI 上寫一篇公平的文章）。
我花了一段時間才打消了我的懷疑，即單個線程如何同時處理多個任務(wù)可以提供值得一寫的性能優(yōu)勢。直到我偶然發(fā)現(xiàn)事件循環(huán)的概念，事情才開始有意義。
事件循環(huán)
我們從積壓的 I/O“任務(wù)”開始。它們可以是 HTTP 請求，將內(nèi)容保存到磁盤，或者在我們的例子中，兩者的混合。同步 Python 工作流程（讀作：標(biāo)準(zhǔn) Python）將從開始到結(jié)束一次執(zhí)行一項任務(wù)。這就像在車管所排隊等候，那里只有一根線，而這位女士討厭你們所有人。
事件循環(huán)以不同的方式處理事情，以便更快地完成任務(wù)。給定許多任務(wù)，事件“循環(huán)”通過獲取新任務(wù)并將它們委托給線程來工作。該循環(huán)不斷檢查正在進(jìn)行的任務(wù)是否有停機(jī)（或完成）。當(dāng)委派任務(wù)“等待”外部因素（例如 HTTP 請求）時，事件循環(huán)會通過啟動線程中的另一個任務(wù)來填充死區(qū)時間。如果事件循環(huán)發(fā)現(xiàn)分配的任務(wù)已完成，則該任務(wù)將從其線程中刪除，收集該任務(wù)的輸出，并且循環(huán)從隊列中選擇另一個任務(wù)來占用該線程：
異步 I/O 事件循環(huán)
與 DMV 線路不同，事件循環(huán)的工作方式與餐廳有些相似（請繼續(xù)我的說法）。盡管有許多桌子和一個廚房，但服務(wù)器通過在桌子（任務(wù)）和廚房（線程）之間輪換來處理“積壓”。在廚房里連續(xù)準(zhǔn)備多個食物訂單比接受單個訂單并等待它們被創(chuàng)建/提供給下一個顧客之前要高效得多。
協(xié)程：異步運(yùn)行的函數(shù)
異步 Python 腳本不定義函數(shù)- 它們定義協(xié)程。協(xié)程（用 定義async def，而不是def）可以在完成之前停止執(zhí)行，通常等待另一個協(xié)程完成。下面的代碼片段演示了協(xié)程的最簡單示例：
"""Define a Coroutine function to be executed asynchronously."""
import asyncio
from logger import LOGGER
async def simple_coroutine(number: int):
"""
Wait for a time delay & display number associated with coroutine.
:param int number: Number to identify the current coroutine.
"""
await asyncio.sleep(1)
LOGGER.info(f"Coroutine {number} has finished executing.")協(xié)程.py
simple_coroutine()在記錄消息之前暫停執(zhí)行 1 秒。協(xié)程不能像常規(guī)函數(shù)那樣被調(diào)用；除非在 asyncio 事件循環(huán)內(nèi)運(yùn)行，否則嘗試運(yùn)行simple_coroutine(1)將不起作用。幸運(yùn)的是，創(chuàng)建事件循環(huán)很容易：
import asyncio from coroutines import simple_coroutine # Import our coroutine asyncio.run(simple_coroutine(1))
運(yùn)行協(xié)程
asyncio.run()創(chuàng)建一個事件循環(huán)，并運(yùn)行傳遞給它的協(xié)程。當(dāng)您的腳本有一個所有邏輯源自的入口點(diǎn)時，創(chuàng)建事件循環(huán)是最好的?；蛘撸琣syncio.gather()如果您只想執(zhí)行少量協(xié)程，則可以接受任意數(shù)量的協(xié)程：
import asyncio from coroutines import simple_coroutine # Import our coroutine asyncio.gather( simple_coroutine(1) simple_coroutine(2) simple_coroutine(3) )
在事件循環(huán)內(nèi)運(yùn)行 3 個協(xié)程
運(yùn)行此腳本將執(zhí)行所有三個協(xié)程并記錄以下內(nèi)容：
1 2 3
asyncio.gather()三個協(xié)程的輸出
您認(rèn)為完成上述操作需要多長時間？3秒，也許？或者我們是否能夠通過魔法來優(yōu)化我們的代碼？
您可能會驚訝地發(fā)現(xiàn)，運(yùn)行上述代碼始終能在幾乎一秒內(nèi)執(zhí)行（或者在糟糕的一天偶爾會執(zhí)行1.01 秒）。如果我們使用 Python 的內(nèi)置time.perf_counter()來計算函數(shù)的執(zhí)行時間，我們可以直接看到這一點(diǎn)：
import asyncio
import time
from coroutines import simple_coroutine # Import our coroutine
def async_gather_example()
start_time = time.perf_counter()
asyncio.gather(
simple_coroutine(1)
simple_coroutine(2)
simple_coroutine(3)
)
print(
f"Executed {__name__} in {time.perf_counter() - start_time:0.2f} seconds."
)
async_example()跟蹤執(zhí)行 3 個休眠 1 秒的協(xié)程的執(zhí)行時間
果然，該腳本幾乎只花了1 秒：
Executed async_example in 1.01 seconds.
輸出async_example()
我們的協(xié)程simple_coroutine()需要 1 秒才能自行執(zhí)行。上面令人印象深刻的是，我們調(diào)用了這個協(xié)程 3 次，運(yùn)行時間接近 1 秒，而同步Python 腳本確實需要 3 秒。更重要的是，執(zhí)行這些任務(wù)的開銷僅不到.01幾秒，這意味著我們的協(xié)程幾乎同時完成。
使用任務(wù)
asyncio.gather()在上面的示例中，我們回避了 Asyncio 中的一個基本數(shù)據(jù)結(jié)構(gòu)：Task.
協(xié)程是可以異步運(yùn)行的函數(shù)。當(dāng)以特定方式運(yùn)行數(shù)百或數(shù)千個此類函數(shù)時，如果能夠“管理”這些函數(shù)，那就太好了。了解協(xié)程何時失?。ㄒ约叭绾翁幚硭?，或者只是檢查循環(huán)當(dāng)前正在處理哪個協(xié)程，特別是當(dāng)我們的事件循環(huán)可能需要幾分鐘或幾小時才能執(zhí)行或有可能失敗時。
管理任務(wù)
在更復(fù)雜的工作流程中，任務(wù)提供了幾種有用的方法來幫助我們管理正在執(zhí)行的任務(wù)：
.set_name([name])（和.get_name()）：為任務(wù)命名，以便以人類可讀的方式來識別哪個任務(wù)。
.cancel(msg=[message])：取消事件循環(huán)中的任務(wù)，允許循環(huán)繼續(xù)執(zhí)行其他任務(wù)。對于無響應(yīng)或不太可能完成的任務(wù)很有用。
.canceled()：返回True任務(wù)是否被取消，否則False返回。
.done()：返回True任務(wù)是否成功完成，否則False返回。
.result()：返回任務(wù)的結(jié)果。canceled任務(wù)將包含有關(guān)任務(wù)被取消原因的異常消息，而done任務(wù)將僅返回done。尚未調(diào)用的任務(wù)將返回InvalidStateError異常。
許多其他方法都可以在Asyncio 的 Task 文檔中找到。
創(chuàng)建任務(wù)
Coroutine使用 Asyncio包裝sTask很簡單。之前的運(yùn)行asyncio.gather()為我們解決了這個問題，但這只是一種捷徑，使我們無法利用任務(wù)的優(yōu)勢，因為任務(wù)被實例化為通用對象并立即執(zhí)行。如果我們事先創(chuàng)建任務(wù)，我們可以將元數(shù)據(jù)與它們關(guān)聯(lián)起來，并在準(zhǔn)備好時在事件循環(huán)中執(zhí)行它們。
我們將創(chuàng)建一個名為 的新協(xié)程 create_tasks()，該協(xié)程將：
創(chuàng)建n 個Task 實例simple_coroutine()。
在創(chuàng)建時為每個任務(wù)分配一個名稱。
以 Python 列表的形式返回所有任務(wù)，稍后可以通過事件循環(huán)執(zhí)行：
"""Create multiple tasks from a Coroutine."""
import asyncio
from asyncio import Task
from typing import List
from logger import LOGGER
from asyncio_intro_part1.coroutines import simple_coroutine
async def create_tasks(num_tasks: int) -> List[Task]:
"""
Create n number of asyncio tasks to be executed.
:param int num_tasks: Number of tasks to create.
:returns: List[Task]
"""
task_list = []
LOGGER.info(f"Creating {num_tasks} tasks to be executed...")
for i in range(num_tasks):
task = asyncio.create_task(
simple_coroutine(i),
name=f"Task #{i}"
)
task_list.append(task)
LOGGER.info(f"Created Task: {task}")
return task_list任務(wù).py
行動中的任務(wù)
定義了我們的create_tasks()方法后，就到了有趣的部分了：查看任務(wù)的創(chuàng)建、執(zhí)行和完成。在項目的根部，我們將定義最后一個函數(shù)async_tasks_example()來演示這一點(diǎn)：
...
from .tasks import create_tasks
async def async_tasks_example():
"""Create and inspect tasks to wrap simple functions."""
task_list = await create_tasks(5)
done, pending = await asyncio.wait(task_list)
if done:
LOGGER.success(
f"{len(done)} tasks completed: {[task.get_name() for task in done]}."
)
if pending:
LOGGER.warning(
f"{len(done)} tasks pending: {[task.get_name() for task in pending]}."
)__init__.py
我們首先將通過創(chuàng)建的 5 個任務(wù)分配create_tasks()給該task_list變量。發(fā)生這種情況時，我們會看到在tasks.py中添加的正確日志記錄：
17:00:53 PM | INFO: Creating 5 tasks to be executed... 17:00:53 PM | INFO: Created Task: <Task pending name='Task #0' coro=<simple_coroutine() running at /Users/toddbirchard/Projects/asyncio-tutorial-part1/asyncio_intro_part1/coroutines.py:7>> 17:00:53 PM | INFO: Created Task: <Task pending name='Task #1' coro=<simple_coroutine() running at /Users/toddbirchard/Projects/asyncio-tutorial-part1/asyncio_intro_part1/coroutines.py:7>> 17:00:53 PM | INFO: Created Task: <Task pending name='Task #2' coro=<simple_coroutine() running at /Users/toddbirchard/Projects/asyncio-tutorial-part1/asyncio_intro_part1/coroutines.py:7>> 17:00:53 PM | INFO: Created Task: <Task pending name='Task #3' coro=<simple_coroutine() running at /Users/toddbirchard/Projects/asyncio-tutorial-part1/asyncio_intro_part1/coroutines.py:7>> 17:00:53 PM | INFO: Created Task: <Task pending name='Task #4' coro=<simple_coroutine() running at /Users/toddbirchard/Projects/asyncio-tutorial-part1/asyncio_intro_part1/coroutines.py:7>>
在tasks.py中創(chuàng)建5個任務(wù)的輸出文章來源：http://www.zghlxwxcb.cn/article/583.html
我們隨后通過 執(zhí)行這五個任務(wù)asyncio.wait(task_list)。asyncio.wait()嘗試完成 task_list 中的所有任務(wù)并返回“已完成”和“待處理”任務(wù)的元組。由于添加了一些日志記錄和任務(wù)名稱的存在，我們可以確認(rèn)所有任務(wù)均已成功完成：文章來源地址http://www.zghlxwxcb.cn/article/583.html
17:00:54 PM | INFO: Coroutine 0 has finished executing. 17:00:54 PM | INFO: Coroutine 1 has finished executing. 17:00:54 PM | INFO: Coroutine 2 has finished executing. 17:00:54 PM | INFO: Coroutine 3 has finished executing. 17:00:54 PM | INFO: Coroutine 4 has finished executing. 17:00:54 PM | SUCCESS: 5 tasks completed: ['Task #1', 'Task #4', 'Task #3', 'Task #0', 'Task #2'].
到此這篇關(guān)于使用 Asyncio 進(jìn)行異步 Python 簡介的文章就介紹到這了,更多相關(guān)內(nèi)容可以在右上角搜索或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
