asyncio

asyncio 用 async/await 编写 单线程、协作式 的并发：在等待 I/O 时让出控制权，由 事件循环 调度其他协程。适合 I/O 密集、结构化 的网络服务、客户端与任务编排。同进程内需要 OS 线程时见 threading；CPU 密集 纯 Python 计算见 multiprocessing；「提交任务、取 Future」可用 concurrent.futures 的 ThreadPoolExecutor / ProcessPoolExecutor（常与 asyncio 的 run_in_executor 配合）。

适用性：在 wasm32-emscripten / wasm32-wasi 上 不可用（见 WebAssembly 平台）。

何时用协程

协程函数 vs 协程对象：async def foo() 定义的是 协程函数；调用 foo() 得到 协程对象。仅创建对象 不会执行，须 await 或交给 create_task / asyncio.run 调度。

直接运行的 Python 脚本默认是单进程、单线程的；asyncio.run(main()) 会创建事件循环并在结束时关闭。

模块成员总览

asyncio 分 高层 API（应用编写）与 低层 API（库作者操作事件循环）。下文侧重常用高层入口。

运行与调度

同步与队列（协程专用，非线程安全）

子模块（按需深入）

可等待对象与基本写法

三类主要 可等待（awaitable）对象：协程、Task、Future。

import asyncio

async def main():
    print('hello')
    await asyncio.sleep(1)
    print('world')

asyncio.run(main())

import asyncio
import time

async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)

async def main():
    print(f"started at {time.strftime('%X')}")
    async with asyncio.TaskGroup() as tg:
        tg.create_task(say_after(1, 'hello'))
        tg.create_task(say_after(2, 'world'))
    print(f"finished at {time.strftime('%X')}")  # 约 2 秒，非 3 秒

asyncio.run(main())

3.11 之前无 TaskGroup 时，常用 task1 = create_task(...); task2 = create_task(...); await task1; await task2。

并发等待：gather 与超时

import asyncio

async def factorial(name, n):
    f = 1
    for i in range(2, n + 1):
        await asyncio.sleep(0.1)
        f *= i
    return f

async def main():
    results = await asyncio.gather(
        factorial('A', 3),
        factorial('B', 4),
    )
    print(results)

asyncio.run(main())

gather 被取消时，默认会取消尚未完成的子任务；TaskGroup 在失败传播与嵌套取消上语义更 结构化，新代码可优先 TaskGroup。

任务取消与后台任务

• task.cancel() 会在协程内引发 asyncio.CancelledError（继承 BaseException）；清理逻辑放 try/finally。
• 不要 随意吞掉 CancelledError 或调用 uncancel()，否则 TaskGroup / timeout() 可能行为异常。
• 发射后不管 的后台任务：必须保存强引用，否则可能被 GC：

background_tasks = set()

def fire_and_forget(coro):
    task = asyncio.create_task(coro)
    background_tasks.add(task)
    task.add_done_callback(background_tasks.discard)

异步同步原语

asyncio 的 Lock、Event 等与 threading 不能混用：它们 不是线程安全的，只用于 同一事件循环内 的协程协调。原语方法 没有 timeout 参数，用 asyncio.wait_for() 包一层。

import asyncio

async def worker(lock, i):
    async with lock:
        print('worker', i)

async def main():
    lock = asyncio.Lock()
    async with asyncio.TaskGroup() as tg:
        for i in range(4):
            tg.create_task(worker(lock, i))

asyncio.run(main())

Condition 须用 while 或 wait_for(predicate) 检查条件（虚假唤醒）。Queue：await q.put(item) / await q.get()，适合协程间生产者-消费者。

与阻塞代码协作

在协程中 禁止 长时间阻塞调用（time.sleep、同步 requests.get、阻塞式 threading.Lock 等）——会冻结整个循环。

CPU 密集若放进 to_thread / 线程池，仍受 GIL 限制；真要多核用进程池。

网络与子进程（入口）

• 流：asyncio.open_connection / start_server 返回 StreamReader/StreamWriter，读写用 await reader.read()、writer.write() + drain()。
• 子进程：create_subprocess_exec / create_subprocess_shell（见 asyncio.subprocess）。
• REPL：python -m asyncio 可直接 await（3.12+ 审计事件），便于试验。

应用开发通常 只 用 asyncio.run + 高层 API；自定义事件循环、信号处理、跨线程 call_soon_threadsafe 见官方「事件循环」与「开发」章节。

相关模块速查

易错点（排错速记）

• 只调用 main() 不 run / 不 await：得到协程对象，代码不执行。
• 协程里 time.sleep / 同步 HTTP / 阻塞锁：卡住事件循环，表现为「全站卡死」。
• create_task 不保存引用：后台任务可能被 GC，中途消失。
• 把 asyncio.Lock 当线程锁：多线程同时 acquire 不安全。
• gather 里异常未处理：默认一个失败即抛，其它任务可能仍在跑。
• 在 TaskGroup 内吞 CancelledError：结构化取消链断裂。
• CPU 密集全写 async：算力仍单核；应 offload 到进程池或 C 扩展。
• 与 threading / multiprocessing 混用同一锁：语义不同，不可互换。
• WebAssembly：本模块不可用。

资料：asyncio 文档 · 协程与任务 · 同步原语