3、线程间通信机制：Event、Condition、Queue的原理与选择
多线程编程里，线程间怎么“说话”，是个绕不开的话题。说白了，就是怎么让一个线程知道另一个线程干完了活，或者怎么把数据安全地递过去。
我这些年做高并发系统，见过太多因为通信机制选错而翻车的案例。有人用错了 Condition，导致死锁；有人滥用 Event，性能一塌糊涂。今天咱们就把 Event、Condition、Queue 这三兄弟掰扯清楚。
3.1 Event：最简单的“信号灯”
Event 本质上就是个布尔标志位。一个线程等它亮，另一个线程把它点亮。就这么简单。
我习惯把它用在“一次性通知”的场景。比如服务启动时，主线程等所有子线程初始化完毕，再继续往下走。
set()— 把标志置为 True，唤醒所有等待的线程clear()— 把标志置为 Falsewait()— 阻塞，直到标志为 Trueis_set()— 检查当前状态
import threading
# 模拟：主线程等所有 worker 就绪
ready_event = threading.Event()
def worker(worker_id):
# 模拟初始化
print(f"Worker {worker_id} 初始化中...")
time.sleep(1)
print(f"Worker {worker_id} 就绪")
ready_event.set() # 点亮信号灯
# 主线程
ready_event.wait() # 阻塞，直到有 worker 点亮
print("所有 worker 就绪，开始干活")
3.2 Condition：更精细的“等待-通知”
Condition 比 Event 高级一点。它允许线程在满足某个条件时等待，另一个线程在条件变化时通知它。
嗯，这里要注意：Condition 必须和锁（Lock）配合使用。它内部维护了一个锁，你调用 wait() 时会释放锁，被唤醒后又会重新获取锁。
我曾经在一个资源池项目里用过 Condition。多个线程竞争有限的连接，没有连接时就等着，有人归还了就通知一个线程去抢。
import threading
class ConnectionPool:
def __init__(self, size):
self._pool = [f"conn-{i}" for i in range(size)]
self._cond = threading.Condition()
def acquire(self):
with self._cond:
while not self._pool: # 用 while 不是 if，防止虚假唤醒
self._cond.wait()
return self._pool.pop()
def release(self, conn):
with self._cond:
self._pool.append(conn)
self._cond.notify() # 通知一个等待的线程
Condition 的 notify() 只唤醒一个线程，notify_all() 唤醒所有。我个人习惯：如果不知道唤醒谁合适，就用 notify_all()，虽然有点浪费，但安全。
3.3 Queue：生产者的“传送带”
Queue 是我用得最多的通信方式。它把同步的细节都封装好了，你只管往里放、往外取就行。
说白了，Queue 就是“生产者-消费者”模式的完美实现。它内部用了 Condition 和锁，但对外暴露的是简单的 put() 和 get()。
| 方法 | 说明 | 阻塞行为 |
|---|---|---|
put(item, block=True, timeout=None) |
放入数据 | 队列满时阻塞，直到有空位或超时 |
get(block=True, timeout=None) |
取出数据 | 队列空时阻塞，直到有数据或超时 |
task_done() |
标记任务完成 | 配合 join() 使用，统计未完成任务数 |
join() |
等待所有任务完成 | 阻塞直到 task_done() 调用次数等于 put 次数 |
import threading
import queue
import time
q = queue.Queue(maxsize=10)
def producer():
for i in range(20):
q.put(f"data-{i}")
print(f"生产: data-{i}")
time.sleep(0.1)
def consumer():
while True:
data = q.get()
print(f"消费: {data}")
q.task_done() # 标记处理完成
# 启动消费者线程（设为守护线程，主线程结束它就结束）
t = threading.Thread(target=consumer, daemon=True)
t.start()
producer()
q.join() # 等待所有数据被消费
print("所有任务完成")
3.4 怎么选？一张图说清楚
你想想看，这三个东西各有各的脾气。选错了，轻则性能差，重则死锁。我总结了一个决策流程：
3.5 实战对比：三种场景怎么选
| 场景 | 推荐方案 | 原因 |
|---|---|---|
| 线程初始化完成通知 | Event | 一次性通知，不需要传递数据，简单高效 |
| 资源池（连接池、线程池） | Condition | 需要精确控制等待和通知，资源不足时等待，归还时通知 |
| 生产者-消费者（任务队列） | Queue | 需要传递数据，且可能有多个生产者和消费者 |
| 多阶段流水线处理 | 多个 Queue 串联 | 每个阶段一个队列，解耦且易于扩展 |
| 超时等待某个条件 | Condition + wait(timeout) | Event 的 wait() 也支持超时，但 Condition 更灵活 |
3.6 总结一下
Event 就是个开关，适合“喊一嗓子”的场景。Condition 像个会议室，大家排队等条件满足。Queue 就是传送带，数据从一头进另一头出。
我个人习惯：能上 Queue 就上 Queue，它封装得最好，不容易出错。如果只是简单通知，用 Event。需要精细控制等待条件，才用 Condition。
记住一点：线程间通信，安全第一，性能第二。别为了省几个微秒的锁开销，搞出死锁或者数据不一致，那才是真正的灾难。