既然你已經(jīng)對(duì)異步 I/O 這種并發(fā)模型有了一定的了解，現(xiàn)在是時(shí)候探索 Python 的實(shí)現(xiàn)方式了。Python 的 `asyncio` 包以及其相關(guān)的兩個(gè)關(guān)鍵字 `async` 和 `await`，各自有不同的用途，但它們共同幫助你聲明、構(gòu)建、執(zhí)行和管理異步代碼。

協(xié)程和協(xié)程函數(shù) 異步 I/O 的核心是協(xié)程的概念，協(xié)程是一種可以暫停其執(zhí)行并在稍后恢復(fù)的對(duì)象。在此期間，它可以將控制權(quán)交給事件循環(huán)，事件循環(huán)可以執(zhí)行另一個(gè)協(xié)程。協(xié)程對(duì)象是通過(guò)調(diào)用協(xié)程函數(shù)（也稱為異步函數(shù)）生成的。你可以使用 async def 構(gòu)造來(lái)定義一個(gè)協(xié)程函數(shù)。

在編寫你的第一段異步代碼之前，請(qǐng)考慮以下同步運(yùn)行的示例：

import time

def count():
    print("One")
    time.sleep(1)
    print("Two")
    time.sleep(1)

def main():
    for _ in range(3):
        count()

if __name__ == "__main__":
    start = time.perf_counter()
    main()
    elapsed = time.perf_counter() - start
    print(f"{__file__} executed in {elapsed:0.2f} seconds.")

`count()` 函數(shù)打印 “One”，然后等待一秒，接著打印 “Two”，再等待一秒。`main()` 函數(shù)中的循環(huán)執(zhí)行了三次 `count()`。在 `if __name__ == "__main__"` 條件下，你在執(zhí)行開(kāi)始時(shí)記錄當(dāng)前時(shí)間，調(diào)用 `main()`，計(jì)算總時(shí)間，并將其顯示在屏幕上。

當(dāng)你運(yùn)行這個(gè)腳本時(shí)，你會(huì)得到以下輸出：

$ python countsync.py
One
Two
One
Two
One
Two
countsync.py executed in 6.03 seconds.

該腳本交替打印“One”和“Two”，每次打印操作之間間隔一秒。總共運(yùn)行時(shí)間略多于六秒。

如果你將此腳本更新為使用 Python 的異步 I/O 模型，那么它看起來(lái)會(huì)像下面這樣：

import asyncio

async def count():
    print("One")
    await asyncio.sleep(1)
    print("Two")
    await asyncio.sleep(1)

async def main():
    await asyncio.gather(count(), count(), count())

if __name__ == "__main__":
    import time

    start = time.perf_counter()
    asyncio.run(main())
    elapsed = time.perf_counter() - start
    print(f"{__file__} executed in {elapsed:0.2f} seconds.")

現(xiàn)在，你使用 `async` 關(guān)鍵字將 `count()` 轉(zhuǎn)換為一個(gè)協(xié)程函數(shù)，它打印 “One”，等待一秒，然后打印 “Two”，再等待一秒。你使用 `await` 關(guān)鍵字等待 `asyncio.sleep()` 的執(zhí)行。這將控制權(quán)交還給程序的事件循環(huán)，表示：“我要睡一秒鐘。你趁機(jī)運(yùn)行其他任務(wù)吧。”

`main()` 函數(shù)是另一個(gè)協(xié)程函數(shù)，它使用 `asyncio.gather()` 并發(fā)運(yùn)行三個(gè) `count()` 實(shí)例。你使用 `asyncio.run()` 函數(shù)啟動(dòng)事件循環(huán)并執(zhí)行 `main()`。

將這個(gè)版本的性能與同步版本進(jìn)行比較：

$ python countasync.py
One
One
One
Two
Two
Two
countasync.py executed in 2.00 seconds.

得益于異步 I/O 方法，總執(zhí)行時(shí)間僅為兩秒多一點(diǎn)，而不是六秒，這展示了 `asyncio` 在處理 I/O 密集型任務(wù)時(shí)的高效性。

雖然使用 `time.sleep()` 和 `asyncio.sleep()` 可能看起來(lái)很平常，但它們是耗時(shí)過(guò)程的替代品，這些過(guò)程涉及等待時(shí)間。調(diào)用 `time.sleep()` 可以代表一個(gè)耗時(shí)的阻塞函數(shù)調(diào)用，而 `asyncio.sleep()` 則用于替代一個(gè)非阻塞調(diào)用，該調(diào)用也需要一些時(shí)間來(lái)完成。

正如你將在下一節(jié)中看到的，等待某件事（包括 `asyncio.sleep()`）的好處是，周圍的函數(shù)可以暫時(shí)將控制權(quán)讓給另一個(gè)能夠立即做些事情的函數(shù)。相比之下，`time.sleep()` 或任何其他阻塞調(diào)用都與異步 Python 代碼不兼容，因?yàn)樗鼤?huì)在睡眠期間完全停止一切。

關(guān)鍵字 async 和 await

此時(shí)，有必要對(duì) `async`、`await` 以及它們幫助你創(chuàng)建的協(xié)程函數(shù)進(jìn)行更正式的定義：

- `async def` 語(yǔ)法結(jié)構(gòu)引入了一個(gè)協(xié)程函數(shù)或異步生成器。
- `async with` 和 `async for` 語(yǔ)法結(jié)構(gòu)分別引入了異步的 `with` 語(yǔ)句和 `for` 循環(huán)。
- `await` 關(guān)鍵字會(huì)暫停所在協(xié)程的執(zhí)行，并將控制權(quán)交還給事件循環(huán)。

為了進(jìn)一步澄清最后一點(diǎn)，當(dāng) Python 在 `g()` 協(xié)程的作用域中遇到 `await f()` 表達(dá)式時(shí)，`await` 會(huì)告訴事件循環(huán)：暫停 `g()` 的執(zhí)行，直到返回 `f()` 的結(jié)果。與此同時(shí)，讓其他內(nèi)容運(yùn)行。

在代碼中，最后一點(diǎn)大致如下所示：

async def g():
    result = await f()  # 暫停，當(dāng)f()返回結(jié)果時(shí)候回到g() 
    return result

圍繞何時(shí)以及如何使用 async 和 await，還有一套嚴(yán)格的規(guī)則。無(wú)論你還在學(xué)習(xí)語(yǔ)法，還是已經(jīng)熟悉了 async 和 await 的使用，這些規(guī)則都很有幫助：

使用 async def 構(gòu)造，你可以定義一個(gè)協(xié)程函數(shù)。它可以使用 await、return 或 yield，但這三者都是可選的：

• 普通的協(xié)程函數(shù)可以使用 await、return 或兩者都用。要調(diào)用一個(gè)協(xié)程函數(shù)，你必須通過(guò) await 來(lái)獲取它的結(jié)果，或者直接在事件循環(huán)中運(yùn)行它。
• 在 async def 函數(shù)中使用 yield 會(huì)創(chuàng)建一個(gè)異步生成器。要迭代這個(gè)生成器，你可以使用 async for 循環(huán)或推導(dǎo)式。
• async def 不能使用 yield from，否則會(huì)引發(fā) SyntaxError。
• 在 async def 函數(shù)之外使用 await 也會(huì)引發(fā) SyntaxError。你只能在協(xié)程體中使用 await。

以下是一些簡(jiǎn)潔的例子，總結(jié)了這些規(guī)則：

async def f(x):
    y = await z(x)  # Okay - `await` and `return` allowed in coroutines
    return y

async def g(x):
    yield x  # Okay - this is an async generator

async def m(x):
    yield from gen(x)  # No - SyntaxError

def n(x):
    y = await z(x)  # No - SyntaxError (no `async def` here)
    return y

最后，當(dāng)你使用 `await f()` 時(shí)，要求 `f()` 是一個(gè)可等待的對(duì)象，這可以是另一個(gè)協(xié)程，或者是一個(gè)定義了 `.__await__()` 特殊方法的對(duì)象，該方法返回一個(gè)迭代器。在大多數(shù)情況下，你只需要關(guān)注協(xié)程。

以下是一個(gè)更詳細(xì)的例子，展示了異步 I/O 如何減少等待時(shí)間。假設(shè)你有一個(gè)名為 `make_random()` 的協(xié)程函數(shù)，它不斷生成范圍在 [0, 10] 內(nèi)的隨機(jī)整數(shù)，當(dāng)其中一個(gè)超過(guò)閾值時(shí)返回。在下面的例子中，你異步運(yùn)行這個(gè)函數(shù)三次。為了區(qū)分每次調(diào)用，你使用顏色：

import asyncio
import random

COLORS = (
    "\033[0m",  # End of color
    "\033[36m",  # Cyan
    "\033[91m",  # Red
    "\033[35m",  # Magenta
)

async def main():
    return await asyncio.gather(
        makerandom(1, 9),
        makerandom(2, 8),
        makerandom(3, 8),
    )

async def makerandom(delay, threshold=6):
    color = COLORS[delay]
    print(f"{color}Initiated makerandom({delay}).")
    while (number := random.randint(0, 10)) <= threshold:
        print(f"{color}makerandom({delay}) == {number} too low; retrying.")
        await asyncio.sleep(delay)
    print(f"{color}---> Finished: makerandom({delay}) == {number}" + COLORS[0])
    return number

if __name__ == "__main__":
    random.seed(444)
    r1, r2, r3 = asyncio.run(main())
    print()
    print(f"r1: {r1}, r2: {r2}, r3: {r3}")

該程序定義了 `makerandom()` 協(xié)程，并使用三種不同的輸入并發(fā)運(yùn)行它。大多數(shù)程序?qū)⒂尚⌒?、模塊化的協(xié)程組成，并且有一個(gè)包裝函數(shù)用于連接每個(gè)較小的協(xié)程。在 `main()` 中，你將三個(gè)任務(wù)聚集在一起。對(duì) `makerandom()` 的三次調(diào)用構(gòu)成了你的任務(wù)池。

雖然此示例中的隨機(jī)數(shù)生成是一個(gè) CPU 密集型任務(wù)，但其影響可以忽略不計(jì)。`asyncio.sleep()` 模擬了一個(gè) I/O 密集型任務(wù)，并說(shuō)明了只有 I/O 密集型或非阻塞任務(wù)才能從異步 I/O 模型中受益。

The Async I/O Event Loop

在異步編程中，事件循環(huán)就像一個(gè)無(wú)限循環(huán)，它監(jiān)控協(xié)程，獲取關(guān)于哪些協(xié)程處于空閑狀態(tài)的反饋，并尋找可以在此期間執(zhí)行的任務(wù)。當(dāng)協(xié)程等待的資源變得可用時(shí)，它能夠喚醒處于空閑狀態(tài)的協(xié)程。

在現(xiàn)代 Python 中，啟動(dòng)事件循環(huán)的推薦方式是使用 `asyncio.run()`。這個(gè)函數(shù)負(fù)責(zé)獲取事件循環(huán)，運(yùn)行任務(wù)直到它們完成，并關(guān)閉循環(huán)。當(dāng)你在同一代碼中已經(jīng)運(yùn)行了另一個(gè)異步事件循環(huán)時(shí)，不能調(diào)用這個(gè)函數(shù)（asyncio.run()`）。

你也可以使用 `get_running_loop()` 函數(shù)獲取正在運(yùn)行的循環(huán)實(shí)例：

loop = asyncio.get_running_loop()

如果你需要在 Python 程序中與事件循環(huán)交互，上述模式是一個(gè)很好的方法。`loop` 對(duì)象支持通過(guò) `.is_running()` 和 `.is_closed()` 進(jìn)行自省。當(dāng)你想通過(guò)傳遞循環(huán)作為參數(shù)來(lái)安排回調(diào)時(shí)，這可能會(huì)很有用。注意，如果當(dāng)前沒(méi)有正在運(yùn)行的事件循環(huán)，`get_running_loop()` 會(huì)引發(fā)一個(gè) `RuntimeError` 異常。

更重要的是理解事件循環(huán)表面之下的工作原理。以下幾點(diǎn)值得強(qiáng)調(diào)：

- 協(xié)程本身幾乎不做任何事情，直到它們與事件循環(huán)關(guān)聯(lián)起來(lái)。
- 默認(rèn)情況下，異步事件循環(huán)在單個(gè)線程和單個(gè) CPU 核心上運(yùn)行。在大多數(shù) `asyncio` 應(yīng)用程序中，通常只有一個(gè)事件循環(huán)，通常在主線程中。在不同的線程中運(yùn)行多個(gè)事件循環(huán)在技術(shù)上是可行的，但并不常見(jiàn)，也不推薦。
- 事件循環(huán)是可插拔的。你可以編寫自己的實(shí)現(xiàn)，并讓它像 `asyncio` 中提供的事件循環(huán)一樣運(yùn)行任務(wù)。

關(guān)于第一點(diǎn)，如果你有一個(gè)等待其他協(xié)程的協(xié)程，那么單獨(dú)調(diào)用它幾乎沒(méi)有什么效果：

>>> import asyncio

>>> async def main():
...     print("Hello...")
...     await asyncio.sleep(1)
...     print("World!")
...

>>> routine = main()
>>> routine
<coroutine object main at 0x1027a6150>

在這個(gè)例子中，直接調(diào)用 `main()` 會(huì)返回一個(gè)協(xié)程對(duì)象，你不能單獨(dú)使用它。你需要使用 `asyncio.run()` 將 `main()` 協(xié)程安排到事件循環(huán)中執(zhí)行：

>>> asyncio.run(routine)
Hello...
World!

通常，你會(huì)將 `main()` 協(xié)程包裹在 `asyncio.run()` 調(diào)用中。你可以使用 `await` 來(lái)執(zhí)行低級(jí)別的協(xié)程。

最后，事件循環(huán)是可插拔的，這意味著你可以使用任何有效的事件循環(huán)實(shí)現(xiàn)，這與你的協(xié)程結(jié)構(gòu)無(wú)關(guān)。`asyncio` 包附帶了兩種不同的事件循環(huán)實(shí)現(xiàn)。

默認(rèn)的事件循環(huán)實(shí)現(xiàn)取決于你的平臺(tái)和 Python 版本。例如，在 Unix 上，默認(rèn)通常是 `SelectorEventLoop`，而 Windows 使用 `ProactorEventLoop` 以獲得更好的子進(jìn)程和 I/O 支持。

還有第三方事件循環(huán)可供選擇。例如，`uvloop` 包提供了一種替代實(shí)現(xiàn)，承諾比 `asyncio` 的循環(huán)更快。

通常，你會(huì)將 `main()` 協(xié)程包裹在 `asyncio.run()` 調(diào)用中。你可以使用 `await` 來(lái)執(zhí)行低級(jí)別的協(xié)程。

最后，事件循環(huán)是可插拔的，這意味著你可以使用任何有效的事件循環(huán)實(shí)現(xiàn)，這與你的協(xié)程結(jié)構(gòu)無(wú)關(guān)。`asyncio` 包附帶了兩種不同的事件循環(huán)實(shí)現(xiàn)。

默認(rèn)的事件循環(huán)實(shí)現(xiàn)取決于你的平臺(tái)和 Python 版本。例如，在 Unix 上，默認(rèn)通常是 `SelectorEventLoop`，而 Windows 使用 `ProactorEventLoop` 以獲得更好的子進(jìn)程和 I/O 支持。

還有第三方事件循環(huán)可供選擇。例如，`uvloop` 包提供了一種替代實(shí)現(xiàn)，承諾比 `asyncio` 的循環(huán)更快。

異步輸入輸出交互式解釋器

從 Python 3.8 開(kāi)始，`asyncio` 模塊包含了一個(gè)專門的交互式 shell，稱為 `asyncio` REPL。這個(gè)環(huán)境允許你在頂層直接使用 `await`，而無(wú)需將代碼包裹在對(duì) `asyncio.run()` 的調(diào)用中。這個(gè)工具便于在 Python 中實(shí)驗(yàn)、調(diào)試和學(xué)習(xí) `asyncio`。

要啟動(dòng) REPL，你可以運(yùn)行以下命令：

$ python -m asyncio
asyncio REPL 3.13.3 (main, Jun 25 2025, 17:27:59) ... on darwin
Use "await" directly instead of "asyncio.run()".
Type "help", "copyright", "credits" or "license" for more information.
>>> import asyncio
>>>

一旦出現(xiàn) `>>>` 提示符，你就可以在那里開(kāi)始運(yùn)行異步代碼了?？紤]下面的例子，其中你重用了上一節(jié)的代碼：

>>> import asyncio

>>> async def main():
...     print("Hello...")
...     await asyncio.sleep(1)
...     print("World!")
...

>>> await main()
Hello...
World!

這個(gè)例子和前面的例子效果一樣，但是不需要使用asyncio.run()來(lái)運(yùn)行main()，你可以直接使用await。

到此這篇關(guān)于python中asyncio處理異步IO的實(shí)現(xiàn)示例的文章就介紹到這了,更多相關(guān)asyncio處理異步IO內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論