Python asyncio 模块

编辑：陈珺

58 次浏览	3次

asyncio 是 Python 标准库中的一个模块，用于编写异步 I/O 操作的代码。

asyncio 提供了一种高效的方式来处理并发任务，特别适用于 I/O 密集型操作，如网络请求、文件读写等。

通过使用 asyncio，你可以在单线程中同时处理多个任务，而无需使用多线程或多进程。

为什么需要 asyncio？

在传统的同步编程中，当一个任务需要等待 I/O 操作（如网络请求）完成时，程序会阻塞，直到操作完成。这会导致程序的效率低下，尤其是在需要处理大量 I/O 操作时。

asyncio 通过引入异步编程模型，允许程序在等待 I/O 操作时继续执行其他任务，从而提高了程序的并发性和效率。

想象一下你正在经营一家餐厅：

同步模式（普通函数）： 你只有一个厨师。客人 A 点了一份牛排，厨师开始煎牛排（这需要等待 5 分钟）。在煎牛排的这 5 分钟里，厨师完全被占用，不能做任何其他事，即使客人 B 只想点一杯水，也必须干等着。

异步模式（asyncio）： 你有多个厨师（实际上还是一个，但非常聪明）。厨师开始煎客人 A 的牛排后，发现需要等待，他立刻把这份牛排标记为等待中，然后转头去给客人 B 倒水。倒完水回来，看看牛排是不是快好了，如果还没好，又可以去处理客人 C 的订单。这样，在等待 I/O（如煎牛排、网络请求、读写文件）的时间里，厨师（CPU）一直在高效地工作。

asyncio 就是 Python 用来实现这种聪明工作模式的标准库，它允许你编写单线程并发的代码，特别适用于网络爬虫、Web 服务器、微服务等 I/O 密集型场景。

它的核心是事件循环、协程和任务。

asyncio 的核心概念

1. 协程（Coroutine）

协程是 asyncio 的核心概念之一。它是一个特殊的函数，可以在执行过程中暂停，并在稍后恢复执行。协程通过 async def 关键字定义，并通过 await 关键字暂停执行，等待异步操作完成。

import asyncio

async def say_hello():
print("Hello")
await asyncio.sleep(1)
print("World")

2. 事件循环（Event Loop）

事件循环是 asyncio 的核心组件，负责调度和执行协程。它不断地检查是否有任务需要执行，并在任务完成后调用相应的回调函数。

async def main():
await say_hello()

asyncio.run(main())

3. 任务（Task）

任务是对协程的封装，表示一个正在执行或将要执行的协程。你可以通过 asyncio.create_task() 函数创建任务，并将其添加到事件循环中。

async def main():
task = asyncio.create_task(say_hello())
await task

4. Future

Future 是一个表示异步操作结果的对象。它通常用于底层 API，表示一个尚未完成的操作。你可以通过 await 关键字等待 Future 完成。

async def main():
future = asyncio.Future()
await future

基础用法与代码示例

让我们通过一个经典的并发访问多个网址的例子来理解上述概念。

假设我们需要获取三个不同网址的内容。使用同步方式会顺序执行，总耗时是三次请求耗时的总和。使用 asyncio，我们可以让这三个请求同时发出，总耗时接近于最慢的那一次请求。

同步版本（作为对比）

import time
import requests

def fetch_url(url):
"""模拟一个耗时的网络请求（同步版本）"""
print(f"开始获取: {url}")
time.sleep(2) # 模拟 2 秒网络延迟
print(f"完成获取: {url}")
return f"来自 {url} 的数据"

def main_sync():
urls = ['https://example.com/1', 'https://example.com/2', 'https://example.com/3']
results = []
start = time.time()

for url in urls:
result = fetch_url(url) # 必须等上一个完成才能开始下一个
results.append(result)

end = time.time()
print(f"同步版本总耗时: {end - start:.2f} 秒")
print(f"结果: {results}")

if __name__ == "__main__":
main_sync()

预期运行结果：

开始获取: https://example.com/1
完成获取: https://example.com/1
开始获取: https://example.com/2
完成获取: https://example.com/2
开始获取: https://example.com/3
完成获取: https://example.com/3
同步版本总耗时: 6.00 秒
结果: [‘来自 https://example.com/1 的数据‘, ‘来自 https://example.com/2 的数据‘, ‘来自 https://example.com/2 的数据‘]

总共花了约 6 秒。

异步版本（使用 asyncio）

我们需要用 aiohttp 库来替代 requests 进行异步 HTTP 请求。首先安装它：pip install aiohttp。

import asyncio
import aiohttp
import time

async def fetch_url_async(session, url):
"""模拟一个耗时的网络请求（异步版本）"""
print(f"开始异步获取: {url}")
# 注意：这里我们使用 aiohttp 的异步 get 方法，并用 await 等待
async with session.get(url) as response:
# 模拟处理响应也需要时间
await asyncio.sleep(2) # 使用 asyncio.sleep 模拟 I/O 等待，它不会阻塞线程
text = await response.text()
print(f"完成异步获取: {url}")
return f"来自 {url} 的数据 (长度: {len(text)})"

async def main_async():
urls = ['https://httpbin.org/get', 'https://httpbin.org/delay/1', 'https://httpbin.org/headers']

async with aiohttp.ClientSession() as session: # 创建异步 HTTP 会话
# 为每个 URL 创建一个任务（Task）
tasks = []
for url in urls:
# create_task 会将协程加入事件循环，立即开始调度
task = asyncio.create_task(fetch_url_async(session, url))
tasks.append(task)

print("所有任务已创建，开始并发执行...")

# 使用 asyncio.gather 并发运行所有任务，并等待它们全部完成
# gather 返回一个结果列表，顺序与传入的任务顺序一致
results = await asyncio.gather(*tasks)

return results

if __name__ == "__main__":
start = time.time()
# asyncio.run() 是启动事件循环并运行顶层协程的简便方法
final_results = asyncio.run(main_async())
end = time.time()

print(f"\n异步版本总耗时: {end - start:.2f} 秒")
for res in final_results:
print(res)

预期运行结果：

所有任务已创建，开始并发执行...
开始异步获取: https://httpbin.org/get
开始异步获取: https://httpbin.org/delay/1
开始异步获取: https://httpbin.org/headers
（大约 2 秒后，所有请求几乎同时完成）
完成异步获取: https://httpbin.org/headers
完成异步获取: https://httpbin.org/get
完成异步获取: https://httpbin.org/delay/1

异步版本总耗时: 2.10 秒  # 注意！总耗时远小于 6 秒
来自 https://httpbin.org/get 的数据 (长度: 274)
来自 https://httpbin.org/delay/1 的数据 (长度: 392)
来自 https://httpbin.org/headers 的数据 (长度: 177)

代码解析：

async def：定义了协程函数 fetch_url_async 和 main_async。
await：在 fetch_url_async 中，我们 await session.get() 和 await response.text()，这告诉事件循环："这个网络请求需要时间，你先去执行其他就绪的任务吧"。
asyncio.create_task()：将 fetch_url_async 协程包装成 Task，使其被事件循环调度，实现并发。
asyncio.gather(*tasks)：一个非常实用的函数，它并发运行所有传入的协程/任务，并等待它们全部完成，最后收集所有结果。
asyncio.run(main_async())：Python 3.7+ 推荐的方式，它负责创建事件循环、运行协程并关闭循环。

关键函数与参数说明

下面以表格形式列出 asyncio 中几个最常用的高级函数：

函数	主要作用	常用参数说明
*`asyncio.run(coro, , debug=False)`**	运行一个顶层协程，管理事件循环的生命周期。是程序的主入口。	`coro`: 要运行的协程对象。 `debug`: 设为 `True` 可启用事件循环的调试模式。
*`asyncio.create_task(coro, , name=None)`**	将协程包装成一个 Task 对象，并排入事件循环等待调度。这是实现并发的主要方式。	`coro`: 要包装的协程对象。 `name`: （Python 3.8+）为任务指定一个名称，便于调试。
*`asyncio.gather(aws, return_exceptions=False)`**	并发运行多个异步任务（`aws` 可接受协程、任务等），并等待所有完成，返回结果列表。	`*aws`: 可变参数，传入多个异步对象。 `return_exceptions`: 默认为 `False`，任何任务抛出异常都会立即传播给 `gather` 的调用者。设为 `True` 时，异常会作为结果的一部分返回。
`asyncio.sleep(delay, result=None)`	异步地休眠指定秒数。这是与 `time.sleep`（阻塞）的关键区别。	`delay`: 休眠的秒数。 `result`: 休眠结束后返回的值。
*`asyncio.wait(aws, , timeout=None, return_when=ALL_COMPLETED)`**	并发运行任务，并等待满足指定条件。返回两个集合 `(done, pending)`，分别是已完成和未完成的任务。	`aws`: 异步对象集合。 `timeout`: 超时时间（秒）。 `return_when`: 决定何时返回，可选：`FIRST_COMPLETED`（第一个完成）、`FIRST_EXCEPTION`（第一个异常）、`ALL_COMPLETED`（全部完成，默认）。
*`asyncio.to_thread(func, /, args, kwargs)`	（Python 3.9+）将一个普通的、可能阻塞的同步函数放到一个单独的线程中运行，并返回一个可 `await` 的协程。用于处理 CPU 密集型或阻塞式 I/O。	`func`: 要在线程中运行的同步函数。 `args, *kwargs`: 传递给函数的参数。

可视化理解：异步任务调度流程

图解说明：这个流程图展示了事件循环如何像调度员一样工作。它维护一个任务队列，当一个任务执行到 await（例如等待网络响应）时，它会被挂起，事件循环立即从队列中找出下一个可以运行（就绪）的任务来执行。当被挂起任务的 I/O 操作完成后，事件循环会收到通知，将该任务状态改回就绪，并在未来某个时刻继续执行它。通过这种方式，在 I/O 等待期间，CPU 被充分利用来执行其他任务，实现了单线程内的并发。

asyncio 的基本用法

1. 运行协程

要运行一个协程，你可以使用 asyncio.run() 函数。它会创建一个事件循环，并运行指定的协程。

import asyncio

async def main():
print("Start")
await asyncio.sleep(1)
print("End")

asyncio.run(main())

2. 并发执行多个任务

你可以使用 asyncio.gather() 函数并发执行多个协程，并等待它们全部完成。

import asyncio

async def task1():
print("Task 1 started")
await asyncio.sleep(1)
print("Task 1 finished")

async def task2():
print("Task 2 started")
await asyncio.sleep(2)
print("Task 2 finished")

async def main():
await asyncio.gather(task1(), task2())

asyncio.run(main())

3. 超时控制

你可以使用 asyncio.wait_for() 函数为协程设置超时时间。如果协程在指定时间内未完成，将引发 asyncio.TimeoutError 异常。

import asyncio

async def long_task():
await asyncio.sleep(10)
print("Task finished")

async def main():
try:
await asyncio.wait_for(long_task(), timeout=5)
except asyncio.TimeoutError:
print("Task timed out")

asyncio.run(main())

asyncio 的应用场景

asyncio 特别适用于以下场景：

网络请求：如 HTTP 请求、WebSocket 通信等。
文件 I/O：如异步读写文件。
数据库操作：如异步访问数据库。
实时数据处理：如实时消息队列处理。

常用类、方法和函数

1. 核心函数

方法/函数	说明	示例
`asyncio.run(coro)`	运行异步主函数（Python 3.7+）	`asyncio.run(main())`
`asyncio.create_task(coro)`	创建任务并加入事件循环	`task = asyncio.create_task(fetch_data())`
*`asyncio.gather(coros)`**	并发运行多个协程	`await asyncio.gather(task1, task2)`
`asyncio.sleep(delay)`	异步等待（非阻塞）	`await asyncio.sleep(1)`
`asyncio.wait(coros)`	控制任务完成方式	`done, pending = await asyncio.wait([task1, task2])`

2. 事件循环（Event Loop）

方法	说明	示例
`loop.run_until_complete(future)`	运行直到任务完成	`loop.run_until_complete(main())`
`loop.run_forever()`	永久运行事件循环	`loop.run_forever()`
`loop.stop()`	停止事件循环	`loop.stop()`
`loop.close()`	关闭事件循环	`loop.close()`
`loop.call_soon(callback)`	安排回调函数立即执行	`loop.call_soon(print, "Hello")`
`loop.call_later(delay, callback)`	延迟执行回调	`loop.call_later(5, callback)`

3. 协程（Coroutine）与任务（Task）

方法/装饰器	说明	示例
`@asyncio.coroutine`	协程装饰器（旧版，Python 3.4-3.7）	`@asyncio.coroutine` `def old_coro():`
`async def`	定义协程（Python 3.5+）	`async def fetch():`
`task.cancel()`	取消任务	`task.cancel()`
`task.done()`	检查任务是否完成	`if task.done():`
`task.result()`	获取任务结果（需任务完成）	`data = task.result()`

4. 同步原语（类似threading）

类	说明	示例
`asyncio.Lock()`	异步互斥锁	`lock = asyncio.Lock()` `async with lock:`
`asyncio.Event()`	事件通知	`event = asyncio.Event()` `await event.wait()`
`asyncio.Queue()`	异步队列	`queue = asyncio.Queue()` `await queue.put(item)`
`asyncio.Semaphore()`	信号量	`sem = asyncio.Semaphore(5)` `async with sem:`

5. 网络与子进程

方法/类	说明	示例
`asyncio.open_connection()`	建立TCP连接	`reader, writer = await asyncio.open_connection('host', 80)`
`asyncio.start_server()`	创建TCP服务器	`server = await asyncio.start_server(handle, '0.0.0.0', 8888)`
`asyncio.create_subprocess_exec()`	创建子进程	`proc = await asyncio.create_subprocess_exec('ls')`

6. 实用工具

方法	说明	示例
`asyncio.current_task()`	获取当前任务	`task = asyncio.current_task()`
`asyncio.all_tasks()`	获取所有任务	`tasks = asyncio.all_tasks()`
`asyncio.shield(coro)`	保护任务不被取消	`await asyncio.shield(critical_task)`
`asyncio.wait_for(coro, timeout)`	带超时的等待	`try: await asyncio.wait_for(task, 5)`

实例

1. 基本协程示例

import asyncio

async def hello():
print("Hello")
await asyncio.sleep(1)
print("World")

asyncio.run(hello()) # Python 3.7+

2. 并发执行任务

async def fetch(url):
print(f"Fetching {url}")
await asyncio.sleep(2)
return f"Data from {url}"

async def main():
results = await asyncio.gather(
fetch("url1.com"),
fetch("url2.com")
)
print(results)

asyncio.run(main())

3. 使用异步队列

async def producer(queue):
for i in range(5):
await queue.put(i)
await asyncio.sleep(0.1)

async def consumer(queue):
while True:
item = await queue.get()
print(f"Consumed {item}")
queue.task_done()

async def main():
queue = asyncio.Queue()
await asyncio.gather(
producer(queue),
consumer(queue) )

注意事项

Python版本：部分功能需Python 3.7+（如asyncio.run()）。
阻塞操作：避免在协程中使用同步阻塞代码（如time.sleep()）。
调试：设置PYTHONASYNCIODEBUG=1环境变量可启用调试模式。
取消任务：被取消的任务会引发CancelledError，需妥善处理。

58 次浏览	3次

< 首页