启动与等待：Task、汇合、取消与超时

三个远程接口各耗 100ms：写成三行顺序 await，总时长约 300ms；若用 create_task 先 启动 再 gather 汇合，总时长约 100ms——差距不在「会不会 async def」，而在 启动与等待 是否拆开。异步工程里大半 bug 来自：该并发时串行、该 await 时 fire-and-forget、该 cancel 时硬等、或在 async 函数里藏 阻塞 调用。本篇按 Task 生命周期与常用模式展开。

顺序 await vs 并发 create_task ​

**顺序 **await****：子 协程 一个接一个跑完。

import asyncio


async def fetch(n: int) -> int:
    await asyncio.sleep(0.1)
    return n


async def sequential() -> None:
    a = await fetch(1)
    b = await fetch(2)
    c = await fetch(3)
    print(a, b, c)


asyncio.run(sequential())
# 约 0.3s 后打印 1 2 3

并发 启动：先 create_task，再统一 await。

import asyncio


async def fetch(n: int) -> int:
    await asyncio.sleep(0.1)
    return n


async def concurrent() -> None:
    t1 = asyncio.create_task(fetch(1))
    t2 = asyncio.create_task(fetch(2))
    t3 = asyncio.create_task(fetch(3))
    print(await t1, await t2, await t3)


asyncio.run(concurrent())
# 约 0.1s 后打印 1 2 3

create_task 立刻 把 协程 排进 事件循环；三个 sleep 重叠。await t1 只是 汇合 点——不是 启动 点。若写成 await fetch(1) 三次而没有 Task，就没有并发。

gather：一批 awaitable 的汇合 ​

asyncio.gather(*aws) 把多个 awaitable 并发驱动，返回结果列表（顺序与传入一致，与完成先后无关）：

import asyncio


async def tag(n: int) -> str:
    await asyncio.sleep(0.05 * n)
    return f"t{n}"


async def main() -> None:
    results = await asyncio.gather(tag(3), tag(1), tag(2))
    print(results)


asyncio.run(main())
# ['t3', 't1', 't2']

任一子任务未捕获异常时，gather 默认 立即 把异常抛给调用者（其它 Task 可能被 cancel，依版本与 loop 而定）。return_exceptions=True 时异常对象进结果列表，适合批量抓取失败：

import asyncio


async def maybe_fail(x: int) -> int:
    if x == 2:
        raise ValueError("bad")
    return x


async def main() -> None:
    out = await asyncio.gather(
        maybe_fail(1), maybe_fail(2), maybe_fail(3),
        return_exceptions=True,
    )
    print(out)


asyncio.run(main())
# [1, ValueError('bad'), 3]

gather 解决「等多路一起完成」；不等的时候别 gather——用 create_task 后台跑并在合适节点 await。

Task 生命周期与 cancel ​

import asyncio


async def job() -> str:
    try:
        await asyncio.sleep(1)
        return "ok"
    except asyncio.CancelledError:
        print("job: cancel received")
        raise


async def supervisor() -> None:
    t = asyncio.create_task(job(), name="worker-1")
    await asyncio.sleep(0.02)
    t.cancel()
    try:
        await t
    except asyncio.CancelledError:
        print("supervisor: task cancelled")


asyncio.run(supervisor())

cancel() 向 Task 注入 CancelledError，在 **下一个 await 处生效。try/finally 仍会执行；若 吞掉 CancelledError 且不 re-raise，Task 可能标成成功——这是常见 bug。

Task 状态同 Future：pending → done / cancelled。await task 在 cancelled 时抛 CancelledError。

清理模式：父 协程 退出前 cancel 子 Task 并 gather(..., return_exceptions=True)：

import asyncio


async def child(n: int) -> int:
    try:
        await asyncio.sleep(10)
        return n
    except asyncio.CancelledError:
        print(f"child {n} cleanup")
        raise


async def parent() -> None:
    tasks = [asyncio.create_task(child(i)) for i in range(3)]
    try:
        await asyncio.sleep(0.01)
    finally:
        for t in tasks:
            t.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)


asyncio.run(parent())

超时：wait_for 与 asyncio.timeout ​

asyncio.wait_for(aw, timeout) 在超时后 cancel 底层 Task 并抛 TimeoutError：

import asyncio


async def slow() -> str:
    await asyncio.sleep(10)
    return "never"


async def main() -> None:
    try:
        await asyncio.wait_for(slow(), timeout=0.05)
    except asyncio.TimeoutError:
        print("timed out")


asyncio.run(main())

超时后须确认底层 Task 已 done——若 协程 吞 cancel，会泄漏。3.11+ 推荐 asyncio.timeout 上下文：

import asyncio


async def work() -> None:
    async with asyncio.timeout(0.05):
        await asyncio.sleep(1)


async def main() -> None:
    try:
        await work()
    except TimeoutError:
        print("ctx timeout")


asyncio.run(main())

超时解决的是「不能无限占 调度 资源」；与 Semaphore 解决「不能无限 in-flight」互补。

Semaphore：有界并发 ​

asyncio.Semaphore(n) 最多允许 n 个 协程 同时持有：

import asyncio


async def limited(name: str, sem: asyncio.Semaphore) -> None:
    async with sem:
        print(f"{name} enter")
        await asyncio.sleep(0.1)
        print(f"{name} leave")


async def main() -> None:
    sem = asyncio.Semaphore(2)
    await asyncio.gather(*[limited(str(i), sem) for i in range(5)])


asyncio.run(main())

五份 Task 同时 启动，但同一时刻最多两个 enter——第三个 async with sem 会在 await 上 暂停 直到有人释放。Semaphore 保护下游连接数、API 配额，避免无界 gather 把远端打挂。

误用：async 里的同步阻塞 ​

import asyncio
import time


async def bad() -> None:
    time.sleep(0.5)


async def main() -> None:
    asyncio.create_task(bad())
    t0 = time.perf_counter()
    await asyncio.sleep(0.6)
    print(f"elapsed ~{time.perf_counter() - t0:.2f}s")


asyncio.run(main())
# elapsed ~0.60s —— bad 与 main 的 sleep 串在同一线程

应 await asyncio.sleep 或 to_thread。磁盘/网络同步 API 要么换 async 库，要么线程池。

误用：fire-and-forget Task ​

import asyncio


async def background() -> None:
    await asyncio.sleep(0.1)
    print("background done")


async def broken_main() -> None:
    asyncio.create_task(background())


asyncio.run(broken_main())
# 往往看不到 background done

create_task 后不 await、不 gather，asyncio.run 结束时会 cancel 未完成任务，异常也可能丢失。修复：

import asyncio


async def background() -> None:
    await asyncio.sleep(0.05)
    print("background done")


async def fixed_main() -> None:
    t = asyncio.create_task(background())
    await t


asyncio.run(fixed_main())

长期后台 Task 应保存引用，并在 shutdown 时汇合；或使用 TaskGroup（3.11+）。

TaskGroup：结构化并发 ​

import asyncio


async def worker(x: int) -> int:
    await asyncio.sleep(0.01)
    return x * 2


async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        a = tg.create_task(worker(1))
        b = tg.create_task(worker(2))
    print(a.result(), b.result())


asyncio.run(main())
# 2 4

退出 TaskGroup 时 等待 全部子 Task；任一失败 cancel 其余——减少 fire-and-forget 泄漏。

先启动、后汇合：读写分离场景 ​

读缓存（快）与拉远端（慢）可 并发 启动，再按业务 汇合：

import asyncio


async def read_cache() -> str | None:
    await asyncio.sleep(0.02)
    return None


async def fetch_remote() -> str:
    await asyncio.sleep(0.1)
    return "remote-data"


async def load() -> str:
    cache_task = asyncio.create_task(read_cache())
    remote_task = asyncio.create_task(fetch_remote())
    cached = await cache_task
    if cached is not None:
        remote_task.cancel()
        return cached
    return await remote_task


asyncio.run(load())
# remote-data

create_task 让慢路径与快路径 同时 进行；命中缓存时可 cancel 慢 Task 省资源。模式要点：启动 要早，等待 决策点要清晰。

模式速查 ​

与事件循环、协程的关系 ​

create_task 把 协程 交给 事件循环 调度；gather 在单 协程 里 await 多个 Future/Task；cancel 通过 loop 向 Task 注入异常；Semaphore 在 await acquire 处 暂停 多余 协程。整条 async 栈最终都落在 事件循环 的就绪队列与 I/O 等待上——模式只是帮你 启动 与 汇合 的惯用手势。

常见误解 ​

误解一：「gather 会创建线程。」 不会；仍是单线程 调度。

误解二：「cancel 之后协程立刻消失。」 要到 **下一个 await；finally 仍运行。

误解三：「Semaphore 代替 gather。」 Semaphore 限流；gather 汇合；常一起用。

误解四：「超时了就不用管 Task。」 须确认 cancel 生效，防泄漏。

asyncio.wait 与 as_completed ​

除 gather 外，asyncio.wait(aws, return_when=...) 返回 (done, pending) 集合，适合「任一完成就继续」：

import asyncio


async def slow(tag: str, t: float) -> str:
    await asyncio.sleep(t)
    return tag


async def main() -> None:
    tasks = {asyncio.create_task(slow("A", 0.1)),
             asyncio.create_task(slow("B", 0.05))}
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    print({t.result() for t in done})
    for t in pending:
        t.cancel()
    await asyncio.gather(*pending, return_exceptions=True)


asyncio.run(main())
# {'B'}

asyncio.as_completed 按完成顺序迭代 awaitable，适合流式处理结果；gather 则保持输入顺序。选型取决于你要 汇合 顺序还是完成先后。

shield：减少被父 cancel 牵连 ​

asyncio.shield(aw) 包一层 Future：外层 Task 被 cancel 时，内层可能继续（语义较细，易误用）。仅在对「子操作必须跑完」有强需求时考虑；多数场景显式 Task 生命周期更清晰。

重复踩坑清单（生产向） ​

1. 在 async 路由里调同步 ORM 阻塞 整个 worker 的 事件循环 → to_thread 或 async driver。
2. create_task 捕获异常却不 await → 用 add_done_callback 记录 task.exception() 或 gather。
3. 高并发无 Semaphore → 远端 429 或本地 fd 耗尽。
4. 超时后假设 Task 已停 → 仍 await 或 cancel 确认 done。
5. 把 gather 里异常全 return_exceptions=True 却从不检查列表里的异常对象 → 静默失败。

并发度与资源：Semaphore 再举一例 ​

模拟最多 2 个出站连接，5 个请求 gather：

import asyncio


async def request(i: int, sem: asyncio.Semaphore) -> int:
    async with sem:
        print(f"req {i} start")
        await asyncio.sleep(0.08)
        print(f"req {i} end")
        return i


async def main() -> None:
    sem = asyncio.Semaphore(2)
    rs = await asyncio.gather(*[request(i, sem) for i in range(5)])
    print(rs)


asyncio.run(main())

日志里同一时刻最多两个 start 未 end——Semaphore 在 async with 的 await acquire 处 暂停 多余 协程，与 事件循环 调度 配合，不创建新线程。

cancel 与 finally：清理仍会执行 ​

import asyncio


async def work() -> None:
    try:
        await asyncio.sleep(10)
    finally:
        print("cleanup always")


async def main() -> None:
    t = asyncio.create_task(work())
    await asyncio.sleep(0.01)
    t.cancel()
    try:
        await t
    except asyncio.CancelledError:
        print("caught cancel")


asyncio.run(main())
# cleanup always
# caught cancel

cancel 不是 kill -9；协程 仍有机会 finally。但若 finally 里再 await 长 I/O，要评估 cancel 语义是否被拖长。

顺序启动、统一 gather ​

import asyncio


async def unit(i: int) -> int:
    await asyncio.sleep(0.05)
    return i * 10


async def main() -> None:
    tasks = [asyncio.create_task(unit(i)) for i in range(4)]
    print(await asyncio.gather(*tasks))


asyncio.run(main())
# [0, 10, 20, 30]

列表推导里 create_task 启动 四个 unit；gather 汇合 结果。切忌写成 [await unit(i) for i in range(4)]——那是 列表推导 + 顺序 await，无并发。

超时 + gather 的组合 ​

import asyncio


async def flaky(i: int) -> int:
    await asyncio.sleep(0.1 * i)
    return i


async def main() -> None:
    tasks = [asyncio.create_task(flaky(i)) for i in range(1, 5)]
    try:
        await asyncio.wait_for(asyncio.gather(*tasks), timeout=0.15)
    except asyncio.TimeoutError:
        print("batch timeout")
        for t in tasks:
            t.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)


asyncio.run(main())

wait_for 包整个 gather；超时后须 cancel 子 Task 并 await 收尾，避免 pending 协程 泄漏。

限流 + gather 一起用 ​

import asyncio


async def call(i: int, sem: asyncio.Semaphore) -> int:
    async with sem:
        await asyncio.sleep(0.05)
        return i


async def main() -> None:
    sem = asyncio.Semaphore(3)
    rs = await asyncio.gather(*[call(i, sem) for i in range(10)])
    print(len(rs), sum(rs))


asyncio.run(main())
# 10 45

十个 Task 同时 create，但 Semaphore(3) 保证同一时刻最多三个 call 在 sleep——兼顾 并发 与保护下游。

启动时机：create_task 放在哪一行 ​

create_task 应尽早：在 await 慢操作 之前 创建其它 Task，才能重叠等待。常见反模式：先 await 慢路径 A，再 create_task(B)——B 的 启动 被 A 阻塞，失去并发。

import asyncio


async def slow(tag: str, t: float) -> str:
    await asyncio.sleep(t)
    return tag


async def wrong_order() -> None:
    a = await slow("A", 0.1)
    t = asyncio.create_task(slow("B", 0.1))
    b = await t
    print(a, b)


async def right_order() -> None:
    tb = asyncio.create_task(slow("B", 0.1))
    a = await slow("A", 0.1)
    b = await tb
    print(a, b)


asyncio.run(wrong_order())  # 约 0.2s
asyncio.run(right_order())  # 约 0.1s

gather 与 create_task 列表推导 ​

import asyncio


async def id(n: int) -> int:
    await asyncio.sleep(0.02)
    return n


async def main() -> None:
    # 正确：先 task 化再 gather
    tasks = [asyncio.create_task(id(i)) for i in range(5)]
    print(await asyncio.gather(*tasks))


asyncio.run(main())

**gather(coro1, coro2)** 也会内部 ensure_future，但显式 create_task 便于稍后 cancel、命名、add_done_callback。

cancel 传播：TaskGroup vs 手管 ​

TaskGroup 里子 Task 失败会 cancel 兄弟 Task——结构化 并发。手管 list 时须在 except / finally 自己 cancel 未完成任务，否则 pending 协程 泄漏到进程退出。

Semaphore 与 Lock 分工 ​

Semaphore(n) 限 并发度（最多 n 个同时 in-flight）。Lock 互斥（同一时刻一个 持有者）。限连接池用 Semaphore；保护共享可变结构用 Lock（或把状态推给单 Task 串行化）。

import asyncio


async def critical(lock: asyncio.Lock, i: int) -> None:
    async with lock:
        print("enter", i)
        await asyncio.sleep(0.01)
        print("leave", i)


async def main() -> None:
    lock = asyncio.Lock()
    await asyncio.gather(*[critical(lock, i) for i in range(3)])


asyncio.run(main())

wait_for 与底层 Task 生命周期 ​

wait_for(aw, timeout) 在超时 cancel aw 对应的 Task。若 aw 是 协程 而非 Task，wait_for 会先 ensure_future。超时后应用 try/except TimeoutError，并检查 Task 是否 done；必要时再 await 以消费 CancelledError。

误用：gather 里混 coroutine 与 Task ​

import asyncio


async def f(n: int) -> int:
    await asyncio.sleep(0.05)
    return n


async def main() -> None:
    t = asyncio.create_task(f(1))
    # gather 接受 mix；但 t 已启动，f(2) 由 gather 内部 task 化
    print(await asyncio.gather(t, f(2)))


asyncio.run(main())
# [1, 2]

理解：已 create_task 的 Task 与 裸 coroutine 混 gather 时，启动 时刻可能不同——排查并发 bug 时要画时间线。

生产 checklist ​

• 每个 create_task 是否有 汇合（await / gather / TaskGroup）？
• async 路径是否无 time.sleep / 同步 HTTP / 同步 DB？
• Outbound 是否有 Semaphore 或连接池上限？
• 超时路径是否 cancel 并 await 清理？
• 异常是否从 Task 传回（gather 或 task.exception()）？

四章串联回顾 ​

事件循环 轮转 ready / I/O / timer；Future pending → done 唤醒 await；协程 在 await 暂停；本篇 create_task 启动、gather 汇合、cancel / Semaphore 限流——构成日常 async 工程骨架。下一章 多线程与多进程 回答：何时不必坚持 asyncio，何时 GIL 逼你用进程。

asyncio.as_completed 按完成顺序消费 ​

import asyncio


async def job(n: int) -> int:
    await asyncio.sleep(0.01 * n)
    return n


async def main() -> None:
    tasks = [asyncio.create_task(job(i)) for i in (3, 1, 2)]
    for coro in asyncio.as_completed(tasks):
        print(await coro)


asyncio.run(main())
# 1, 2, 3 按完成先后

gather 保序；as_completed 保「谁先完谁先处理」——流式聚合、早停场景常用。

FIRST_COMPLETED 与早停 ​

import asyncio


async def probe(n: int) -> str:
    await asyncio.sleep(0.05 * n)
    return f"ok-{n}"


async def main() -> None:
    tasks = {asyncio.create_task(probe(i)) for i in (1, 3, 5)}
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    print({t.result() for t in done})
    for t in pending:
        t.cancel()
    await asyncio.gather(*pending, return_exceptions=True)


asyncio.run(main())
# {'ok-1'}

竞速多个镜像、多个 DNS 时常见 FIRST_COMPLETED + cancel 其余。

Lock 限临界区，Semaphore 限并行度 ​

别把 Semaphore 当 Lock 用（除非 n=1 且语义确为互斥）。连接池 Semaphore(10) 表示十个并发请求；保护共享 dict 用 Lock 或单 writer Task。

取消自己的 Task：CancelledError 必须再抛 ​

import asyncio


async def worker() -> None:
    try:
        await asyncio.sleep(100)
    except asyncio.CancelledError:
        print("cleanup")
        raise


async def main() -> None:
    t = asyncio.create_task(worker())
    await asyncio.sleep(0.01)
    t.cancel()
    try:
        await t
    except asyncio.CancelledError:
        print("main saw cancel")


asyncio.run(main())

except CancelledError: pass 会吞 cancel，Task 可能显示成功——框架代码才谨慎吞，业务 finally 清理后应 re-raise。

与 事件循环、Future、协程 的闭环 ​

create_task 绑定 协程 与 Task(Future)；gather await 多个 Future；cancel 经 loop 注入 CancelledError 在 协程 await 点生效；Semaphore 在 await acquire 暂停 协程 以保护资源。四篇机制在此合龙成 async 工程常用手势。

批量 cancel：gather return_exceptions ​

import asyncio


async def endless() -> None:
    try:
        await asyncio.sleep(999)
    except asyncio.CancelledError:
        print("stopped")
        raise


async def main() -> None:
    ts = [asyncio.create_task(endless()) for _ in range(3)]
    await asyncio.sleep(0.01)
    for t in ts:
        t.cancel()
    await asyncio.gather(*ts, return_exceptions=True)


asyncio.run(main())
# stopped × 3

return_exceptions=True 让 CancelledError 进结果列表而不中断 gather——批量收尾常用。

限流下载：Semaphore 实例共享 ​

同一 Semaphore 实例须跨 Task 共享；每请求 Semaphore(n) 新建则 限流 失效——这是 误用 高发点。

create_task 与 loop 关闭 ​

loop 关闭 后 create_task 拒收；shutdown 顺序：cancel Task → await gather → loop.close()。否则 RuntimeError: Event loop is closed。

模式选择表（扩展） ​

反模式：async 入口里调 sync main ​

import asyncio


def sync_main() -> None:
    asyncio.run(asyncio.sleep(0))  # 仅演示：sync 调 run 可以


async def async_main() -> None:
    # sync_main()  # 在 async 里调 sync 再 run → 嵌套 loop 风险
    await asyncio.sleep(0)

async 栈顶应 await，不要在 async 内再 asyncio.run 另一棵 协程 树。

四章合读后的实践顺序 ​

1. 画 事件循环（单线程 ready/I/O/timer）。
2. 标每个 I/O Future pending 点。
3. 写 协程 await 链。
4. 用 create_task/gather/Semaphore/cancel 组织 并发 与 汇合。

真实场景：并发拉取多 URL（结构） ​

import asyncio


async def fetch_one(i: int) -> dict:
    await asyncio.sleep(0.05)
    return {"id": i, "status": 200}


async def fetch_all(n: int, limit: int) -> list:
    sem = asyncio.Semaphore(limit)
    async def bounded(i: int) -> dict:
        async with sem:
            return await fetch_one(i)
    tasks = [asyncio.create_task(bounded(i)) for i in range(n)]
    return await asyncio.gather(*tasks)


async def main() -> None:
    rows = await fetch_all(8, 3)
    print(len(rows), rows[0])


asyncio.run(main())

create_task 启动 八条；Semaphore(3) 限同时三条 in-flight；gather 汇合 结果列表——三板斧齐用。

取消外层：TaskGroup 自动 cancel 兄弟 ​

import asyncio


async def ok() -> int:
    await asyncio.sleep(0.01)
    return 1


async def boom() -> int:
    await asyncio.sleep(0.01)
    raise ValueError("fail")


async def main() -> None:
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(ok())
            tg.create_task(boom())
    except* ValueError:
        print("group failed")


asyncio.run(main())

TaskGroup 失败 cancel 其它 Task——比手 gather 后 cancel 更不易漏。

超时层叠：connect 与 read 分别 wait_for ​

连接与读 body 应分别 超时——一个 wait_for(connect)、一个 wait_for(read)，避免 connect hang 拖死整 Task。模式上是 嵌套 await + wait_for，不是一个大 gather 包所有。

日志：记录 Task 名与 gather 耗时 ​

import asyncio
import time


async def step(tag: str) -> None:
    await asyncio.sleep(0.02)
    print(tag)


async def main() -> None:
    t0 = time.perf_counter()
    await asyncio.gather(
        asyncio.create_task(step("a"), name="step-a"),
        asyncio.create_task(step("b"), name="step-b"),
    )
    print(f"gather took {time.perf_counter()-t0:.3f}s")


asyncio.run(main())

name= 在 create_task 传（3.8+），trace 可读。gather 耗时约 max(子任务) 而非 sum——验证并发是否生效。

误用汇总（扩展） ​

• Semaphore 每任务新建 → 限流 无效。
• gather 内 顺序 coroutine 无 create_task → 仍可能并发（gather 内部 ensure_future），但 启动 时机与显式 create_task 不同，易误读 trace。
• shield 滥用 → cancel 语义混乱。
• wait_for 超时后不 await 底层 Task → 泄漏。
• sync HTTP 在 async def → 阻塞 事件循环 全局卡顿。

从模式回到机制 ​

create_task 把 协程 登记到 事件循环 ready；gather 在单 协程 内 await 多 Future；cancel 经 CancelledError 在 await 生效；Semaphore 在 await acquire 暂停。每模式对应机制章的一条线——模式不会脱离 loop、Future、协程 独立存在。

附录：并发模式决策树（文字） ​

1. 是否需要 同时等待 多个 I/O？否 → 单 await 链；是 → 2。
2. 是否需要 限制 同时 in-flight 数？是 → Semaphore + create_task；否 → 3。
3. 是否需要 任一完成即继续？是 → wait FIRST_COMPLETED / as_completed；否 → gather。
4. 是否需要 整体超时？是 → wait_for(gather(...)) 或 asyncio.timeout。
5. 是否需要 结构化 cancel？是 → TaskGroup；否 → 手 gather + finally cancel。

按树走，少选错 API。

顺序 vs 并发：同一业务两种写法对照 ​

顺序：总延迟 sum(t_i)，代码线性，易读，适合强依赖步骤（先 auth 再 fetch）。并发：总延迟 approx max(t_i)，适合独立 I/O（多资源 并行 拉取）。误判依赖 把可并发写成顺序，是性能 bug；误判独立 把必须顺序写成 gather，是逻辑 bug。

Semaphore 获取顺序与饥饿 ​

async with sem 按 await acquire 顺序排队；高负载下后到的 Task 可能 饥饿——若需公平，考虑 Queue 或 limit + shuffle（按业务）。多数 API 限流 Semaphore 足够。

cancel 与超时组合 ​

import asyncio


async def work() -> None:
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("work cancelled")
        raise


async def main() -> None:
    t = asyncio.create_task(work())
    try:
        await asyncio.wait_for(t, timeout=0.05)
    except asyncio.TimeoutError:
        print("wait_for timeout")
    assert t.done()


asyncio.run(main())

wait_for 超时 cancel Task；work 打印 cancelled；main 断言 done——完整 cancel 链。

fire-and-forget 的安全变体 ​

若必须 不 await（例如 后台 metrics flush）：

import asyncio


def fire_and_forget(coro) -> None:
    task = asyncio.create_task(coro)
    def _log(t: asyncio.Task) -> None:
        if t.cancelled():
            return
        if t.exception():
            print("bg error", t.exception())
    task.add_done_callback(_log)

强引用 Task（闭包或集合）+ done_callback 记异常——仍建议在 shutdown await 或 cancel。

四章实践项目 imaginary walkthrough ​

想象 async 爬虫：loop 上 N Task fetch；每 fetch await socket Future；Semaphore 限 并发；gather 汇合 HTML；wait_for 限单页超时；失败 cancel 整批 TaskGroup。事件循环 轮转、Future pending、协程 await、模式 create_task/gather——全用上。脑中跑通这条线，四篇文章就 毕业 了。

与多线程章的预告 ​

asyncio 解决不了 CPU 并行 与 sync-only 库的所有问题；下一章 多线程/多进程 讲 GIL 与何时 to_thread 仍不够。模式章的 Semaphore、gather 留在 单 loop 世界；跨 进程 要用 Queue、IPC，那是另一条线。

读者自查：模式是否真会选用 ​

给定「十个 URL、最多三并发、总超时五秒、任一失败取消其余」——应组合 Semaphore(3)、create_task、TaskGroup 或 gather、wait_for、cancel 哪些？在纸上画 Task 时间线再写代码，比直接 gather 少踩坑。

与前三章的交叉索引 ​

无 loop 则无 调度；无 Future done 则 await 永返；无 create_task 则无 并发 启动；无 Semaphore 则无 有界 并发。模式 是机制的组合拳。

压测时观察什么 ​

并发 连接 数上升时：CPU 是否单核打满（GIL+CPU work）？loop lag 是否上升（阻塞）？Task 数是否泄漏（fire-and-forget）？Semaphore 是否生效（下游 429）？指标比感觉可靠。

从模式到规范：团队可以约定的四条 ​

1. async 路径禁止 sync 阻塞 I/O（lint 或 code review）。
2. 每个 create_task 必须配对 汇合 或 documented fire-and-forget + callback 记错。
3. 出站调用默认 Semaphore + timeout。
4. cancel 路径必须 integration test（CancelledError + finally）。

规范把 模式 固化，减少 async 背锅。

最后一轮机制巩固：gather 内部直觉 ​

gather 可理解为：对每入参 ensure_future → 单 协程 内 await 等价于等 全部 Future done → 收集结果列表。任一 exception 默认 fail-fast（除非 return_exceptions）。因此 gather 不是「魔法并行」，是 Future 汇合 语法糖。

create_task 与 ensure_future 选型 ​

应用 代码 create_task 显式 启动；库 代码 ensure_future 兼容 coroutine/Future 入参。gather 内部用 ensure_future——你传 coroutine 或 Task 皆可，但 启动 时刻仍受 gather 调用时机约束。

模式章一句话总结四条 API ​

• create_task：现在就开始 调度。
• gather：等这批 都完 再往下。
• cancel：请 协程 在下一 await 退出。
• Semaphore：同时只准 N 个 进 临界 await。

四条覆盖多数 async 业务 并发 需求；其余 wait/as_completed/TaskGroup 是变体。

与 阻塞、事件循环 的再次提醒 ​

所有 模式 都假设 loop 线程未被 sync 阻塞。Semaphore 救不了 time.sleep；gather 救不了 sync HTTP。模式 + 非阻塞 I/O 才成立。

合并阅读建议 ​

模式 章假设你已理解 loop、Future、协程。create_task 无 loop 无意义；gather 无 Future done 永不返回；cancel 无 协程 await 点无效；Semaphore 无 await acquire 无法 限流。若某模式 不好使，先回前两章查 pending 与 阻塞，再怀疑 API 用错。

手写测试：gather 是否真并发 ​

用 time.perf_counter 包 sequential await 与 gather+create_task，确认后者约 max 单任务时长——实验比读文更能纠正「写了 async 就等于快」的错觉。

import asyncio
import time


async def t(d: float) -> None:
    await asyncio.sleep(d)


async def seq() -> None:
    t0 = time.perf_counter()
    await t(0.1)
    await t(0.1)
    print("seq", time.perf_counter() - t0)


async def par() -> None:
    t0 = time.perf_counter()
    await asyncio.gather(asyncio.create_task(t(0.1)), asyncio.create_task(t(0.1)))
    print("par", time.perf_counter() - t0)


asyncio.run(seq())
asyncio.run(par())
# seq ~0.2, par ~0.1

gather + create_task 通过计时验收 并发——建议本地跑一遍。

模式来自机制，不是反过来 ​

create_task、gather、cancel、Semaphore 不是四条独立咒语；分别是 启动调度、汇合 Future、注入 CancelledError、在 await 上排队。换语言仍会有类似模式；Python asyncio 只是标准库命名。懂机制 后查 文档 只是认名字。

结束语前的检查表 ​

• 何时 create_task 而非顺序 await？
• gather 与 wait FIRST_COMPLETED 选型？
• cancel 后为何要 await Task？
• Semaphore 实例是否共享？
• fire-and-forget 如何 记错？

五项打勾，模式 章可以收工，异步四章 闭环。

深度：TaskGroup 与 ExceptionGroup（3.11+） ​

TaskGroup 失败时抛 ExceptionGroup，可用 except* ValueError 匹配（PEP 654）。结构化 并发 的错误处理与 gather fail-fast 不同——多 Task 失败时 ExceptionGroup 可携带多个异常。cancel 兄弟 Task 仍是 TaskGroup 核心承诺。

深度：asyncio.run 与 Web 服务器 ​

Uvicorn 等 不 每请求 asyncio.run；请求 handler 是 已在 loop 上 的 协程。create_task/gather 写在 handler 里，与写在 asyncio.run(main()) 里 机制相同——区别只是 loop 生命周期更长。

模式章收束 ​

启动（create_task）、汇合（gather）、打断（cancel）、限流（Semaphore）——四个动词覆盖 async 业务 并发 的 80%；其余 wait、as_completed、TaskGroup、timeout 是 边界 与 结构化 增强。熟练四个动词 + 理解 loop/Future/协程，即可读大多数 asyncio 生产代码。

模式章节完整复述（一页纸） ​

create_task 尽早启动 子 协程；gather 汇合 多 Future 结果；cancel 在 await 点注入 CancelledError 且 finally 后 often re-raise；Semaphore 在 async with 限并发；避免 sync 阻塞 与无 汇合 的 fire-and-forget；timeout 后 await 确认 Task done。背下这段，写 async 服务时先过一遍 检查表 再提交 PR。

asyncio 四章如何复习 ​

按 23→24→25→26 顺序通读一遍，每章做文内 自查 问题；然后只读四篇 小结 与 一页纸复述，看能否在纸上画出 loop → Future pending → await 挂起 → set_result → 恢复 与 create_task/gather 时间线。画得出，asyncio 机制层 毕业。

并发错误案例复盘（虚构） ​

某服务在 async def handler 里 requests.get 调第三方 API——QPS 上百后 P99 飙升，heartbeat Task 停跳，诊断 time.sleep 式 阻塞 loop。修法：httpx.AsyncClient 或 await asyncio.to_thread(requests.get, ...)，并 Semaphore 限出站。create_task 与 gather 本身没问题，阻塞 才是根因。此类事故占 async 生产故障大多数。

取消与客户端断开 ​

HTTP 客户端断开时，ASGI 栈 cancel 请求 Task；你的 handler 应在 await 链上响应 CancelledError，finally 关连接。gather 的多子 Task 须 TaskGroup 或 finally cancel，否则 orphan Task 继续 await 远端。cancel 不是边缘情况，是在线服务 常态。

限流参数如何选取 ​

Semaphore 大小应来自下游 连接池上限、合同 QPS、或压测 拐点——不是 Semaphore(9999) 等于没 限流。timeout 应略大于下游 P99，小于客户端 总超时。create_task 数量可以大，Semaphore 控制 in-flight，gather 仍可 汇合 全部 Task（完成时间 spread 受 限流 影响）。

温故：模式相关的四个「先…后…」 ​

先 create_task，后 await —— 并发 启动 早于 汇合。先 Semaphore acquire，后 I/O —— 限流 在 await 点生效。先 cancel，后 gather return_exceptions —— 批量收尾 CancelledError。先 timeout，后确认 task.done —— 超时路径不 泄漏 Task。四个 先…后… 违反其一，就会出现「以为 并发 其实串行」「以为 cancel 其实还在跑」的经典 regression。

温故：四 API 与机制的一行映射 ​

create_task → loop ready；gather → 多 Future await；cancel → CancelledError @ await；Semaphore → await acquire 排队。一行映射方便 code review 时快速判断 PR 是否 踩 机制 红线。

收束 ​

工程上 async 的胜负手是 启动与等待是否配对、阻塞是否清零、cancel 与 timeout 是否收尾。create_task、gather、cancel、Semaphore 四词对应四件事；与 loop、Future、协程 三章合读，构成完整 asyncio 主线。

建议用 time.perf_counter 各跑一遍「三次顺序 await sleep(0.1)」与「三次 create_task + gather」，用数字确认 并发 是否真实发生——模式章最好的配套实验就是这两个计时对比。上线前对关键路径做一次 并发 vs 串行 计时，能提前发现「写了 gather 却仍串行」的配置错误。异步四章至此收束；后续 多线程与多进程 将讨论何时不必强上 asyncio。通读后应能画出一次 create_task → gather → cancel 的完整时间线。把四个 API 与时间线对齐，async Code Review 会快很多。建议对照 模式速查 与 决策树 两节，为当前项目各选一条 并发 路径并写进团队 wiki。能向同事解释为何用 gather 而非三次顺序 await，本章目标即达成。与 协程 章合读时，重点盯住 create_task 究竟在何时把 协程 送进 事件循环 的 ready 队列——并发 从这一行开始。以上模式均可在 CPython 3.11+ 本地用 time.perf_counter 做计时验证。通读 asyncio 四章后再接 Web 或爬虫项目，体会最深。机制优先，API 次之。本章至此收工，后续见系列第 27 篇。

小结 ​

create_task 启动；gather 汇合——先 启动 再 await 才能重叠 I/O。cancel 经 CancelledError 在 await 点生效。Semaphore 平衡 并发 与资源。避免同步 阻塞、只 create_task 不 await、超时后遗留 Task。异步四章至此串起 事件循环、Future、协程 与工程模式；下一章可转向多线程与多进程的分工。