Python asyncio.Condition.wait_for 如何处理虚假唤醒

一个异步消费者明明刚从 await condition.wait() 返回，下一行读取队列却抛出 IndexError，这不是矛盾。被唤醒只说明等待结束了，不保证共享状态在当前协程重新拿到锁时仍满足业务条件。

asyncio.Condition.wait_for(predicate) 的处理方式，是在持有 Condition 底层锁时检查谓词；谓词为假就继续等待，醒来后重新拿锁并再次检查，直到谓词为真才返回。它把容易漏写的 while not predicate(): await condition.wait() 封装起来，因此正适合防御虚假唤醒和多个等待者之间的状态竞争。

官方文档：https://docs.python.org/3/library/asyncio-sync.html#asyncio.Condition.wait_for

故障现场：等待结束了，队列却还是空的

假设两个消费者都在等同一个队列。生产者放入一个元素后调用 notify_all()，两个消费者都会进入可运行状态，但它们仍然要竞争同一把锁。先拿到锁的消费者取走唯一元素；第二个消费者稍后拿到锁时，队列已经空了。

import asyncio
from collections import deque

queue = deque()
condition = asyncio.Condition()


async def unsafe_consumer() -> str:
    async with condition:
        # 错误点：一次通知不等于队列在重新拿锁时一定非空
        if not queue:
            await condition.wait()

        # 多个等待者竞争时，这里仍可能面对空队列
        return queue.popleft()

这类故障往往不是每次复现。只有多个任务同时等待、生产速度较慢，或者一次通知唤醒多个消费者时，时间窗口才明显。日志里可能只看到“收到通知”与“空队列异常”挨在一起，于是很容易把问题误判为队列实现或事件循环调度异常。

根因不在通知，而在把“唤醒”当成“条件成立”

asyncio.Condition 把事件通知和互斥锁组合在一起。调用 wait() 前必须持有锁；等待期间它会释放锁，让生产者能够修改共享状态；任务醒来后会先重新获取锁，然后才从 wait() 返回。

但通知本身没有携带“队列非空”“额度足够”或“状态已完成”这样的业务保证。Python 官方文档还直接说明，任务可能从 wait() 虚假返回，所以调用者必须重新检查状态，并准备再次等待。

在实际工程里，“虚假唤醒”可以分成两类看待：

• 原语层面的虚假返回：wait() 返回，但没有任何可依赖的业务状态变化。
• 业务层面的竞争失效：通知时条件确实成立，但当前任务重新拿到锁之前，另一个任务已经改变了状态。

对调用者来说，两类风险的解法相同：不要依据“我被通知了”继续，而要依据“受锁保护的谓词现在为真”继续。

wait_for 如何把防御性循环写对

Condition.wait_for(predicate) 接收一个普通可调用对象。它先执行谓词；若结果为假，就调用 wait()，醒来后再执行谓词，直到结果解释为真。最终返回值就是谓词最后一次计算得到的值。

async def safe_consumer() -> str:
    async with condition:
        # 谓词在持锁状态下读取共享队列，并在每次唤醒后重新检查
        await condition.wait_for(lambda: bool(queue))

        # wait_for 返回时仍持有同一把锁，因此可以原子地取走元素
        return queue.popleft()

它在语义上相当于下面这个循环。关键不是方法名，而是 while：条件不满足就继续等待，而不是用 if 只判断一次。

async def safe_consumer_manual() -> str:
    async with condition:
        # while 可以覆盖虚假返回和其他消费者先取走资源两种情况
        while not queue:
            await condition.wait()

        # 退出循环说明当前持锁视图中的队列确实非空
        return queue.popleft()

谓词应该同步、短小、无副作用。不要把协程函数传进去，也不要在谓词里执行网络请求、磁盘 I/O 或修改共享对象。它可能被调用多次，每次都应只回答“现在是否允许继续”。

生产者也必须遵守同一把锁的边界

只改消费者还不够。生产者应先通过 async with condition 获取底层锁，在锁内更新共享状态，再调用通知。这样等待者重新拿锁后看到的状态，与通知所对应的修改处于同一个同步边界。

async def producer(item: str) -> None:
    async with condition:
        # 先在 Condition 的锁内改变谓词依赖的共享状态
        queue.append(item)

        # 一个元素通常只需唤醒一个等待者，减少无效竞争
        condition.notify(1)

如果一次状态变化能满足多个任务，才考虑 notify_all()。即使使用 notify(1)，消费者也不能省略谓词循环，因为代码以后可能出现新的通知来源、状态回滚或不同等待条件。

多个条件也可以共享一把 asyncio.Lock，适合不同任务关注同一状态对象的不同谓词。例如“队列非空”和“队列未满”可以分别使用两个 Condition，但都基于同一把锁，避免读取到互相矛盾的状态。

带返回值的谓词能减少重复读取

wait_for 返回谓词的最终值，而不仅是固定的 True。因此谓词可以返回一个受锁保护的对象或索引，只要假值代表“继续等待”，真值代表“可以继续”。不过，复杂谓词会降低可读性，队列场景通常用布尔判断最清楚。

state = {"ready_item": None}


async def wait_ready_item() -> str:
    async with condition:
        # 返回对象本身；None 表示继续等待，字符串表示条件已满足
        item = await condition.wait_for(lambda: state["ready_item"])

        # 在持锁状态下清空槽位，避免另一个任务重复消费
        state["ready_item"] = None
        return item

这里的前提仍然是：所有读写 state["ready_item"] 的协程都遵守同一把锁。如果有代码绕开锁直接赋值，Condition 无法替你建立一致性。

超时和取消要放在条件循环外层处理

asyncio 的同步原语方法不直接接收 timeout 参数。Python 3.11 及以上可以用 asyncio.timeout() 限制整个等待区间。超时上下文会通过取消当前任务结束等待，并在上下文外转换成内置 TimeoutError。

async def consume_with_timeout(seconds: float) -> str | None:
    try:
        # 超时覆盖整个条件等待，不需要自行计算每轮剩余时间
        async with asyncio.timeout(seconds):
            async with condition:
                await condition.wait_for(lambda: bool(queue))
                return queue.popleft()
    except TimeoutError:
        # 超时不是虚假唤醒；调用方明确决定返回空结果
        return None

较老版本可以使用 await asyncio.wait_for(condition.wait_for(predicate), timeout=seconds)。无论哪种写法，都不要吞掉外部任务取消产生的 asyncio.CancelledError；确需清理资源时用 try/finally，完成清理后让取消继续传播。

防复发检查清单

• 等待者是否通过 async with condition 持有底层锁？
• 代码是否用 wait_for(predicate) 或 while，而不是 if + wait()？
• 谓词读取的共享状态是否只在同一把锁下修改？
• 生产者是否先修改状态，再在锁内调用 notify() 或 notify_all()？
• 谓词是否同步、快速、无副作用，并能安全重复调用？
• 多个消费者是否测试过“一个资源唤醒多个任务”的竞争情况？
• 超时是否覆盖整个等待过程，取消是否正确向上传播？

相关问题

wait_for 的 predicate 可以是 async 函数吗？

不应该。它要求普通可调用对象，并把返回值解释为布尔值；协程对象本身是真值，还没有被等待，会让条件判断失去意义。需要异步 I/O 时，应在 Condition 外完成，再在锁内更新共享状态。

notify_all 之后为什么仍要检查谓词？

因为所有等待者只是获得继续竞争的机会。它们逐个重新获取锁，前面的任务可能已经消耗或改变共享状态，后面的任务必须重新判断。

只有一个消费者时可以直接 wait 吗？

仍不建议。官方文档明确说明 wait() 可能虚假返回；使用 wait_for 还能让代码在未来增加消费者或通知来源时保持正确。

Condition 和 Event 有什么区别？

Event 维护一个布尔标志，适合广播某个持久状态；Condition 把通知和锁结合，适合围绕复杂共享状态反复检查谓词。队列容量、状态机阶段和多条件资源通常更适合 Condition。

谓词返回真后还会丢失条件吗？

wait_for 返回时调用方仍持有 Condition 的锁，所以只要所有参与者都遵守同一把锁，调用方可以立即读取或消费状态，不会被另一个协程插入修改。

这次故障的根因可以压缩成一句话：通知是“重新检查”的信号，不是“直接继续”的许可证。把业务条件写成谓词，让 wait_for 在锁内反复检查，再把状态修改与通知放进同一个 Condition 边界，虚假唤醒就不会再穿透到业务代码。