Python 异步并发在测试基础设施里的正确用法

asyncio 最大的误用不是写错语法，而是用错地方：把"IO 密集"和"CPU 密集"混在一个事件循环里，或者在同步测试框架里硬塞 async。2025 年把测试基础设施（Mock WS 服务、设备调度、SSE 推送）整体迁到 asyncio，这篇总结哪些地方该异步、哪些地方坚持同步。

一、判断标准：你的瓶颈在等待吗

异步化的收益公式：并发数 × 等待时间占比。测试基础设施里各组件的画像：

组件	主要等待	该不该 asyncio
Mock WS 服务（50 连接收发）	网络 IO，99% 在等	✅ 核心场景
设备心跳聚合（50 台 × 10s）	网络 IO	✅
用例结果回收（大量小 HTTP）	网络 IO	✅
任务调度决策（装箱算法）	CPU，微秒级	❌ 纯同步函数
报告生成（HTML 渲染）	CPU + 少量文件 IO	❌ 同步或丢线程池
本地日志落盘	文件 IO，本地	❌ 同步写（本地盘够快）

经验法则：一个组件里超过 80% 的时间在等网络，才值得异步化；否则同步代码 + 线程池更简单可靠。Mock WS 服务是典型该异步的：同步 socket 方案下 50 个连接要 50 个线程，每线程一个 select 循环，内存和切换成本都难看；asyncio 下一个事件循环全搞定。

二、事件循环里的三个纪律

1. 禁止在协程里跑阻塞调用

这是 asyncio 头号事故来源：

# ❌ 灾难：整个事件循环卡 2 秒，50 个连接全断
async def handle_msg(conn):
    result = heavy_compute(payload)      # 阻塞！
    await conn.send(result)

# ✅ CPU 活丢线程池
async def handle_msg(conn):
    result = await asyncio.to_thread(heavy_compute, payload)
    await conn.send(result)

团队约定：协程里只允许 await 和纯计算（微秒级）。code review 时看到协程里出现 time.sleep / requests.post / 大循环，直接打回。

2. 任务生命周期要可取消

Mock 服务里"广播 500 个连接"这种 fan-out，取消语义必须正确：

async def broadcast(self, payload: dict):
    tasks = [conn.send_json(payload) for conn in self.clients.values()]
    # gather 返回即全部完成/失败；要中断时用 TaskGroup（3.11+）
    async with asyncio.TaskGroup() as tg:
        for t in tasks:
            tg.create_task(t)
    # 客户端断开 → 对应 task 抛 ConnectionClosed，TaskGroup 自动取消兄弟任务

TaskGroup 的价值在级联取消：一个慢客户端卡住，不影响其他客户端的完成；取消整个 broadcast 时，所有 in-flight 的 send 一起收掉。手写 gather + try/except 的取消逻辑，我们吃过"取消后还有一半 send 在飞"的亏。

3. 背压要显式

设备心跳聚合、SSE 推送都是"生产快于消费"的场景。无界队列 = 内存无限涨：

class HeartbeatAggregator:
    def __init__(self):
        # 有界队列 + 显式丢弃策略（心跳只认最新值，旧的直接丢）
        self.q = asyncio.Queue(maxsize=1000)

    async def push(self, hb):
        try:
            self.q.put_nowait(hb)
        except asyncio.QueueFull:
            self.dropped += 1        # 计数进监控，不阻塞生产方

丢弃策略必须显式声明（心跳丢旧、结果不可丢），而不是"队列满了才想办法"。测试结果回收这类不可丢的数据，用"满了就 await put（背压到上游）"，上游慢就整体变慢——宁可慢，不能丢。

三、和 pytest 的边界

测试框架本身（pytest）是同步的，async 组件怎么测？边界清晰化：

异步组件的单元测试：pytest-asyncio，每个测试一个干净的事件循环（event_loop fixture function 级）；
集成测试：异步服务跑在独立进程，pytest 走 HTTP/WS 客户端测它——跨进程边界就是天然的同步/异步隔离带，测试代码不需要 async；
禁止：在 pytest 的 fixture 里 loop.run_until_complete 手动管事件循环——99% 的"async 测试偶发挂"来自 fixture 里的事件循环泄漏。

四、可观测性：async 代码出事时最难查

同步代码的 traceback 是完整调用栈，async 代码的"现场"散落在 N 个协程里：

asyncio.all_tasks() + task.print_stack() 写进 /debug/dump 端点——服务卡住时 dump 全部协程栈，比猜快十倍；
每个长任务（>1s 的协程）进耗时看板：await 前后打点，"等待耗时 vs 计算耗时"分开统计——P99 劣化先看等待还是计算，方向就对了；
结构化日志带 task_id：一个业务帧的生命周期（接收→处理→下发）全部可串联。

五、小结

asyncio 在测试基础设施里的正确姿势：IO 密集组件（Mock WS、心跳聚合、结果回收）用 asyncio，CPU 活和同步框架坚持原样；事件循环三纪律（不阻塞、可取消、显式背压）；测试边界跨进程隔离。异步不是架构信仰，是"等待占比 80%"这个数字的函数。

ScofieldYu