Hotdry.
归档
systems-engineering

Python asyncio：协程与 Task 的调度差异及高并发优化

剖析 await 非上下文切换：协程 yield/resume 用户态调度 vs Task 并发执行，避免 OS 开销与 GIL 瓶颈的高并发 I/O 参数与清单。

在 Python asyncio 中，许多开发者误以为 await 会引发类似于线程的上下文切换，导致高并发场景下性能担忧。实际上，await 仅触发协程的 yield 操作，这是用户态的轻量级暂停与恢复机制，由事件循环在单线程内协作调度，无操作系统内核介入的开销。这种设计特别适合 I/O 密集型任务，能有效规避 GIL 瓶颈与线程切换的毫秒级延迟。

协程 yield/resume 的非切换本质

协程（coroutine）通过 async def 定义，调用返回协程对象而不立即执行。只有在事件循环中调度时才运行。await coroutine 时，当前协程挂起等待目标协程完成，整个过程是串行阻塞：事件循环不会切换其他任务，直到 await 完成。例如：

import asyncio
import time

async def coro1(): await asyncio.sleep(2); return "c1"
async def coro2(): await asyncio.sleep(1); return "c2"

async def main():
    start = time.time()
    r1 = await coro1()  # 阻塞 2s
    r2 = await coro2()  # 再阻塞 1s
    print(time.time() - start)  # ~3s

asyncio.run(main())

这里总耗时约 3 秒，因为 await 直接推进协程执行，直至完成再恢复调用者。底层，协程帧（frame）状态保存到堆栈，yield 点仅数十字节用户态拷贝，无寄存器保存或 TLB 刷新，与 OS 线程切换（~1-10μs vs 数 μs）天差地别。

Task：并发调度的关键封装

Task 是 Future 子类，对协程的 “执行代理”：asyncio.create_task(coro) 立即将协程注册事件循环的就绪队列（ready queue），后台并发运行。主协程可继续，而非阻塞等待。

async def main():
    start = time.time()
    t1 = asyncio.create_task(coro1())
    t2 = asyncio.create_task(coro2())
    r1, r2 = await t1, await t2
    print(time.time() - start)  # ~2s（max(2,1)）

asyncio.run(main())

总耗时降至最长任务时间（2s），证明并发：任务创建即调度，await task 仅检查 Future 状态，若 done 则立即返回。事件循环在 _run_once() 迭代中，按优先级（延迟队列 → I/O 回调 → 就绪回调）轮询 ready 队列，实现公平调度。

引用官方文档：“Task 是 Future 的子类，代表协程执行，用于并发调度。” 这避免了串行累加延迟。

高并发 I/O 优化：参数与阈值

为工程化高并发（如 API 爬虫、Web 服务），需参数化管理 Task 生命周期，避免泄漏或饥饿。

  1. 并发控制：

    • asyncio.gather(*coros, return_exceptions=True)：批量并发，异常不中断。阈值：任务数 ≤ CPU*10（I/O 密集），>100 用 asyncio.Semaphore(50) 限流。
    • asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED, timeout=30)：部分完成即返回，超时回收。生产阈值：单个 Task ≤5s，全批 ≤60s。

  2. 监控与回滚：

    • task.done() / task.cancelled()：轮询状态。回调：task.add_done_callback(lambda t: print(t.result() if t.done() else 'cancelled'))。
    • 超时策略：await asyncio.wait_for(task, timeout=10)，超时时 CancelledError，回滚重试（指数退避：0.1s * 2^n，n≤5）。
    • 资源管理：async with asyncio.TaskGroup() as tg: tg.create_task(coro)（Python 3.11+），自动取消子任务。

  3. 避免 GIL/OS 瓶颈：

    • 纯 I/O：单事件循环足矣，10k+ 连接无压力（uvloop 替换提升 2-4x）。
    • CPU 混杂：loop.set_task_factory 优先 I/O Task；>50% 计算用 asyncio.to_thread(sync_func) 卸载。
    • 阈值监控：Prometheus 指标 task_queue_size > 1000 告警，switch_count/sec > 1e6 调 loop.slow_callback_duration=0.1 日志慢协程。
参数 推荐值 作用
gather return_exceptions True 容错
wait timeout 30s 防挂起
Semaphore 100 限流
wait_for timeout 5s 单任务

落地清单

  • 启动：asyncio.run(main())，禁用调试 loop.set_debug(False) 生产。
  • 并发：优先 asyncio.gather / as_completed(tasks) 迭代最快完成者。
  • 异常：try: await task except asyncio.CancelledError: logger.warning("Task cancelled")。
  • 规模：>1k Task 用 asyncio.all_tasks() 周期清理 if not t.done(): t.cancel()。
  • 基准：压测 1k 并发，目标 P99 <100ms，CPU <50%。

此机制下，单核处理万级 I/O 并发，远超多线程（GIL + 开销）。实际如 FastAPI 默认事件循环，吞吐提升 5-10x。

资料来源：Python 官方 asyncio 文档、Stack Overflow 讨论及工程实践（如 uvloop 基准）。

（正文约 950 字）

查看归档