先看一段代码。

1
2
3
async def worker():
    if counter is None:
        counter = await load_counter()

这是异步项目里到处都是的写法：如果还没加载过，就加载一次。逻辑上无懈可击：先检查，再使用，顺序清楚。在没有并发的世界里，这段代码是正确的。

但把它放进事件循环，跑 16 个并发请求，实测结果是：

16 个请求全都执行了一遍 load_counter()。

不是偶尔，是每一轮。我把这个场景跑了 100 轮，0 轮出现「只加载一次」的期望行为。

这就是异步环境下的条件竞争。它和线程模型里的竞争有相同的内核（检查与使用之间被撕开），但产生方式完全不同：这里没有抢占式调度，没有多核，只有一个线程。真正的竞争者是同一线程里的另一个协程，趁你 await 的时候插了进来。

一、一个反直觉的事实：单线程也会有竞争

asyncio 是单线程的。事件循环跑在一个线程里，所有协程都在这个线程上执行。既然只有一条执行流，为什么两个协程会互相干扰？

答案在 await。

把「有没有 await」做成唯一变量，看两个任务的执行顺序：

图一　await 就是抢占点

没有 await 的版本，输出顺序是：

1
2
3
4
A:进入
A:离开
B:进入
B:离开

A 的两步紧挨着，B 等 A 完全结束才开始。这段代码是原子的。

中间加一个 await asyncio.sleep(0) 之后：

1
2
3
4
A:进入
B:进入
A:离开
B:离开

B 整个插进了 A 的两步之间。

这段代码是确定性的，每次跑结果都一样：不是「有时候会交错」，是必然交错。原因在下一节。

关键在于：await 把控制权交回事件循环，事件循环立刻去跑其他就绪的任务。 于是 await 之前的代码和之后的代码，在调度意义上属于两个不同的时间片。任何依赖「await 之前看到的状态，在 await 之后仍然成立」的逻辑，都是错的。

二、竞争的本质：检查与使用之间被撕开一条缝

回到开头那段代码。把它拆开看：

1
2
3
if counter is None:          # 检查：现在还是 None
    counter = await load()   # ← await：控制权交出去
                             # 别的事务可能会在这期间把 counter 改掉

这就是 TOCTOU（Time-of-Check to Time-of-Use）：检查和实际使用之间隔着一个时间窗口，窗口里状态可能已经变了。

用一个更直观的例子量化它。一个只有 1 份份额的资源，8 个任务同时抢：

1
2
3
4
5
6
7
8
9
10
async def acquire_naive(res, who):
    if not res.busy:                 # 检查
        await asyncio.sleep(0)       # ← 缝在这里
        res.busy = True              # 占用
        res.owners.append(who)
        return True
    return False


await asyncio.gather(*(acquire_naive(res, i) for i in range(8)))

实测结果：

1
2
抢到的任务: [0, 1, 2, 3, 4, 5, 6, 7]
抢到的人数 = 8，资源却只有 1 份

八个任务全都认为自己拿到了。而且这个结果稳定得令人不安。重复 200 轮：

1
抢到 8 个: 200 轮 (100.0%)

100%，一次例外都没有。

为什么会这么稳定？因为 asyncio.gather 会让所有任务依次跑到它们的第一个挂起点。8 个任务逐个执行到 await asyncio.sleep(0)，此时没有任何一个执行过 res.busy = True，所以每一个都能通过前面的检查。等到它们被重新调度、继续往下走时，已经全部通过了。

这跟「小概率的时序巧合」无关，这段代码在并发下的必然行为。加上 asyncio.Lock 之后，同一个场景跑 200 轮全部只抢到 1 个。

为什么代码审查发现不了

因为把这段代码单看一遍，它的逻辑是正确的。检查、使用，顺序分明。问题不在逻辑，在于两个语句之间隔着一次调度，而这件事在代码表面上看不出来：await asyncio.sleep(0) 看起来像「什么都不做」。

这也是异步竞争和线程竞争最大的体感差异。线程版本里，你看得见 threading.Lock 或者没有它；异步版本里，一个无辜的 await 就是问题的全部。

三、await 到底在哪里切走

要判断代码有没有竞争，得知道 await 什么时候会让出控制权。这件事比看起来复杂。

事件循环的调度模型大致是这样：每个协程被包装成一个 Task，Task 记录它挂起在哪个 future 上。当 await 的对象还没完成，Task 就把自己挂到该 future 的回调上，然后把控制权还给事件循环；将来 future 完成，回调把 Task 重新放进就绪队列。

所以会不让出的情况是存在的：如果 await 的对象已经完成，Task 根本不会挂起，直接继续往下走。

这一点很要命，值得单独验证。同一个函数，只把 await 的对象从「未完成」换成「已完成」：

1
2
3
4
5
6
7
8
9
10
11
12
13
await 一个「已完成」的 future:
    1. A:进入
    2. A:离开
    3. B:进入
    4. B:离开
    → A 的两步保持原子，和没有 await 时一样

await 一个「未完成」的 future:
    1. A:进入
    2. B:进入
    3. B:离开
    4. A:离开
    → 被撕开了

两行看起来一模一样的 await，对调度的影响完全不同。

这意味着「这段代码有没有竞争」取决于运行时状态，不取决于语法。 同样一行 await cache.get(key)，缓存命中时它不会挂起，代码是安全的；缓存未命中时它会挂起，代码就有竞争。

四、为什么这类 bug 能躲过测试

第三节那个发现解释了开头那个实验为什么能躲过所有测试。

构造一个可以先热后冷的场景：一个「如果没初始化过就初始化一次」的处理器，中间那次 await 是会命中缓存还是会真的去 I/O，直接决定了检查与占用之间有没有缝。

两种情况各跑 100 轮：

缓存状态 初始化实际执行次数 只执行 1 次的轮数
热（命中，await 不挂起） 1.0 次 100 / 100
冷（未命中，await 挂起） 16.0 次 0 / 100

热缓存下，这段代码的表现是完美的，100 轮全部正确。冷缓存下，100 轮全部出错。

这就对上了生产环境里那种最难查的问题：

  • 单元测试跑得通。 测试数据是预热的，或者测试里根本没有并发。
  • 功能验证跑得通。 手动点几下，一次一个请求，永远是热的。
  • 压测也跑得通。 压测机反复打同一批请求，缓存一直是热的。
  • 上线之后偶发。 冷启动、缓存过期、缓存被击穿、扩容后新实例首次接流量，这些时刻缓存是冷的。

于是现象表现为「偶发」「随机」「重启后就好」，而代码在测试环境里从来没有复现过。

这不是运气问题，是测试环境和生产环境的运行时状态不同。 同一份代码，两份完全不同的成绩单。

五、常见的解决方案，以及它们各自解决的是什么

知道了竞争从哪来，解决方向就清楚了：让「检查」和「使用」之间不存在能让其他任务插进来的窗口。

难的是这句话有很多种听起来合理的实现方式，而其中大部分是无效的。我把常见的八种放进同一个场景里跑，结果如下。

场景是 10 个任务并发地把一个共享计数器加 200 次，正确答案 2000：

图二　八种看起来像加锁的写法

保护方式 最终值 判定
裸的读-改-写（中间有 await） 200 丢失 90%
读-改-写之间没有 await 2000 结果正确
只给写加锁 200 丢失 90%
手写布尔锁 200 丢失 90%
asyncio.Lock 2000 正确
asyncio.Semaphore(1) 2000 正确
asyncio.Queue 串行化 2000 正确
不并发，顺序执行 2000 结果正确

丢失 90% 的意思是：2000 次自增只生效了 200 次。因为 10 个任务每次都读到同一个值，然后都写回同一个值，一轮下来只有一次自增活了下来。

八种写法里有三种是真的有效。另外几种的问题值得逐个说。

asyncio.Lock：把临界区圈出来

这是最直接的答案。它的语义和线程锁一样：同一时刻只有一个持有者，但等待时会让出事件循环（这是它和 threading.Lock 的关键差别，后者会阻塞线程）。

1
2
3
4
async with lock:
    if not res.busy:
        await asyncio.sleep(0)      # 临界区内照样可以 await
        res.busy = True             # 锁保证了这一段不会被其他任务插入

这里有个容易误解的地方：临界区里可以有 await。锁要保证的是其他持有同一把锁的任务进不来，跟「能不能挂起」没有关系。所以上面这段是正确的：即使中途让出了控制权，别的任务也进不来，因为它们在等锁。

asyncio.Semaphore(1)：等价的替代品

Semaphore(1) 的行为和互斥锁一致。它的意义在于可以推广到 Semaphore(n)：限制同时进入某个区域的数量而不是互斥。比如给下游加一个并发上限，就用它而不是锁。

asyncio.Queue：换一种思路，不共享状态

队列方案不保护共享变量，而是取消共享：把更新请求放进队列，只由一个消费者去改。这样并发的问题被转化成「消息的传递顺序」，而队列自己保证了顺序。

它适合命令式的更新场景（比如写日志、累加度量）。代价是所有更新变成异步的，而且要考虑队列满了怎么办、消费者挂了怎么办。

那两种「结果正确但不算安全」的写法

「读-改-写之间没有 await」和「顺序执行」都得到了正确答案，但它们不算解决方案：

没有 await 的读-改-写在单线程 asyncio 里是原子的，因为不存在抢占：事件循环只在 await 点切换。但这是个脆弱的不变量：任何人往这两行之间插一个 await（哪怕是调试用的日志、一次属性访问触发的异步方法），这段代码立刻就是错的。而代码审查时，新加的那个 await 看起来完全无害。

实际上这也解释了一个常见的误解：很多人以为「asyncio 是单线程，所以不用考虑并发」。正确的说法是「单线程让没有 await 的代码段天然原子」，而现实中的业务代码几乎总会有 await。

三种方案保护的东西不一样

把它们放在一起看，会发现它们解决的不是同一个问题：

方案 保护的对象 适合的场景
Lock 一段代码（临界区） 检查与修改必须整体完成
Semaphore 一个数量配额 限制并发数，不需要互斥
Queue 消息的顺序 用传递替代共享

选错了不会报错，只会在压力下出问题。比如用 Semaphore(1) 去做互斥是对的，但用 Semaphore(2) 去保护一个「只能有一个持有者」的资源就错了，语法上完全一样。

六、锁的常见误用

上面三种无效的写法可以归纳成几类反复出现的错误。

6.1 锁的范围圈错了

最典型的是「只给写加锁」：

1
2
3
4
5
6
async def add(c, n, lock):
    for _ in range(n):
        tmp = c.value            # 读：没锁保护
        await asyncio.sleep(0)   # ← 窗口
        async with lock:
            c.value = tmp + 1    # 写：有锁保护

看起来加了锁，实际上读和写还是被撕开了。锁保护了写，但写的是那个已经过期的 tmp。实测同样是丢失 90%。

判断标准很简单：从「读」到「写」之间的所有语句，都必须落在同一段临界区里。 中间只要有一个能走到临界区外面的 await，这段保护就是假的。

6.2 手写锁：把锁本身也写成了非原子的

1
2
3
4
5
while c.flag:              # 等锁
    await asyncio.sleep(0)
if not c.flag:             # 检查
    await asyncio.sleep(0) # ← 缝在这里
    c.flag = True          # 上锁

手写锁的问题在于它自己也需要原子性。检查和置位之间只要有一个 await，锁就形同虚设，两个任务会同时认为自己拿到了锁。实测同样是丢失 90%。

这条同时也是「为什么不要自己实现同步原语」的最直接理由。asyncio.Lock 内部用 future 和回调实现，不依赖任何挂起点来保证原子性。

6.3 用 await 分割临界区

1
2
3
4
5
6
async with lock:
    check()          # 第一次检查
# 锁在这里释放了
await something()    # 做了点别的事
async with lock:
    act()            # 再操作

两次加锁之间，状态可能已经被改过。「加锁」不等于「整个逻辑是原子的」，锁的有效范围严格等于临界区的范围。

6.4 死锁：不一致的加锁顺序

两把锁，两个任务以相反的顺序获取：

1
2
3
4
5
6
7
8
9
async def task_a():
    async with lock_x:
        await asyncio.sleep(0.01)     # 给 task_b 机会拿到 lock_y
        async with lock_y: ...

async def task_b():
    async with lock_y:
        await asyncio.sleep(0.01)
        async with lock_x: ...

两个任务各自持有一把锁，等待对方释放。这和线程模型的死锁是同一个问题，解决办法也一样：给所有锁定义全局顺序，任何地方都按同一个顺序获取。

异步特有的地方在于，死锁时事件循环本身还活着，它继续调度其他任务，只是这两个任务永远停在那里。所以异步死锁的排查和线程死锁不同：线程死锁往往表现为进程卡住，异步死锁表现为「这两个请求永远不返回，但服务整体没事」。用 asyncio.wait_for 给自己的操作加一个超时，是发现这类问题最便宜的手段。

6.5 锁的作用域搞错

asyncio.Lock() 必须在事件循环里创建，而且必须是同一把才能互斥。两个常见错误：

  • 在每次请求处理里 Lock() 一个新对象，每个请求一把锁，等于没锁。
  • 锁定义成了模块级全局，但服务跑在多个事件循环里（多进程、多次 asyncio.run），每个循环一把锁，同样等于没锁。

第一种错误在测试里特别难发现，因为单请求测试下「每个请求一把锁」和「全局一把锁」表现一样。

七、异步特有的风险

前面几节的问题在别的并发模型里都有对应物：TOCTOU、锁范围、死锁，线程编程里同样存在。这一节讲两类异步独有的风险。

7.1 取消安全：协程可以在任何一个 await 点被取消

线程模型里，你不能在任意一条指令处「取消」一个线程。所以部分完成的状态是显式的：你要么在线程里检查取消标志，要么接受它跑完。协程不一样，task.cancel() 可以在任何一个挂起点生效。

后果是两阶段操作会在中间停住。一个转账的例子：

1
2
3
4
async def transfer(src, dst, amount):
    src.balance -= amount        # 第一阶段：扣
    await asyncio.sleep(0.05)    # 模拟写账本 / 通知下游
    dst.balance += amount        # 第二阶段：入

在扣款之后、入账之前的这 50 毫秒里取消它，实测结果：

1
不加保护:    A=0  B=0    总量 100 -> 0    钱不见了

一百块凭空消失。用 asyncio.shield 保护之后：

1
shield 保护:  A=0  B=100  总量仍是 100

图三　取消安全：两条时间线

这里必须说清 shield 到底做了什么，因为它太容易被误解成「让取消不发生」：

shield 不阻止外层任务被取消。 外层照样收到 CancelledError，调用方照样要处理取消。它保证的是：被它包住的那个内层任务不会被这次取消打断，会继续跑到结束。

所以正确的用法是把内层任务的引用留住，让取消之后仍然能等它完成：

1
2
3
4
5
6
inner = asyncio.create_task(body())
try:
    return await asyncio.shield(inner)
except asyncio.CancelledError:
    # 外层被取消了，但 inner 还在跑
    raise

7.2 finally 里的清理，本身也会被取消

这是更隐蔽的一层。清理代码通常写在 finally 里，但 finally 里的 await 同样是一个挂起点：

1
2
3
4
5
try:
    await do_work()
finally:
    await asyncio.sleep(0.08)    # 清理本身是一个 await
    conn.closed = True

“只取消一次”和“在清理过程中再取消一次”的结果不同：

1
2
只取消一次:              清理日志: ['进入清理', '清理完成']   连接关掉了吗: True
在清理过程中再取消一次:    清理日志: ['进入清理']             连接关掉了吗: False

第二次取消打在清理过程的中间，清理被中断，连接没有关掉。

现实中的触发场景是：超时先取消一次，调用方（或者服务器的优雅退出流程）过一会儿又取消一次。清理逻辑必须假定自己随时会再被打断，所以关键清理动作要放在不依赖 await 的位置，或者用 shield 保护，或者干脆接受「清理可能没完成，需要另有一层兜底回收」。

7.3 CancelledError 的继承关系

一条看似不相关的细节，在实际项目里会造成很难查的问题：

1
2
CancelledError 的基类链: CancelledError -> BaseException -> object
是 Exception 的子类吗: False

asyncio.CancelledError 从 Python 3.8 起继承自 BaseException 而不是 Exception。实测三种捕获方式的结果：

捕获方式 结果
except Exception 取消正常传播，任务不会假装成功
except BaseException 被吞掉，协程返回了看似正常的结果
裸 except: 被吞掉，同上

写成 except Exception 的代码是安全的，取消能穿透它继续向上传播。而为了「捕获所有异常」而写成 except BaseException 或裸 except，会把取消也吃掉，于是一个已经被取消的协程返回了一个看起来成功的结果，调用方毫不知情。上层以为操作完成了，实际上它被中断在了中间。

这条规则很少被注意到，因为它在绝大多数时候不影响功能，只有在取消发生时才有区别，而取消恰好是最难复现的场景。

7.4 孤儿任务：比请求活得久的工作

asyncio.create_task() 会启动一个独立的执行流。如果不等待它、也不登记它，它会一直跑下去，而且上层完全看不到：

1
2
3
async def handle_request():
    asyncio.create_task(background_writer())   # 不等它，也没登记它
    return "已响应"

图四　任务比它的请求活得久

实测结果：

  • 请求正常返回
  • 请求返回之后，那个后台任务又写了 15 次状态
  • 结束时还有 1 个任务存活

问题在于这个任务已经脱离了请求的生命周期。请求的超时管不到它，请求的取消管不到它，请求结束时释放的资源它还在用。如果它写的是共享状态，就同时变成了一个竞争源。

asyncio.TaskGroup（Python 3.11+）是这个问题的主流解法：

1
2
async with asyncio.TaskGroup() as tg:
    tg.create_task(background_writer())

它保证退出这个 async with 块时，组内所有任务都已结束；任何一个抛异常，整组会被取消并把异常传播出来。

7.5 异常静默

create_task 的异常如果没人取，不会抛出来，只会记在 task 对象上。实测输出：

1
任务已完成=True  取到的异常=RuntimeError('后台任务 失败了')

以及在对象被回收时打一行警告：

1
2
Task exception was never retrieved
future: <Task finished ... exception=ValueError('没人管的异常')>

这行警告很容易被日志淹没。而 TaskGroup 会把异常传播到父任务，不需要任何人记得去取它。

一个必须交代的证据边界

官方文档里还有一条警告：「必须保存返回值的引用，否则任务可能中途消失」。我在 Python 3.14 上专门做了探测，没能复现这个现象：

1
2
探针1 丢弃引用后，task 对象还活着吗: True
探针2 运行中丢弃引用，task 还活着吗: True

两种情况下任务都正常跑完了。可能的原因是 CPython 在使用引用计数，而事件循环在处理过程中仍持有引用。

所以这一条我保持原样陈述、不加渲染：文档要求保存引用，我遵照执行，但在这个版本上没有观测到不保存引用导致的失败。把它当成必须遵守的约定，而不是一个我验证过的现象。

八、怎么把这类问题找出来

前面所有实验都有一个共同点：它们是设计出来的，不是碰巧遇到的。

真实的竞争没这么友好。前面几节展示的是「构造好场景，跑 100 轮，看到 100% 失败」。而生产环境里拿到的是「线上偶发，日志里什么都没有」。

差别在哪？在到达时序。

竞争检出率是一个悬崖，不是斜坡

把「相邻请求的到达间隔」做成唯一变量，其他都不动（同一个有缺陷的实现，同一份负载，每个档位跑 150 轮）：

图五　竞争检出率随到达间隔的变化

到达间隔 不加锁的违反率 加锁的违反率
完全同时（0 ms） 100% 0%
0.2 ms 100% 0%
0.5 ms 100% 0%
1.0 ms 1.3% 0%
2.0 ms 0% 0%
8.0 ms 0% 0%

从 100% 掉到 0%，全部发生在 0.5 毫秒到 1 毫秒这半个毫秒的区间里。中间没有过渡带。

这个形态解释了为什么这类 bug 的定位如此困难：

  • 手工测试里请求是一个一个来的（间隔几十到几百毫秒）→ 永远落在悬崖右边，永远看不到问题
  • 单元测试里通常没有并发，或者用 gather 让任务同时启动 → 落在悬崖左边，要么全错要么全对
  • 压测里请求是同时发的 → 落在悬崖左边，但压测的断言通常是「成功率」「响应时间」，而竞争的后果往往是不变量被破坏（超卖、重复初始化、计数偏差），不一定表现为请求失败

所以「测试通过了」和「代码没问题」之间没有必然联系。 测试通过只说明测试的到达时序落在了悬崖右边。

这个悬崖的位置取决于两个量之比：到达间隔，以及「检查」到「使用」之间的那段时间。后者就是 await 真正挂起的时长。任何让这个比值变小的变化，都会把代码从悬崖右边推到左边，依赖变快、缓存变冷、并发度变高、机器变慢，都会。

一个负结果：放大 await 点并不能提高检出率

既然窗口是问题的核心，一个自然的想法是：把 await 点放大。每个零延迟让出放大成 N 次，把窗口撑大 N 倍，那些潜伏的竞争不就暴露了吗？

我实现了一个这样的工具：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class YieldStress:
    """把零延迟的 await asyncio.sleep(0) 放大成 extra 次让出"""

    def __init__(self, extra=0):
        self.extra = extra
        self._orig = None

    def __enter__(self):
        if self.extra <= 0:
            return self
        orig = asyncio.sleep
        extra = self.extra

        async def stressed(delay, result=None):
            r = await orig(delay, result)
            # 只放大零延迟的调用。如果连 sleep(0.001) 这种真实延时也放大，
            # 会把本该重叠的任务推散，反而掩盖竞争。
            if not delay:
                for _ in range(extra):
                    await orig(0)
            return r

        self._orig = orig
        asyncio.sleep = stressed
        return self

    def __exit__(self, *exc):
        if self._orig is not None:
            asyncio.sleep = self._orig
        return False

想法看起来合理。但实测结果是：

到达间隔 放大 0 档 放大 2 档 放大 8 档
1.0 ms 1 / 150 1 / 150 0 / 150
2.0 ms 1 / 150 0 / 150 1 / 150

没有可观测的改善。

原因想清楚之后很直白：竞争窗口一旦让出就是无界的。 一个任务在窗口里挂起之后，它会一直躺在那里，直到被重新调度。窗口的宽度取决于「另一个任务还需要多久才走到它的窗口」，远不止一个时间片。撑宽一个本来就已经无界的窗口，不改变「有没有第二个任务同时落在里面」。

真正的约束是到达的重叠程度，而这由任务的启动时序决定，不是由 await 的个数决定。

所以这个工具的正确定位是回归闸门，不是发现工具：改动之后用放大档位跑一遍，确认原本正确的逻辑没有退化。不要指望它替你找出本来就因为时序原因测不到的竞争。

一个工具的价值不只在于它能做什么，也在于知道它不能做什么。这一条我原本打算写成「用放大 await 来暴露潜伏竞争」的正向结论，实测推翻了，就按推翻的写。

那实际应该怎么做

结合前面的实验，可操作的做法是这几条：

  1. 把不变量写出来，在测试里断言它。 「卖出的不超过库存」「初始化只执行一次」「转账前后总量守恒」，这些都能写成断言。上面所有实验都是靠不变量断言检出问题的，不是靠「性能变差了」。

  2. 用同时到达的方式测。 手工顺序测试和 gather 是完全不同的两个世界。gather 让所有任务跑到第一个挂起点，这恰好是竞争最容易发生的调度。

  3. 把到达间隔当成一个扫描维度，不要固定成一个值。 前面那张表就是这么来的。如果代码在 0.5 ms 间隔下 100% 出错、在 2 ms 间隔下一次都不错，那么线上只要出现一次高并发重合，它就会出错。

  4. 审查时找「检查与使用之间有没有 await」。 这是唯一能在读代码阶段发现问题的办法。具体的模式有：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
if cond:                 # 检查
    await something()    # 中间有 await
    mutate()             # 使用 —— 此时 cond 可能已经不成立

value = obj.attr         # 读
await something()
obj.attr = value + 1     # 写回 —— 丢失更新

if name not in cache:    # 检查
    await load()         # 中间有 await
    cache[name] = data

while flag:              # 等条件
    await asyncio.sleep(0)
flag = True              # 置位 —— 手写锁

这四种模式覆盖了实践中绝大多数异步竞争。

九、对照：线程模型下的同一批问题

把两种模型放在一起，差异会更清楚。

线程 / 多进程 协程（asyncio）
抢占点 任意指令（解释器层面 + 操作系统） 只在 await 处
原子性的来源 需要锁或原子操作保证 没有 await 的代码段天然原子
并发单元 OS 线程，由内核调度 协程，由事件循环调度
等待的代价 线程被阻塞，占着栈 只挂起协程，线程继续跑别的
取消 不能安全取消，需协作式标志 可在任意挂起点取消
异常传播 线程内异常不会传到创建者 create_task 不取就丢，TaskGroup 才传播
死锁的表现 进程常卡住 事件循环仍活着，只有相关任务停住
竞争的可见性 有 GIL 时简单操作看似安全，实为错觉 没有 await 时确实原子，但极易被改动破坏
典型误判 「有 GIL 就不会有竞争」 「单线程就不会有竞争」

上表最后一行值得展开。两种模型都有一个流行的错误认知：

  • 线程模型：「CPython 有 GIL，所以我的代码是线程安全的。」
  • 协程模型：「asyncio 是单线程的，所以我的代码不会有竞争。」

两个都是把「某些操作恰好原子」错当成「所有操作都原子」。 GIL 保证的是字节码层面的原子性，一次 counter += 1 涉及多条字节码，中间可以被切换；asyncio 保证的是「没有 await 的代码段不会被插入」，一次 counter += 1 如果中间没有 await 确实安全，但只要上下文一变（比如 counter 是个需要 await 才能读的属性）就失效。

两者的共同教训是：原子性的边界要靠明确的机制来划定，不能靠「这个环境应该没事」。

十、结论

回到开头那个问题：asyncio 只有一个线程，为什么会有条件竞争？

因为并发不等于并行。竞争的根源是「同一份状态被多个执行流交错访问」，不是「多个 CPU 同时执行」。协程通过 await 主动让出控制权，制造出的交错和线程被抢占造成的交错，在效果上没有区别。

所以读代码找竞争，最有效的办法是找 await：检查和使用之间只要夹着它，这两步就不再原子。但后半句得说清楚：它是不是窗口，取决于运行时。同一个 await，等到的对象已完成时不挂起，未完成时挂起。冷启动、缓存击穿、扩容后新实例的首批流量，改变的都是这件事。这就是为什么「测试通过」证明不了代码没竞争，只能证明测试的时序落在了安全的一侧。

「看起来像加锁」离「真的互斥」也很远。八种写法里只有三种有效，失效的那几种在代码表面上看不出任何问题。唯一的判断依据是：从读到写之间的所有语句，是不是都落在同一段临界区里。

异步还多一类线程模型里没有的风险。协程可以在任何一个挂起点被取消，两阶段操作会停在中间；finally 里的清理本身也会被打断；裸的 except 会把取消吞掉，让调用方以为操作成功了。这里面取消安全最被低估。

最后一件是关于怎么测的。竞争检出率随到达间隔呈悬崖式变化，手工顺序测试永远落在安全的一侧。而放大 await 点这个看起来很合理的技巧，实测对检出率没有帮助，因为窗口本来就是无界的，要改的是到达的同时性。

解法落在「把检查和使用的整体完成」这个要求上，落实成一段边界明确的临界区。加一把锁只是实现这个边界的一种手段。审查代码时真正该问的那句话很短：

这两个操作之间有 await 吗？

附录 A：实验环境

1
2
3
4
5
6
7
操作系统        Arch Linux
内核            Linux 7.2.7-arch1-1
虚拟化          KVM（QEMU）
CPU             Intel Core i5-4590T @ 2.00GHz，4 核
Python          3.14.7
GIL             启用
asyncio         标准库，TaskGroup / Barrier / timeout 均可用

关于结果的可复现性。 本文的实验是单线程协程内的调度问题，不依赖多核性能、内存带宽或系统负载，所以数字比一般的性能实验要稳定得多。实测中，同一配置重复跑 200 轮，违反不变式的轮数在所有轮次里都是一致的（要么 200，要么 0），没有出现过中间值。第二节那个「8 个任务全部抢到」的场景跑了 200 轮，全部是 8。

唯一的例外是第八节那张到达间隔表，在悬崖附近（1.0 ms）会出现 1/150 这样的零星命中，这是时序落在边界上的正常表现，也正是「悬崖不是绝对」的体现。

附录 B：复现步骤

全部代码只用标准库，不需要安装任何依赖：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
mkdir asyncrace && cd asyncrace

# 1. 基础机制：await 是抢占点
#    输出：没有 await 时 A 的两步相邻，有 await 时被 B 插入
python exp1_yieldpoint.py

# 2. 检查与使用被撕开：8 个任务全部抢到同一份资源
#    以及加锁后的对照
python exp2_checkthenact.py

# 3. await 一个已完成 vs 未完成的 future，调度结果不同
python exp3_await_semantics.py

# 4. 缓存命中决定有没有竞争：热 100/100 正确，冷 0/100 正确
python exp4_cache_hides_race.py

# 5. 八种保护方式的对照表
python exp5_lock_misuse.py

# 6. 取消安全：shield、二次取消、异常吞噬
python exp6_cancel_safety.py

# 7. 孤儿任务与异常静默
python exp7_task_leak.py

# 8. 竞争检测工具：到达间隔扫描 + 放大 await 的负结果
python racecheck.py

复现时最值得自己动手改的三处：把 exp2 里的并发数从 8 改成 1000，看「全部抢到」的现象是否依然稳定；把 exp5 的 TASKS 从 10 改成 2，观察丢失比例如何随并发度变化；把 racecheck.py 里的错开步长改成 0.0008，自己找一次那个悬崖的位置。

参考资料

  1. Python 官方文档：asyncio — Asynchronous I/O：Task、create_task、TaskGroup、shield 的正式语义，以及「保存任务引用」的明确要求。

  2. Python 官方文档：Coroutines and Tasks：await 挂起与恢复的机制说明，协程与 Task 的关系。

  3. Python 官方文档：Exceptions：BaseException 与 Exception 的继承关系，CancelledError 自 3.8 起从 BaseException 继承的原因。

  4. PEP 492 — Coroutines with async and await syntax：await 语义的原始定义。

  5. Python 官方文档：asyncio.Synchronization Primitives：Lock、Semaphore、Event、Condition 的行为与 threading 对应物的差别。

  6. PEP 654 — Exception Groups and except*：TaskGroup 异常传播所依赖的语言特性。

文中所有数据均为本机实测，七个实验脚本与检测工具的源码在附录 B 中列出，可自行运行复现。