如何在Python中通过asyncio.Event实现任务间的通信？

asyncio.Event 到底能不能在协程之间传递数据？先说结论：不能。它的定位很明确——就是一个信号量，只负责告诉其他协程“条件已满足”，不带任何附加信息。

很多人一开始会误以为 set("done") 能把“完成”这个值传出去，但一旦这么写，Python 会直接扔给你一个 TypeError: set() takes no arguments。调用 wait() 的协程也不会收到任何返回值——它挂起、等待、然后继续，但永远拿不到 None 以外的结果。

所以，如果你需要携带数据的通知机制，得另寻他路。最稳妥的选择是 asyncio.Queue，它天然支持协程安全的数据传递。如果场景足够简单，用普通变量加 asyncio.Lock 保护也行，或者干脆用闭包变量——前提是确保所有读写都发生在同一个事件循环里。

那正确的用法是什么样的？

Event 最适合的场景是“等某个异步动作完成后再继续”。举个例子：后台监控任务检测到文件就绪后，通知主流程开始处理。

import asyncio

ready = asyncio.Event()

async def monitor_file():
    await asyncio.sleep(2)  # 模拟等待文件生成
    print("文件已就绪")
    ready.set()  # 只发信号，不传内容

async def process_file():
    print("等待文件...")
    await ready.wait()  # 挂起，直到 set() 被调用
    print("开始处理")

async def main():
    await asyncio.gather(monitor_file(), process_file())

asyncio.run(main())

需要注意几个细节：

第一，wait() 可以多次调用。只要 Event 处于 set 状态，后续任何 wait() 都会立刻返回——它不像某些一次性信号量，不会自动重置。如果你希望每次只触发一次，就必须手动调用 clear()，否则第二次 wait() 会直接跳过等待。

第二，is_set() 允许非阻塞查询当前状态，适合那些需要轮询判断的场景，比如在循环中检查某个条件是否已经满足。

多个协程同时等待一个 Event 时会发生什么？

所有正在 wait() 的协程会在 set() 之后几乎同时被唤醒——注意，是“几乎同时”，调度顺序由事件循环决定，没有优先级，也不保证 FIFO 顺序。

这种机制在实际开发中很有用。比如启动阶段，多个 worker 协程都在等待初始化完成信号，一旦所有依赖服务就绪，Event 一触发，所有 worker 就会同时拿到“开工”的指令。再比如写测试时，用它来模拟“所有依赖服务已启动”这个条件。

不过，有几个容易踩的坑，得提前留意：

一是忘记 clear() 导致后续测试用例误判。这在 pytest 的 async fixtures 里尤其常见——上一个测试把 Event 置位了，下一个测试没手动清零，结果 wait() 瞬间返回，测试可能就“假绿”了。

二是 Event 的“记忆性”。如果 set() 先已经发生过，后调用的 wait() 会立刻返回，不会阻塞。这点和 threading.Event 完全一致，属于有状态的行为，不是 bug。

三是不要跨事件循环复用同一个 asyncio.Event 实例。比如你在两个不同的 asyncio.run() 之间共享同一个 Event 对象，会直接触发 RuntimeError。事件循环之间的资源隔离是强制的。

再说一个容易被忽略的细节：Event 本身不解决竞态问题。它只是一个同步开关。如果有两个协程同时修改共享数据，又只靠 Event 来协调，那还是会有竞争——你需要额外的锁机制，或者改用 Queue 来串行化访问。Event 只管“什么时候做”，不管“怎么做”。