python中asyncio处理异步IO的实现示例

2025-11-12 08:29:50 作者：自卷小烧饼

本文主要介绍python中asyncio处理异步IO的实现示例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧

既然你已经对异步 I/O 这种并发模型有了一定的了解，现在是时候探索 Python 的实现方式了。Python 的 `asyncio` 包以及其相关的两个关键字 `async` 和 `await`，各自有不同的用途，但它们共同帮助你声明、构建、执行和管理异步代码。

协程和协程函数异步 I/O 的核心是协程的概念，协程是一种可以暂停其执行并在稍后恢复的对象。在此期间，它可以将控制权交给事件循环，事件循环可以执行另一个协程。协程对象是通过调用协程函数（也称为异步函数）生成的。你可以使用 async def 构造来定义一个协程函数。

在编写你的第一段异步代码之前，请考虑以下同步运行的示例：

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()` 函数打印 “One”，然后等待一秒，接着打印 “Two”，再等待一秒。`main()` 函数中的循环执行了三次 `count()`。在 `if __name__ == "__main__"` 条件下，你在执行开始时记录当前时间，调用 `main()`，计算总时间，并将其显示在屏幕上。

当你运行这个脚本时，你会得到以下输出：

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

该脚本交替打印“One”和“Two”，每次打印操作之间间隔一秒。总共运行时间略多于六秒。

如果你将此脚本更新为使用 Python 的异步 I/O 模型，那么它看起来会像下面这样：

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.")

现在，你使用 `async` 关键字将 `count()` 转换为一个协程函数，它打印 “One”，等待一秒，然后打印 “Two”，再等待一秒。你使用 `await` 关键字等待 `asyncio.sleep()` 的执行。这将控制权交还给程序的事件循环，表示：“我要睡一秒钟。你趁机运行其他任务吧。”

`main()` 函数是另一个协程函数，它使用 `asyncio.gather()` 并发运行三个 `count()` 实例。你使用 `asyncio.run()` 函数启动事件循环并执行 `main()`。

将这个版本的性能与同步版本进行比较：

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

得益于异步 I/O 方法，总执行时间仅为两秒多一点，而不是六秒，这展示了 `asyncio` 在处理 I/O 密集型任务时的高效性。

虽然使用 `time.sleep()` 和 `asyncio.sleep()` 可能看起来很平常，但它们是耗时过程的替代品，这些过程涉及等待时间。调用 `time.sleep()` 可以代表一个耗时的阻塞函数调用，而 `asyncio.sleep()` 则用于替代一个非阻塞调用，该调用也需要一些时间来完成。

正如你将在下一节中看到的，等待某件事（包括 `asyncio.sleep()`）的好处是，周围的函数可以暂时将控制权让给另一个能够立即做些事情的函数。相比之下，`time.sleep()` 或任何其他阻塞调用都与异步 Python 代码不兼容，因为它会在睡眠期间完全停止一切。

关键字 async 和 await

此时，有必要对 `async`、`await` 以及它们帮助你创建的协程函数进行更正式的定义：

- `async def` 语法结构引入了一个协程函数或异步生成器。
- `async with` 和 `async for` 语法结构分别引入了异步的 `with` 语句和 `for` 循环。
- `await` 关键字会暂停所在协程的执行，并将控制权交还给事件循环。

为了进一步澄清最后一点，当 Python 在 `g()` 协程的作用域中遇到 `await f()` 表达式时，`await` 会告诉事件循环：暂停 `g()` 的执行，直到返回 `f()` 的结果。与此同时，让其他内容运行。

在代码中，最后一点大致如下所示：

async def g():
    result = await f()  # 暂停，当f()返回结果时候回到g() 
    return result

围绕何时以及如何使用 async 和 await，还有一套严格的规则。无论你还在学习语法，还是已经熟悉了 async 和 await 的使用，这些规则都很有帮助：

使用 async def 构造，你可以定义一个协程函数。它可以使用 await、return 或 yield，但这三者都是可选的：

普通的协程函数可以使用 await、return 或两者都用。要调用一个协程函数，你必须通过 await 来获取它的结果，或者直接在事件循环中运行它。
在 async def 函数中使用 yield 会创建一个异步生成器。要迭代这个生成器，你可以使用 async for 循环或推导式。
async def 不能使用 yield from，否则会引发 SyntaxError。
在 async def 函数之外使用 await 也会引发 SyntaxError。你只能在协程体中使用 await。

以下是一些简洁的例子，总结了这些规则：

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

最后，当你使用 `await f()` 时，要求 `f()` 是一个可等待的对象，这可以是另一个协程，或者是一个定义了 `.__await__()` 特殊方法的对象，该方法返回一个迭代器。在大多数情况下，你只需要关注协程。

以下是一个更详细的例子，展示了异步 I/O 如何减少等待时间。假设你有一个名为 `make_random()` 的协程函数，它不断生成范围在 [0, 10] 内的随机整数，当其中一个超过阈值时返回。在下面的例子中，你异步运行这个函数三次。为了区分每次调用，你使用颜色：

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()` 协程，并使用三种不同的输入并发运行它。大多数程序将由小型、模块化的协程组成，并且有一个包装函数用于连接每个较小的协程。在 `main()` 中，你将三个任务聚集在一起。对 `makerandom()` 的三次调用构成了你的任务池。

虽然此示例中的随机数生成是一个 CPU 密集型任务，但其影响可以忽略不计。`asyncio.sleep()` 模拟了一个 I/O 密集型任务，并说明了只有 I/O 密集型或非阻塞任务才能从异步 I/O 模型中受益。

The Async I/O Event Loop

在异步编程中，事件循环就像一个无限循环，它监控协程，获取关于哪些协程处于空闲状态的反馈，并寻找可以在此期间执行的任务。当协程等待的资源变得可用时，它能够唤醒处于空闲状态的协程。

在现代 Python 中，启动事件循环的推荐方式是使用 `asyncio.run()`。这个函数负责获取事件循环，运行任务直到它们完成，并关闭循环。当你在同一代码中已经运行了另一个异步事件循环时，不能调用这个函数（asyncio.run()`）。

你也可以使用 `get_running_loop()` 函数获取正在运行的循环实例：

loop = asyncio.get_running_loop()

如果你需要在 Python 程序中与事件循环交互，上述模式是一个很好的方法。`loop` 对象支持通过 `.is_running()` 和 `.is_closed()` 进行自省。当你想通过传递循环作为参数来安排回调时，这可能会很有用。注意，如果当前没有正在运行的事件循环，`get_running_loop()` 会引发一个 `RuntimeError` 异常。

更重要的是理解事件循环表面之下的工作原理。以下几点值得强调：

- 协程本身几乎不做任何事情，直到它们与事件循环关联起来。
- 默认情况下，异步事件循环在单个线程和单个 CPU 核心上运行。在大多数 `asyncio` 应用程序中，通常只有一个事件循环，通常在主线程中。在不同的线程中运行多个事件循环在技术上是可行的，但并不常见，也不推荐。
- 事件循环是可插拔的。你可以编写自己的实现，并让它像 `asyncio` 中提供的事件循环一样运行任务。

关于第一点，如果你有一个等待其他协程的协程，那么单独调用它几乎没有什么效果：

>>> import asyncio

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

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

在这个例子中，直接调用 `main()` 会返回一个协程对象，你不能单独使用它。你需要使用 `asyncio.run()` 将 `main()` 协程安排到事件循环中执行：

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

通常，你会将 `main()` 协程包裹在 `asyncio.run()` 调用中。你可以使用 `await` 来执行低级别的协程。

最后，事件循环是可插拔的，这意味着你可以使用任何有效的事件循环实现，这与你的协程结构无关。`asyncio` 包附带了两种不同的事件循环实现。

默认的事件循环实现取决于你的平台和 Python 版本。例如，在 Unix 上，默认通常是 `SelectorEventLoop`，而 Windows 使用 `ProactorEventLoop` 以获得更好的子进程和 I/O 支持。

还有第三方事件循环可供选择。例如，`uvloop` 包提供了一种替代实现，承诺比 `asyncio` 的循环更快。

通常，你会将 `main()` 协程包裹在 `asyncio.run()` 调用中。你可以使用 `await` 来执行低级别的协程。

最后，事件循环是可插拔的，这意味着你可以使用任何有效的事件循环实现，这与你的协程结构无关。`asyncio` 包附带了两种不同的事件循环实现。

还有第三方事件循环可供选择。例如，`uvloop` 包提供了一种替代实现，承诺比 `asyncio` 的循环更快。

异步输入输出交互式解释器

从 Python 3.8 开始，`asyncio` 模块包含了一个专门的交互式 shell，称为 `asyncio` REPL。这个环境允许你在顶层直接使用 `await`，而无需将代码包裹在对 `asyncio.run()` 的调用中。这个工具便于在 Python 中实验、调试和学习 `asyncio`。

要启动 REPL，你可以运行以下命令：

$ 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
>>>

一旦出现 `>>>` 提示符，你就可以在那里开始运行异步代码了。考虑下面的例子，其中你重用了上一节的代码：

>>> import asyncio

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

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

这个例子和前面的例子效果一样，但是不需要使用asyncio.run()来运行main()，你可以直接使用await。

到此这篇关于python中asyncio处理异步IO的实现示例的文章就介绍到这了,更多相关asyncio处理异步IO内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

python中asyncio处理异步IO的实现示例

关键字 async 和 await

The Async I/O Event Loop

您可能感兴趣的文章: