3. 协程与任务编排

在 Python 异步编程中，async def 定义的是协程函数，调用它并不会立即执行函数体，而是返回一个协程对象。协程对象是“惰性”的：只有被 await 驱动、被包装成 Task 并交给事件循环，或者被 asyncio.run() 接管时，它才会真正开始运行。理解“协程对象—Task—事件循环”这三者的关系，是掌握后续 gather、wait、超时与取消机制的基础。

一、async/await 与协程对象

async def 声明的函数是协程函数，调用后得到协程对象。协程对象本身不会执行，必须被 await 或调度。例如：

import asyncio

async def fetch(name, delay):
    print(f"{name} 开始")
    await asyncio.sleep(delay)
    print(f"{name} 结束")
    return f"{name}-result"

async def main():
    coro = fetch("A", 1)      # 只是协程对象，未执行
    print(type(coro))          # <class 'coroutine'>
    result = await coro        # 此时才真正运行
    print(result)

asyncio.run(main())

这里 fetch("A", 1) 返回的是协程对象，await coro 才把它交给事件循环执行。若只创建协程对象而不 await，Python 会在垃圾回收时给出 “coroutine was never awaited” 警告。协程对象只能被 await 一次，重复 await 会抛出 RuntimeError: cannot reuse already awaited coroutine。

二、Task：把协程交给事件循环并发调度

协程对象是“被动”的，而 Task 是“主动”的：它把协程注册到事件循环，使其可以与其他任务并发推进。asyncio.create_task(coro) 会立即安排协程运行，并返回一个 Task 对象。Task 是 Future 的子类，因此可以 await、可以取消、可以查询状态。

async def main():
    t1 = asyncio.create_task(fetch("A", 1))
    t2 = asyncio.create_task(fetch("B", 2))
    print(t1.done())        # False
    r1 = await t1
    r2 = await t2
    print(r1, r2)

关键区别：await coro 是串行等待，前一个完成才轮到下一个；create_task 则让两个协程几乎同时开始，总耗时约等于较慢的那个（约 2 秒），而不是 3 秒。这正是异步并发的价值所在。需要注意：create_task 必须在事件循环运行中调用，且要保存返回的 Task 引用，否则任务可能被垃圾回收而中途取消。

三、gather 与 wait：批量编排的两种方式

当需要同时等待多个任务时，asyncio.gather 和 asyncio.wait 是两种常用工具，但语义不同。

gather(*aws) 接收多个可等待对象，返回一个聚合 Future，结果按传入顺序组成列表。默认 return_exceptions=False：任一任务抛异常，gather 立即向外抛出该异常，其余任务不会被取消，但结果不再返回。若设为 True，异常会作为结果元素返回，便于逐个检查。

async def main():
    results = await asyncio.gather(
        fetch("A", 1),
        fetch("B", 2),
        fetch("C", 0.5),
    )
    print(results)  # ['A-result', 'B-result', 'C-result']

wait(aws, timeout=None, return_when=...) 返回 (done, pending) 两个集合，不直接给结果，需要自行从 Task 中取 result()。return_when 可取 FIRST_COMPLETED、FIRST_EXCEPTION、ALL_COMPLETED。它适合“谁先完成就先处理谁”的场景，例如竞速、健康检查、多源取最快响应。

done, pending = await asyncio.wait(tasks, timeout=1.5,
                                   return_when=asyncio.FIRST_COMPLETED)
for t in done:
    print("完成:", t.result())
for t in pending:
    t.cancel()

选择建议：需要按顺序收集全部结果用 gather；需要精细控制“何时返回、哪些还在跑”用 wait。注意 wait 在 Python 3.8 之后不再接受协程对象，必须传入 Task 或 Future。

四、超时与取消

超时控制有两种主流写法。Python 3.11 起推荐 asyncio.timeout() 上下文管理器：

async def main():
    try:
        async with asyncio.timeout(2.0):
            await fetch("slow", 5)
    except TimeoutError:
        print("超时，已取消")

旧版本可用 asyncio.wait_for(coro, timeout)，超时会取消内部任务并抛出 TimeoutError。取消的本质是向 Task 注入 CancelledError，协程在下一个 await 点收到该异常。因此协程中若捕获 Exception 要小心，不要把 CancelledError 一并吞掉；正确做法是捕获后重新抛出，或在 finally 中做清理。

async def worker():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("收到取消，清理资源")
        raise          # 必须重新抛出
    finally:
        print("finally 清理")

主动取消用 task.cancel()，随后 await task 会抛出 CancelledError。若任务已进入不可中断的同步阻塞代码，取消不会立即生效，需等其回到 await 点。

五、异常传播与结构化并发

异常传播遵循“谁 await 谁负责”的原则：Task 内部未捕获的异常会保存在 Task 中，直到有人 await 它时才重新抛出；若始终无人 await，事件循环会在任务销毁时打印 “Task exception was never retrieved”。gather 默认会把第一个异常向外抛；wait 不会自动抛，需要检查 task.exception()。

Python 3.11 引入的 asyncio.TaskGroup 提供了结构化并发：组内任一任务失败，其余任务会被自动取消，异常以 ExceptionGroup 形式抛出，避免“孤儿任务”泄漏。

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(fetch("A", 1))
        tg.create_task(fetch("B", 2))
    # 退出时所有任务已完成；若有失败，抛出 ExceptionGroup

掌握以上机制后，就能在真实服务中安全地并发编排多个 I/O 操作：用 Task 启动、用 gather/wait 汇聚、用 timeout 兜底、用取消与异常传播保证资源不泄漏、错误可追踪。