asyncio模块

这篇笔记整理了 Python asyncio 中事件循环、async 与 await、asyncio.create_task、asyncio.gather、AsyncIterator 以及 asyncio.Queue 的用法，最后用一个生产者消费者流水线示例把它们串起来。

一、事件循环与非阻塞 I/O

在同步代码中，等待网络请求或磁盘读写时 CPU 只能处于空闲等待状态。

需要注意，asyncio 的非阻塞主要针对网络 I/O。它没有原生的非阻塞文件 I/O，普通的文件读写仍会阻塞事件循环，需要借助 asyncio.to_thread 或 aiofiles 等第三方库处理。

asyncio 依托单线程与事件循环（Event Loop）机制工作：

• 事件循环负责在后台轮询和调度任务。
• 协作式多任务：当任务遇到耗时 I/O 并通过 await 挂起时，会主动让出控制权，调度中心立即安排其他就绪的任务执行。
• 适用场景：网络爬虫、Web API 服务、微服务通信等 I/O 密集型任务。若遇到纯计算密集型任务，应当使用 multiprocessing 多进程模块。

二、async 与 await

1. async def 定义协程函数

使用 async def 定义的函数是协程函数。直接调用它不会执行函数体，而是返回一个协程对象（Coroutine Object）。启动异步程序的标准入口是 asyncio.run()。

import asyncio

async def fetch_user():
    print("开始拉取数据...")
    # asyncio.sleep 模拟非阻塞 I/O，此处不能使用阻塞主线程的 time.sleep
    await asyncio.sleep(1) 
    print("数据拉取完毕！")
    return {"id": 101, "name": "Alice"}

result = asyncio.run(fetch_user())
print("返回结果:", result)

2. await 的语义与限制

await 的语义是挂起当前协程，让出 CPU 给其他协程运行，待等待的操作完成后再恢复执行。

await 只能写在 async def 函数内部，其后只能接三类可等待对象（Awaitables）：

• 协程对象（调用协程函数的返回值）
• Task（通过 asyncio.create_task 包装的对象）
• Future（表示异步操作最终状态的底层对象）

import asyncio

async def step_a():
    await asyncio.sleep(1)
    return "A 完成"

async def step_b():
    await asyncio.sleep(1)
    return "B 完成"

async def main():
    # 串行执行：总耗时约为 2 秒
    res_a = await step_a()  # 必须等 step_a 完成才继续往下走
    print(res_a)
    res_b = await step_b()
    print(res_b)

asyncio.run(main())

注意：在循环中直接依次 await 协程依然是串行运行的。实现并发需要借助 create_task 或 gather。

三、并发任务管理：asyncio.create_task

asyncio.create_task(coroutine) 用于将协程封装为 Task 对象，并排入事件循环调度。

调用 create_task 后，任务就已排入事件循环，在当前协程让出控制权时开始并发运行，并不需要立刻 await 它。后续对 Task 执行 await，作用是获取其返回值或确认其执行完毕。

import asyncio
import time

async def download(file_name: str, delay: int):
    print(f"[{file_name}] 开始下载...")
    await asyncio.sleep(delay)
    print(f"[{file_name}] 下载完成 (耗时 {delay}s)")
    return f"{file_name}_data"

async def main():
    start = time.perf_counter()

    # 提交任务到事件循环，后台开始并发
    t1 = asyncio.create_task(download("File1", 2))
    t2 = asyncio.create_task(download("File2", 3))

    print("主流程继续执行其他逻辑...")
    await asyncio.sleep(0.5)

    # 等待具体任务的结果
    res1 = await t1
    res2 = await t2

    print(f"全部完成: {res1}, {res2}")
    print(f"总耗时: {time.perf_counter() - start:.2f} 秒")  # 约为 3 秒，而不是 5 秒

asyncio.run(main())

Task 对象的常用操作：

• 取消任务：task.cancel()。被取消的任务在下次 await 时会抛出 asyncio.CancelledError。
• 状态检查：task.done()、task.cancelled()。
• 回调注册：task.add_done_callback(callback_func)。

四、批量结果收集：asyncio.gather

asyncio.gather(*aws, return_exceptions=False) 用于批量聚合多个并发任务。它会自动把传入的协程包装为 Task，并发起调度。返回值严格按照参数传入的先后顺序排列，不会因为某个任务先执行完而打乱顺序。

import asyncio

async def fetch_api(endpoint: str):
    await asyncio.sleep(1)
    if endpoint == "/error":
        raise ConnectionError("网络波动")
    return f"Data from {endpoint}"

async def main():
    endpoints = ["/users", "/error", "/orders"]

    # 动态传参解包 *，设置 return_exceptions=True 开启容错
    results = await asyncio.gather(
        *(fetch_api(ep) for ep in endpoints),
        return_exceptions=True
    )

    for ep, res in zip(endpoints, results):
        if isinstance(res, Exception):
            print(f"接口 {ep} 异常: {res}")
        else:
            print(f"接口 {ep} 成功: {res}")

asyncio.run(main())

参数 return_exceptions 的取值影响：

• False（默认）：只要有一个任务抛出异常，gather 会立刻将该异常向外抛出。
• True：异常会被捕获并作为对应位置的结果放入列表，其他任务继续执行不受阻断。

五、create_task 与 gather 的差异

维度	asyncio.create_task	asyncio.gather
主要定位	单个任务的生命周期控制	多个任务的批量执行与结果汇总
返回值	单个 asyncio.Task 对象	包含所有结果的 list
调度时机	调用时即排入事件循环	调用时即排入事件循环，await 时等待全部完成
控制粒度	支持单独 cancel()、查状态或加回调	作为整体统一等待或取消
典型场景	后台 Worker 协程、流式任务派发	批量请求多个第三方接口并汇总数据

六、流式数据处理：async for

当数据按批次流式获取（如分页拉取、大模型逐字推流、日志采集）时，一次性全部载入内存容易造成内存占用过大，应当使用异步迭代器（AsyncIterator）。

异步生成器（async def + yield）

在协程函数中使用 yield，Python 会自动将其封装为异步生成器，省去手动编写迭代器类的工作。

import asyncio

async def llm_token_stream():
    tokens = ["Deep", "Seek", " / ", "OpenAI", " 正在", "生成", "回答..."]
    for t in tokens:
        await asyncio.sleep(0.2)
        yield t

async def main():
    async for token in llm_token_stream():
        print(token, end="", flush=True)
    print()

asyncio.run(main())

七、缓冲区与解耦：asyncio.Queue

在生产者消费者的架构中，数据生成和处理的速度往往不一致。asyncio.Queue 是专为单线程内协程设计的先进先出（FIFO）队列。因为运行在单线程事件循环中，它不需要加线程锁，但也不能跨进程传递数据（跨进程需使用 multiprocessing.Queue）。

常用 API：

• queue.put(item)：向队列投入元素。若指定了 maxsize 且队列已满，会挂起等待空位。
• queue.get()：从队列提取元素。若队列为空，会挂起等待数据。
• queue.task_done()：消费者处理完一个元素后调用，通知队列该任务处理完毕。
• queue.join()：挂起并等待，直到队列内所有元素都被 get() 且均调用了 task_done()。
• queue.qsize()、empty()、full()：非阻塞方式检查队列状态。

import asyncio
import random

async def producer(queue: asyncio.Queue, count: int):
    for i in range(1, count + 1):
        item = f"Job-{i}"
        await asyncio.sleep(random.uniform(0.1, 0.3))
        await queue.put(item)
        print(f"[生产者] 生产: {item} (当前队列长度: {queue.qsize()})")

async def consumer(name: str, queue: asyncio.Queue):
    while True:
        job = await queue.get()
        print(f"    [{name}] 开始处理: {job}")
        await asyncio.sleep(random.uniform(0.3, 0.6))
        
        # 必须标记完成，否则 queue.join() 会持续等待
        queue.task_done()
        print(f"    [{name}] 处理完毕: {job}")

async def main():
    # 限制容量为 3，提供背压机制，满额时生产者会暂停
    queue = asyncio.Queue(maxsize=3)

    # 启动 2 个消费者作为后台任务
    consumers = [
        asyncio.create_task(consumer(f"Worker-{i}", queue))
        for i in range(1, 3)
    ]

    await producer(queue, count=6)

    # 等待队列中所有已被消费的任务全部标记完成
    await queue.join()
    print("[主流程] 所有任务已完成")

    # 取消后台消费者并回收
    for c in consumers:
        c.cancel()
    await asyncio.gather(*consumers, return_exceptions=True)

asyncio.run(main())

八、完整流水线示例

这个示例综合了前述模块：异步生成器提供流式数据源，asyncio.Queue 控制缓冲与背压，asyncio.create_task 维护工作池，最后用 asyncio.gather 完成收尾。

import asyncio
import time

async def url_stream_generator(total_pages: int):
    for page in range(1, total_pages + 1):
        await asyncio.sleep(0.2)  # 模拟从数据库或上游接口分页拉取
        yield f"https://api.example.com/data?page={page}"

async def worker(worker_id: int, queue: asyncio.Queue, results: list):
    while True:
        url = await queue.get()
        try:
            print(f"[Worker-{worker_id}] 开始爬取: {url}")
            await asyncio.sleep(0.4)  # 模拟网络 I/O 耗时
            results.append({"url": url, "status": 200, "by": worker_id})
        finally:
            queue.task_done()  # 确保即便抛出异常也能正确递减计数

async def main():
    start_time = time.perf_counter()
    queue = asyncio.Queue(maxsize=5)
    results = []

    # 启动 3 个并发 Worker
    worker_tasks = [
        asyncio.create_task(worker(i, queue, results))
        for i in range(1, 4)
    ]

    # 通过 async for 流式投递任务
    print("[调度器] 开始流式投递任务...")
    async for url in url_stream_generator(total_pages=10):
        await queue.put(url)
        print(f"[调度器] 投递成功: {url}")

    # 等待队列数据全部消费完毕
    await queue.join()
    print("[调度器] 队列清空，任务全部完毕")

    # 退出 Worker 并收集状态
    for task in worker_tasks:
        task.cancel()
    await asyncio.gather(*worker_tasks, return_exceptions=True)

    print("\n运行统计:")
    print(f"成功处理: {len(results)} 条")
    print(f"总耗时: {time.perf_counter() - start_time:.2f} 秒")

asyncio.run(main())

九、常见问题与注意事项

1. 避免在协程中直接调用同步阻塞函数

在 async def 中调用 time.sleep()、requests.get() 或执行耗时的纯 CPU 计算，会阻塞当前主线程，进而使整个事件循环无法调度其他协程。

• 遇到网络 I/O，优先替换为 aiohttp、httpx 等异步库。
• 若必须调用无法修改的同步阻塞函数，使用 asyncio.to_thread 放入内置线程池执行，避免挂起主事件循环：

import time
import asyncio

async def main():
    res = await asyncio.to_thread(time.sleep, 2)

asyncio.run(main())

2. 遗漏 queue.task_done() 导致挂起

每次 put() 一个元素，队列内部的未完成任务计数器就会加 1；消费者 get() 取出元素并处理完成后，必须调用 queue.task_done() 将计数器减 1。一旦遗漏该调用，await queue.join() 的计数永远无法归零，主流程会一直挂起。建议在消费者中将 task_done() 放入 try…finally 块中。