Python asyncio同步原语与异步生成器

可视化演示

展示asyncio中Lock、Semaphore、Event等同步原语的协程调度过程，交互调节并发数查看锁竞争与信号量控制效果。

并发协作困境

后厨再忙也得有规矩：谁先上灶、谁让一让，全靠默契。asyncio 也一样，单线程加一个事件循环，协程之间的“默契”就写在协作式调度里。事件循环是总调度员，它把 CPU 交给某个协程执行，直到这个协程主动 await——await 就相当于“让出炒勺”，调度器趁机把活交给下一个协程。这套机制让大量 IO 等待不再空转，但也埋下了一个隐患：协程之间可以随时被打断、再恢复。

麻烦在于，默契挡不住“手滑”。两个协程同时想去改同一个变量时，事情就失控了。比如一条 x += 1 看起来是一句话，实际上要分三步：先读取旧值，然后计算，再写回新值。如果协程 A 在“读完旧值、还没写回”的间隙遇到 await 让出了控制权，协程 B 插进来也读了同一个旧值，两人各自加一写回，结果只加了一次。这就是竞态条件（Race Condition）：不是“同时发生”，而是“交错发生”造成的丢失更新。

$$\displaystyle x \leftarrow x+1 \ \text{分三步:}\quad v \leftarrow x;\ \ \text{await 让出};\ \ x \leftarrow v+1$$

1. 读取旧值：协程 A 把共享变量的当前值读进自己的临时变量

$$v_A \leftarrow x$$

2. 让出控制权：遇到 await（比如 sleep 或网络请求），事件循环切换到协程 B

$$\text{await} \Rightarrow \text{调度器切到协程 B}$$

3. 写回新值：协程 B 也读到同样的旧值并加一写回，A 再写回时把 B 的结果覆盖

$$x \leftarrow v_A + 1 \Rightarrow \text{丢失一次} +1$$

4. 最终结果错误：三次 x += 1 后，x 最终只增加了 2

$$x_{\text{final}} = x_0 + 2 \neq x_0 + 3$$

后厨里，这就是“两个帮工同时往一道菜里加盐”的灾难：只要在“尝咸淡”和“落盐勺”之间被别的事打断，菜就毁了。所以后厨需要“挂牌”和“排队”这类规矩，而 asyncio 的同步原语，正是把规矩变成代码的工具。

Lock互斥锁

灶台挂牌：哪位大厨把“占用”牌挂上，其他人就得在门口等着，直到他做完菜、摘下牌子走出来。asyncio.Lock 就是这块牌子，它解决的问题非常朴素——把“读、改、写”这三步牢牢包成一个不许插队的整体，也就是临界区。

使用 Lock 的方式很简单：进临界区之前 await lock.acquire()，出来之后 lock.release()。但人有三急，代码会忘事——一旦忘记 release，所有等待的协程就永远堵在门口。所以 Python 更推荐 async with lock: 这种写法，它会在进入时自动获取锁，退出时自动释放，相当于“自动挂牌、自动摘牌”。

$$\displaystyle L \in \{0,1\}, \quad \text{acquire: } L=1 \ (\text{否则等待}), \quad \text{release: } L=0 + \text{唤醒队首}$$

1. 请求获取：await lock.acquire()，锁空闲则标记占用，否则进入等待队列挂起

$$L = 0 \Rightarrow L = 1,\ \text{继续执行}$$

2. 临界区操作：读写共享数据，此刻只有本协程能进入这一段代码

$$x \leftarrow x + 1 \ (\text{安全})$$

3. 释放与唤醒：lock.release() 归还锁，并唤醒等待队列里的一个协程

$$L \leftarrow 0,\ \text{唤醒队首协程}$$

4. 推荐写法：async with lock 自动完成获取与释放，杜绝忘记 release

$$\text{async with } \mathrm{lock}: \cdots$$

“挂牌”规矩听着简单，但若大厨端着菜忘了摘牌，整条出菜线就堵死了。代码里对应的就是“忘记 release”造成的死锁。还要记住：asyncio.Lock 不是可重入锁，同一个协程对同一把锁连拿两次，等于把自己挂死在自己挂的牌子上。

Semaphore信号量

外卖窗口不是“只能一个人用”，而是“最多同时接待五个人”。Semaphore 就是这种“限位”工具，它其实是 Lock 的广义版：Lock 等价于 Semaphore(1)，而 Semaphore(3) 允许最多三个协程同时进入受保护区域。

Semaphore 内部维护一个计数器。await sem.acquire() 把计数减一，减到 0 时后来者排队等待；sem.release() 把计数加一并唤醒一个等待者。它特别适合限流场景：数据库连接池、爬虫并发数、同时下载的任务数——你想限制“最多 N 个”，就用 asyncio.Semaphore(N)。

$$\displaystyle S = \mathrm{limit} - N_{\text{占用}}, \quad \text{acquire: } S \gets S-1,\quad \text{release: } S \gets S+1$$

1. 设定上限：Semaphore(5) 相当于在窗口摆了 5 个外卖格口

$$S_{\text{init}} = 5$$

2. 占用一个：await sem.acquire() 让计数减一，减到 0 则排队

$$S \gets S - 1 \ (S > 0 \ \text{时成功，否则等待})$$

3. 归还一个：sem.release() 让计数加一，并唤醒一个排队的协程

$$S \gets S + 1$$

4. 并发容量：任意时刻最多 limit 个协程同时处于受保护区域

$$N_{\text{并发}} \le \mathrm{limit}$$

就像外卖窗口只能同时递五份外卖一样，连接池、限流器都可以用信号量“限位”。但窗口格口被占用后必须归还——如果协程拿到信号量却忘了 release，额度就永久少一个，最终所有协程都会堵在门口。

Event事件

后厨装了一个出菜铃。服务员不用一直盯着后厨看，而是在出菜口等着；铃一响，所有服务员的手机同时收到通知，一起涌向窗口。asyncio.Event 就是这枚出菜铃，它不负责“限制人数”，只负责“广播信号”。

Event 内部是一个布尔状态：False 表示未设置，True 表示已设置。协程调用 await e.wait() 时，如果状态是 False 就挂起；e.set() 把状态置为 True 并唤醒所有等待者；e.clear() 把它重置回 False，供下一轮复用。Event 有个容易忽视的特性——它有“记忆”：set 之后再调用 wait 的协程会立即返回，直到 clear 才重新开始等待。

$$\displaystyle E \in \{\mathrm{False},\mathrm{True}\}, \quad \text{wait: 阻塞直到 } E=\mathrm{True}, \quad \text{set: } E \gets \mathrm{True}$$

1. 创建事件：asyncio.Event()，初始状态为 False，相当于铃声未响

$$E = \mathrm{False}$$

2. 等待铃响：await e.wait()，状态为 False 则协程挂起休眠

$$\text{await } \mathrm{e.wait()} \Rightarrow \text{挂起}$$

3. 触发广播：e.set() 把状态置为 True，并唤醒全部等待者

$$E \leftarrow \mathrm{True},\ \text{全体唤醒}$$

4. 重置事件：e.clear() 回到 False，供下一轮场景复用

$$E \leftarrow \mathrm{False}$$

出菜铃的妙处在于广播：一声铃响，所有服务员同时动起来。它适合“多个协程等同一个信号”的场景，比如模型训练完成、缓存预热结束、批量数据就绪——一个 set 通知所有人，干净利落。

Condition条件

大厨与服务员之间有一部对讲机。服务员问：“水煮鱼好了吗？”大厨说：“还没，你先去等会儿。”（wait）等鱼出锅，大厨喊：“水煮鱼好了！”（notify）服务员跑回来端菜。Condition 就是这台对讲机，它把“锁”和“等待通知”绑在了一起。

Condition 的核心价值在于“原子性”。使用前必须先拿到锁；cond.wait() 会先释放锁，再把协程挂起——这保证了“检查条件”和“进入睡眠”之间不会被其他协程插队，从而不会错过通知（missed wakeup）。被唤醒后，协程还要重新抢回锁才能继续执行。

$$\displaystyle \mathrm{wait()}: \ \text{释放锁} \to \text{挂起}; \qquad \mathrm{notify()}: \ \text{唤醒一个等待者并让它重新抢锁}$$

1. 持锁检查：在 with cond 的保护下检查条件，比如“队列是否满了”

$$\text{持有锁时检查条件}$$

2. 原子等待：cond.wait() 释放锁并挂起自己，不占着锁干等

$$L \leftarrow 0,\ \text{挂起等待通知}$$

3. 通知唤醒：cond.notify() 唤醒一个等待者，notify_all() 唤醒全部

$$\mathrm{notify} \Rightarrow \text{唤醒一个协程}$$

4. 重新抢锁：被唤醒的协程要重新 acquire 锁成功后才能继续

$$L \leftarrow 1,\ \text{继续执行}$$

对讲机和出菜铃的区别在于：铃响后所有人自己去取；而对讲机通知的是一个“正在等待的特定流程”。更重要的是，Condition 必须“边持锁边等待”才不会漏听，这正是生产者-消费者队列的经典场景。

异步生成器

回转寿司店的后厨不会在开餐前把一百盘寿司全摆上桌，而是一盘一盘现做现放；客人吃一盘，传送带补一盘——这就是惰性生产。异步生成器（Async Generator）就是这条传送带：它不一次性生产完整数据，而是生产一条、交出一条。

同步生成器用 yield 让函数暂停与恢复；异步生成器则是在 async def 函数里写 yield。这样一来，每次产出数据之前都可以先 await 等待 IO——比如读数据库、发网络请求。它把“异步取数据”和“逐条交给消费者”两件事无缝衔接起来。

$$\displaystyle \text{async def } g(k): \quad \mathrm{item}_k = \text{await 生产}(k); \quad \text{yield } \mathrm{item}_k$$

1. 定义生成器：async def 函数体内出现 yield，函数即成为异步生成器

$$\text{async def } \mathrm{g}(k)$$

2. 异步生产：先 await 网络、数据库等 IO 操作，得到一条数据

$$\mathrm{item}_k = \text{await } \mathrm{fetch}(k)$$

3. 逐条产出：yield 把这条数据交给消费者，并暂停在这里

$$\text{yield } \mathrm{item}_k$$

4. 惰性节奏：只有被消费者索取时才执行到下一个 yield

$$\text{消费一条} \Rightarrow \text{再生产一条}$$

寿司带的好处是“现做现吃”：内存不爆、首字节更快。拉取分页数据、流式读大文件、LLM 逐 token 输出，全都是这个节奏——异步生成器正是流式数据的最佳源头。

异步迭代async for

服务员端着盘子沿着传送带走，看到一盘夹一盘，直到传送带尽头——这就是 async for。有了异步生成器这条“传送带”，还需要一把合适的“夹子”，async for 就是为异步迭代器量身定做的语法糖。

一个对象只要实现了 __aiter__（返回异步迭代器）和 __anext__（返回一个 awaitable 的下一项），就是异步迭代器。async for x in a: 会自动循环调用 __anext__ 并 await 它，直到捕获 StopAsyncIteration 异常才结束。也就是说，async for = for 循环 + 自动 await 取下一项。

$$\displaystyle \text{async for } x \ \text{in } a: \quad x_n = \text{await } a.\_\_anext\_\_() \quad \text{直到 StopAsyncIteration}$$

1. 获取迭代器：循环开始前调用 aiter() 拿到异步迭代器对象

$$\mathrm{it} = a.\_\_aiter\_\_()$$

2. 取下一个：每一轮循环都 await 一次 it.anext()，得到一条数据

$$x_n = \text{await } \mathrm{it}.\_\_anext\_\_()$$

3. 循环终止：anext 抛出 StopAsyncIteration 时，循环正常结束

$$\mathrm{StopAsyncIteration} \Rightarrow \text{结束循环}$$

4. 使用限制：普通 for 循环和 next() 无法处理异步迭代器

$$\text{必须 async for，或手动 await } \_\_anext\_\_$$

沿传送带夹菜与开餐前把菜全堆上桌的差别，就是 async for 与一次性拉取列表的差别。它把“异步生产”和“异步消费”优雅地缝合在一起，是流式 AI 服务的基石写法。

AI 应用场景

Python 演示

import asyncio  async def number_generator(n):    # 异步生成器: yield 让函数变成"一件一件出菜"的传送带    for i in range(n):        await asyncio.sleep(0.05)  # 模拟 IO 等待, 让出事件循环        yield i  async def worker(lock, counter, name, total):    # 三个协程模拟后厨帮工, 竞争修改同一个共享计数器    for _ in range(total):        async with lock:  # 进入临界区前自动拿到锁            counter["value"] += 1            current = counter["value"]        print(f"{name} safe write, count={current}")        await asyncio.sleep(0.02)  async def main():    lock = asyncio.Lock()          # 灶台挂牌, 同一时刻只许一人改写    counter = {"value": 0}         # 用字典模拟可变共享变量    await asyncio.gather(          # 并发启动三个协程        worker(lock, counter, "Chef-A", 4),        worker(lock, counter, "Chef-B", 4),        worker(lock, counter, "Chef-C", 4),    )    print(f"final count: {counter['value']} (expect 12)")    print("=== consume async generator ===")    async for num in number_generator(5):  # 沿传送带逐盘夹取        print(f"got: {num}")  if __name__ == "__main__":    asyncio.run(main())

1. async with lock:相当于“自动挂牌摘牌”，进入前 acquire、退出后 release，绝不会忘记归还锁。
2. await asyncio.gather(...)一次性并发调度三个协程，像同时安排三位帮工上工。
3. counter["value"] += 1读改写三步，正因在锁内才不会被其他协程撕成两半。
4. await asyncio.sleep(0.05)模拟 IO 让出 CPU，这正是协程交错发生的地方。
5. yield i函数体内出现 yield 就变成生成器，配上 async 就是异步生成器。
6. async for num in number_generator(5):自动逐条 await 取数据，直到生产完毕。

常见误区

面试题