Python 与数据实践 · 第 21 章

asyncio 实战

第 19 章拆解了事件循环的运转模型，第 20 章又用 Future 和协程手搓了一套驱动机制。造轮子的部分到此为止——从本章起，我们把自制的零件全部换成官方和社区的现成 API，用工程级的姿势写异步代码。你会看到：之前…

约 34 分钟13,548 字39 个代码示例

第 19 章拆解了事件循环的运转模型，第 20 章又用 Future 和协程手搓了一套驱动机制。造轮子的部分到此为止——从本章起，我们把自制的零件全部换成官方和社区的现成 API，用工程级的姿势写异步代码。你会看到：之前费尽心思实现的东西，标准库里都有名字。

本章示例统一沿用一个场景：并发拉取多个数据源的指标——pull_metrics(source, cost) 模拟"从 source 拉取指标，耗时 cost 秒"：

async def pull_metrics(source: str, cost: float) -> str:
    await asyncio.sleep(cost)
    return f"{source} 指标就绪"

1. 先补齐两个异步协议

第 11 章讲过迭代协议，第 12 章讲过上下文管理协议。它们的异步版本，语法只是多一个 a。

1.1 异步生成器：`async def` 里写 `yield`

在协程函数中使用 yield，得到的就是异步生成器——每次产出值之前，都可以先 await 一段异步操作：

async def metric_stream(rounds: int):
    for i in range(rounds):
        await asyncio.sleep(1)
        yield f"第 {i + 1} 批指标"


async def main() -> None:
    # 异步生成器必须用 async for 遍历
    async for batch in metric_stream(3):
        print(batch)  # 每隔 1 秒打印一批


asyncio.run(main())

1.2 自定义异步迭代器

和同步迭代器一样，也可以自己实现协议方法，只不过这回是 __aiter__ 和 __anext__：

class Ticker:
    """每隔 interval 秒报一次数，共报 count 次"""

    def __init__(self, count: int, interval: float = 0.5):
        self._count = count
        self._interval = interval
        self._current = 0

    def __aiter__(self) -> "Ticker":
        return self

    async def __anext__(self) -> int:
        if self._current >= self._count:
            raise StopAsyncIteration
        self._current += 1
        await asyncio.sleep(self._interval)
        return self._current


async def main() -> None:
    async for seq in Ticker(5):
        print(seq)  # 每隔 0.5 秒打印 1 2 3 4 5

1.3 异步上下文管理器：`async with`

async with 背后是 __aenter__ / __aexit__ 一对协议方法——进出上下文本身也可能是耗时操作，比如建立和断开一次远程连接：

class DataSource:
    """模拟一个握手、挥手都要花时间的远程数据源"""

    async def __aenter__(self) -> "DataSource":
        print("建立连接")
        await asyncio.sleep(0.5)  # 握手耗时
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        print("断开连接")
        await asyncio.sleep(0.5)  # 挥手耗时


async def main() -> None:
    async with DataSource() as ds:
        print("读取数据中")

两套协议放在一起对照：

同步版	异步版
`__iter__` / `__next__`	`__aiter__` / `__anext__`
`for x in it`	`async for x in ait`
`StopIteration`	`StopAsyncIteration`
`def` + `yield`（生成器）	`async def` + `yield`（异步生成器）
`__enter__` / `__exit__`	`__aenter__` / `__aexit__`
`with ctx`	`async with actx`

2. 任务调度全家桶

2.1 `sleep`：官方版的自制闹钟

第 20 章我们在 demo/alarm.py 里手写过"等 N 秒"：

def ring_after(seconds: float) -> asyncio.Future:
    loop = asyncio.get_event_loop()
    fut = loop.create_future()
    loop.call_later(seconds, fut.set_result, None)
    return fut

官方的 asyncio.sleep 就是它的工业版，用法零门槛：

async def main() -> None:
    print("开始")
    await asyncio.sleep(1)  # 当前协程挂起 1 秒，事件循环趁机去调度别的协程
    print("一秒后")

2.2 `asyncio.run` 与 `create_task`

asyncio.run 是异步程序的最高层入口——上一章已经拆过它的内幕（创建循环、包装 Task、跑到完成、收尾关闭）：

result = asyncio.run(pull_metrics("MySQL", 1))

create_task 把协程立刻推上跑道，你自己先去忙别的：

async def main() -> None:
    task = asyncio.create_task(pull_metrics("MySQL", 1))  # 协程立即开始排队执行
    print("先干点别的")  # 此刻任务在后台跑
    result = await task  # 需要结果时再来等
    print(result)

2.3 `gather`：并发收账

多个协程同时开跑、等全部完成、按传入顺序收结果——这正是第 20 章作业 collect_all 干的事，官方版本就是 gather：

async def main() -> None:
    outcomes = await asyncio.gather(
        pull_metrics("MySQL", 2),
        pull_metrics("Redis", 1),
        pull_metrics("ES", 3),
    )
    print(outcomes)  # ['MySQL 指标就绪', 'Redis 指标就绪', 'ES 指标就绪']


asyncio.run(main())

四个特性记牢：

所有协程并发执行，总耗时取决于最慢的那个（上面是 3 秒，不是 6 秒）；
结果顺序与传入顺序一致，与完成顺序无关；
默认情况下，任何一个协程抛异常，gather 会立即把异常向外抛（其余协程不会因此取消，但它们的结果拿不到了）；
想一个都不放过，用 return_exceptions=True——异常被当作"结果"原样返回：

async def broken_source() -> str:
    raise ConnectionError("数据源失联")


async def main() -> None:
    outcomes = await asyncio.gather(
        pull_metrics("MySQL", 1),
        broken_source(),
        return_exceptions=True,  # 异常不抛出，按位置躺进结果列表
    )
    print(outcomes)  # ['MySQL 指标就绪', ConnectionError('数据源失联')]

2.4 `wait`：等待策略自选

gather 只回答"全部完成了吗"，wait 提供三种撤退时机：

async def main() -> None:
    tasks = [
        asyncio.create_task(pull_metrics("A", 2)),
        asyncio.create_task(pull_metrics("B", 1)),
        asyncio.create_task(pull_metrics("C", 3)),
    ]

    # FIRST_COMPLETED: 任意一个完成就返回
    # FIRST_EXCEPTION: 任意一个抛异常就返回
    # ALL_COMPLETED:   全部完成才返回（默认）
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    print(f"已完成 {len(done)} 个，未完成 {len(pending)} 个")

    for task in pending:  # 剩下的任务自行处置——取消、或继续等待
        task.cancel()

2.5 `as_completed`：谁先到先处理谁

gather 的结果列表按传入顺序排，as_completed 反其道而行——按完成顺序交付：

async def main() -> None:
    jobs = [pull_metrics("A", 2), pull_metrics("B", 1), pull_metrics("C", 3)]
    for job in asyncio.as_completed(jobs):
        print(await job)  # B → A → C，先完成先打印

典型用途：多个镜像源下载同一份文件，谁先好就用谁。

2.6 `TaskGroup`：结构化并发（Python 3.11+）

async with 一包，作用域内的任务一个都跑不掉：

async def main() -> None:
    # with 块退出前，组内所有任务必然已完成；
    # 若某个任务抛异常，组内其余任务会被自动取消
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(pull_metrics("A", 2))
        t2 = tg.create_task(pull_metrics("B", 1))
        t3 = tg.create_task(pull_metrics("C", 3))

    # 走到这里，三个任务都已安全落地
    print(t1.result(), t2.result(), t3.result())

相比 gather，TaskGroup 的承诺更严格：异常发生时整体止损（其余任务全部取消），而不是让它们孤零零跑完再把结果丢弃。

3. 并发秩序工具箱

协程虽是单线程串行，但"多个协程协同办事"依然会产生秩序问题：抢共享数据、等信号、抢名额、传数据。四个工具各管一摊。

3.1 `Lock`：同一时刻只许一个

多个协程更新同一份数据时，await 切换点之间可能被插队，读-改-写就乱了。Lock 保证持有者独占：

counter: int = 0
lock = asyncio.Lock()


async def bump() -> None:
    global counter
    async with lock:  # 他人持锁时，此处排队等待
        snapshot = counter
        await asyncio.sleep(0)  # 模拟一次耗时处理：此刻让出线程也不会算错
        counter = snapshot + 1


async def main() -> None:
    await asyncio.gather(*[bump() for _ in range(100)])
    print(counter)  # 100 —— 去掉锁试试，多半是个小于 100 的数


asyncio.run(main())

3.2 `Event`：广播一次信号

一个协程等信号，另一个协程发信号（第 20 章作业三 Gate 的官方完全体，支持多个等待者）：

async def standby(flag: asyncio.Event) -> None:
    print("下游服务: 就绪，等放行信号")
    await flag.wait()
    print("下游服务: 收到信号，开始消费")


async def controller(flag: asyncio.Event) -> None:
    print("调度中心: 上游数据校验中……")
    await asyncio.sleep(1)
    flag.set()  # 广播信号，所有等待者集体放行


async def main() -> None:
    flag = asyncio.Event()
    await asyncio.gather(standby(flag), controller(flag))


asyncio.run(main())

3.3 `Semaphore`：并发名额限制

十万个协程同时打一个接口，接口先崩为敬。Semaphore 给并发量上闸：

slots = asyncio.Semaphore(3)  # 同时最多 3 个


async def guarded_pull(source: str) -> str:
    async with slots:  # 超出配额的协程在此排队
        print(f"开始拉取 {source}")
        await asyncio.sleep(1)
        print(f"完成拉取 {source}")
        return source


async def main() -> None:
    sources = [f"数据源{i}" for i in range(10)]
    await asyncio.gather(*[guarded_pull(s) for s in sources])
    # 观察输出：前 3 个先动，之后每完成一个才放进来一个

3.4 `Queue`：生产者与消费者

跨协程传递数据的标准通道，接口和 queue.Queue 高度相似，只是读写都变成了 await：

import random


async def feeder(queue: asyncio.Queue) -> None:
    """生产者：持续采集日志事件"""
    for i in range(10):
        event = f"日志#{i}"
        await queue.put(event)
        print(f"采集到 {event}")
        await asyncio.sleep(random.random())
    await queue.put(None)  # 哨兵：告诉下游"没有更多了"


async def handler(name: str, queue: asyncio.Queue) -> None:
    """消费者：从队列里取事件并处理"""
    while True:
        event = await queue.get()
        if event is None:  # 收到哨兵，收工
            queue.task_done()
            break
        print(f"{name} 处理 {event}")
        queue.task_done()


async def main() -> None:
    queue = asyncio.Queue(maxsize=5)  # 满了生产者会被挂起，形成背压
    await asyncio.gather(
        feeder(queue),
        handler("处理器1", queue),
        handler("处理器2", queue),
    )

4. 超时控制

网络那头永远可能不回话，等待必须有底线。

4.1 `wait_for`

给单个等待加一个闹钟：

async def sluggish_source() -> str:
    await asyncio.sleep(10)
    return "迟到的指标"


async def main() -> None:
    try:
        result = await asyncio.wait_for(sluggish_source(), timeout=2)
    except TimeoutError:
        print("等不了了，超时放弃")

4.2 `asyncio.timeout`（Python 3.11+）

用上下文管理器圈起一段"限时区"，区内的所有 await 共享同一个预算：

async def main() -> None:
    try:
        async with asyncio.timeout(2):
            result = await sluggish_source()
    except TimeoutError:
        print("等不了了，超时放弃")

5. 阻塞代码的安置

现实的代码库里总有些同步阻塞函数（老 SDK、重量级计算），直接扔进协程会按住整个事件循环。正确做法是给它们另找去处：

import time


def read_big_csv() -> str:
    time.sleep(0.5)  # 同步读取大文件的耗时
    return "csv 全部行"


def crunch_numbers() -> int:
    return sum(i * i for i in range(10_000_000))  # CPU 密集计算


async def main() -> None:
    # to_thread：把同步阻塞函数丢进线程池执行，完成后结果送回来
    text = await asyncio.to_thread(read_big_csv)
    print(text)

    # run_in_executor：更底层的写法，可以自己指定执行器
    loop = asyncio.get_running_loop()
    total = await loop.run_in_executor(None, crunch_numbers)
    print(total)

一句话区分：I/O 阻塞扔线程池，CPU 密集扔进程池（ProcessPoolExecutor），别让任何一种堵在事件循环的线程上。

6. 第三方异步生态

标准库只解决了调度问题，具体的网络与文件 I/O 还得靠生态。

6.1 网络请求

aiohttp —— 老牌异步 HTTP 客户端（也是服务端框架）：

import aiohttp


async def fetch_page(url: str) -> str:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()


async def main() -> None:
    html = await fetch_page("https://example.com")
    print(len(html))

httpx —— 同步/异步双模式，API 对齐 requests，上手成本几乎为零：

import httpx


async def fetch_page(url: str) -> str:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.text

6.2 文件 I/O

aiofiles —— 把文件读写搬进线程池再包上 await，接口和内置 open 保持一致：

import aiofiles


async def io_demo() -> None:
    # 写文件
    async with aiofiles.open("dataset.txt", "w") as f:
        await f.write("第一行数据\n")

    # 读文件
    async with aiofiles.open("dataset.txt", "r") as f:
        content = await f.read()
        print(content)

7. 四个高频模式（面试常客）

7.1 并发批处理

"一批任务并发执行、结果统一收集"，是异步世界里最常见的一块积木：

async def check_batch(sources: list[str]) -> list[dict]:
    """并发探测一批数据源，收集各自的健康状态"""

    async def probe(source: str) -> dict:
        async with httpx.AsyncClient() as client:
            resp = await client.get(source)
            return {"source": source, "status": resp.status_code}

    return await asyncio.gather(*[probe(s) for s in sources])

7.2 限速器

Semaphore 只管"同时在跑几个"，不管"每秒放行几个"。后者需要配合定时归还：

class RequestPacer:
    """限制单位时间内的请求量：窗口内超出配额的请求排队等待"""

    def __init__(self, quota: float, window: float = 1.0):
        self._slots = asyncio.Semaphore(quota)
        self._window = window

    async def acquire(self) -> None:
        await self._slots.acquire()

        def give_back() -> None:
            self._slots.release()  # 窗口期一到，名额归还

        loop = asyncio.get_running_loop()
        loop.call_later(self._window, give_back)

    async def __aenter__(self) -> None:
        await self.acquire()

    async def __aexit__(self, *args) -> None:
        pass


async def main() -> None:
    pacer = RequestPacer(quota=2, window=1.0)

    async def call(source: str) -> str:
        async with pacer:
            await asyncio.sleep(0.3)
            return f"{source} 调用完成"

    outcomes = await asyncio.gather(*[call(f"接口{i}") for i in range(6)])
    print(outcomes)  # 观察完成时间：每秒最多放行 2 个


asyncio.run(main())

7.3 重试

不稳定的网络调用，值得一层"失败歇会儿再试"的包装：

async def with_retry(make_call, max_attempts: int = 3, pause: float = 1.0):
    """为不稳定的异步调用添加重试机制"""
    for attempt in range(max_attempts):
        try:
            return await make_call()
        except Exception:
            if attempt == max_attempts - 1:
                raise  # 最后一次也失败，如实向上抛
            print(f"第 {attempt + 1} 次失败，{pause} 秒后再来")
            await asyncio.sleep(pause)


async def main() -> None:
    calls = 0

    async def flaky_api() -> str:
        nonlocal calls
        calls += 1
        if calls < 3:
            raise ConnectionError(f"第 {calls} 次调用失败")
        return "调用成功"

    print(await with_retry(flaky_api, max_attempts=3, pause=0.5))
    # 前两次失败，第三次成功


asyncio.run(main())

注意参数是 make_call 这样"生成协程的工厂"而非协程本身——协程只能被 await 一次，重试每次都要新建。

7.4 优雅关闭

服务不能"啪"地断电，得先谢客、再冲缓冲、最后熄灯：

import signal


class IngestService:
    def __init__(self) -> None:
        self._alive = True

    async def ingest(self) -> None:
        while self._alive:
            try:
                await asyncio.sleep(1)  # 模拟接入一批数据
                print("接入一批数据……")
            except asyncio.CancelledError:
                print("收到取消信号，冲刷缓冲区……")
                break

    def halt(self) -> None:
        print("开始优雅停机……")
        self._alive = False

    async def run(self) -> None:
        loop = asyncio.get_running_loop()
        stop = loop.create_future()

        def on_signal() -> None:
            stop.set_result(None)

        loop.add_signal_handler(signal.SIGINT, on_signal)   # Ctrl+C
        loop.add_signal_handler(signal.SIGTERM, on_signal)  # kill 的默认信号

        task = asyncio.create_task(self.ingest())
        await stop     # 停在这里，等停机信号
        self.halt()
        task.cancel()  # 取消接入任务
        await task     # 等它把取消异常处理完


async def main() -> None:
    service = IngestService()
    await service.run()


# 按 Ctrl+C 触发 SIGINT：程序按部就班收尾退出，而不是当场崩出堆栈
asyncio.run(main())

章节根目录的 demo.py 是这个模式的最小可运行版本，跑起来按一次 Ctrl+C 感受一下。

8. 踩坑清单

异步代码的 bug 大多不报错，只是"悄悄不对"。七条军规收好。

8.1 不要在协程中调用同步阻塞函数

import time


async def bad() -> None:
    time.sleep(1)  # 错！整个事件循环被按住，所有协程集体罚站


async def good() -> None:
    await asyncio.sleep(1)  # 对！主动让出线程


async def acceptable() -> None:
    await asyncio.to_thread(time.sleep, 1)  # 也行！让线程池替你阻塞

8.2 入口始终交给 `asyncio.run`

# 不推荐：手动搭建循环，收尾全靠自己记
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
task = loop.create_task(main())
loop.run_forever()

# 推荐：一行包办循环的创建、运行与关闭
asyncio.run(main())

8.3 小心没被 `await` 的协程对象

async def main() -> None:
    pull_metrics("MySQL", 1)  # 错！只创建了协程对象，永远不会执行，也不报错

    await pull_metrics("MySQL", 1)  # 对

    # 或者交给 gather 统一驱动
    await asyncio.gather(pull_metrics("MySQL", 1), pull_metrics("Redis", 2))

8.4 `gather` 的异常口径

默认模式下，一个协程抛异常、gather 立即向外抛，其余协程继续跑完但结果被丢弃。两种更稳妥的写法：

async def main() -> None:
    # 路线一：return_exceptions=True，事后逐个检查
    outcomes = await asyncio.gather(
        risky_job(),
        safe_job(),
        return_exceptions=True,
    )
    for r in outcomes:
        if isinstance(r, Exception):
            print(f"有任务翻车了: {r}")

    # 路线二：TaskGroup（Python 3.11+）
    # 任一任务异常会自动取消组内其余任务，一个都不留

8.5 用好 debug 模式

asyncio 的调试模式专抓三类隐形事故：协程阻塞循环太久、协程被创建却从未 await、回调执行超时。

# 开启调试模式
asyncio.run(main(), debug=True)

# 或者通过环境变量
# PYTHONASYNCIODEBUG=1 python demo.py

抓长时间阻塞——事件循环给每段协程执行计时，超过阈值就在 stderr 打印警告和执行栈，精确到行：

import time


async def slow_step() -> None:
    print("开始一段同步重活……")
    time.sleep(0.2)  # 同步阻塞：整个循环被按住
    print("重活干完了")


async def main() -> None:
    await slow_step()


asyncio.run(main(), debug=True)

输出类似：

开始一段同步重活……
重活干完了
Executing <Task ...> took 0.200 seconds
    [...执行栈直接指向 time.sleep 所在的行...]

抓未 await 的协程——新手最常犯的"静默丢失"：

async def main() -> None:
    pull_metrics("MySQL", 0.5)  # 忘了 await，协程对象被创建后无人问津
    await asyncio.sleep(1)


asyncio.run(main(), debug=True)

输出类似：

Coroutine 'pull_metrics' was never awaited (at demo.py:...)

自定义慢的标准——默认 0.1 秒就报警，嫌它大惊小怪可以调：

async def main() -> None:
    loop = asyncio.get_running_loop()
    loop.slow_callback_duration = 0.5  # 慢的门槛放宽到 0.5 秒

    def heavy_callback() -> None:
        time.sleep(0.3)  # 0.3 秒 < 0.5 秒，不再触发报警

    loop.call_later(0.1, heavy_callback)
    await asyncio.sleep(0.5)


asyncio.run(main(), debug=True)

8.6 避免抓全局事件循环

# 错误：在模块级别抓事件循环，抓到的多半不是想要的那个
loop = asyncio.get_event_loop()

# 正确：进入协程之后再取"当前正在运行"的循环
async def my_func() -> None:
    loop = asyncio.get_running_loop()

8.7 `CancelledError` 的收尾礼节

捕获 CancelledError 做清理后，必须重新抛出——吞掉它等于拒绝取消，任务会卡在"半死不活"的状态：

async def cleanup_demo() -> None:
    try:
        await long_running_task()
    except asyncio.CancelledError:
        await release_resources()  # 先把资源收拾干净
        raise  # 然后必须让取消继续传播

作业（使用 AI 辅助）

本章的作业是读懂 answer/ 目录下的完整工程：一个多模块协作的自动调研流水线——输入一个调研命题，系统自动把命题拆成若干独立子课题、多路并行调查、把材料拼装成稿、再由大模型提炼成一篇结构完整的调研报告，最后归档落盘。

阅读时带着三个问题：

conductor.py 里子课题的并发执行为什么选 gather + return_exceptions=True，而不是 TaskGroup？两种选择在"个别子课题失败"时的表现差异是什么？
llm_client.py 里的 stream_chat 是一个异步生成器，它产出的分片在哪里被消费？换成一次性返回完整文本，会影响哪些环节？
整条流水线里，哪些步骤是纯函数（无需 await），哪些是异步 I/O？如果把并行的子课题调查改成串行，总耗时会怎么变？

建议的阅读顺序：settings.py → llm_client.py → binder.py → filestore.py → field_agent.py → splitter.py → digest.py → conductor.py → main.py（README.md 里有完整的模块依赖图）。读完之后，把调研命题换成你真正关心的主题（比如某类云服务的选型），配置好 .env 跑一遍，看看产出的报告。

下一章：22 · 构建与发布

1. 先补齐两个异步协议

1.1 异步生成器：async def 里写 yield

1.2 自定义异步迭代器

1.3 异步上下文管理器：async with

2. 任务调度全家桶

2.1 sleep：官方版的自制闹钟

2.2 asyncio.run 与 create_task

2.3 gather：并发收账

2.4 wait：等待策略自选

2.5 as_completed：谁先到先处理谁

2.6 TaskGroup：结构化并发（Python 3.11+）