Python threading.Condition与Event区别及独有实现场景咨询

数据老陈

2026-8-28

先把两个原语的本质差异说透

你之前测试的那种一次性通知所有等待线程的简单生产者消费者场景，两个工具确实能跑出完全一致的结果，但你要知道threading.Event从设计上就是个极简的单标志同步工具，能力边界非常明确：

内部就存了一个布尔值，核心操作只有设为True（set()）、设为False（clear()）、阻塞等标志变True（wait()）三个
没有内置互斥锁，你没法原子性地完成「检查共享状态→不满足就阻塞等待」这一套操作
只要标志被设为True，所有卡在wait()上的线程会全部被唤醒，根本做不到唤醒指定数量的等待线程
没有「等待时自动释放锁、被唤醒后自动重新加锁」的机制，处理不了会反复在成立/不成立之间切换的等待条件

而threading.Condition本质上是绑定了互斥锁的线程等待队列，从设计之初就是用来解决「等待某个共享状态满足特定条件」这类复杂同步场景的，这部分能力Event从根上就不具备。

典型Event无法实现的场景：有界阻塞队列

这个场景是工程里非常常用的同步模型，你靠原生Event根本做不出正确的实现，核心规则如下：

队列有固定最大容量，多个生产者线程并发往队列写数据，队列写满时生产者必须阻塞停手
多个消费者线程并发从队列读数据，队列读空时消费者必须阻塞等数据
生产者写完一条数据只需要唤醒一个等着读的消费者就行，消费者读完一条数据只需要唤醒一个等着写的生产者，没必要把所有等的线程全喊起来造成无意义的抢锁（也就是常说的惊群问题）
所有对队列的读写必须互斥，不能出现两个线程同时改队列导致数据乱掉
整个流程是持续循环运行的，队列会反复在满、空、非满非空的状态切换，根本不存在「某件事发生了之后所有人都不用再等」的一次性标志位——这也是你之前写的简单示例和真实场景的最大区别。

你要是硬要说用Event加一堆额外变量、额外锁也能凑，那本质上是你自己在外面重新实现了一遍Condition的逻辑，还要自己处理信号丢失、唤醒时机错误、惊群之类的一堆坑，根本不算用Event本身的能力实现。

可运行示例代码

import threading
import time
import random
import logging

# 配置日志格式，方便区分线程输出
logging.basicConfig(
    level=logging.DEBUG,
    format='(%(threadName)-9s) %(message)s',
)

# 基于Condition实现的有界阻塞队列
class BoundedBlockingQueue:
    def __init__(self, max_size):
        self.max_size = max_size
        self.queue = []
        # Condition默认自带可重入锁RLock，也支持传入自定义锁
        self.cv = threading.Condition()

    def put(self, item):
        # 先获取和Condition绑定的互斥锁，保证后续操作原子性
        with self.cv:
            # 循环检查条件：队列满时就阻塞等待
            # wait()调用会自动释放持有的锁，被唤醒后会自动重新获取锁再继续执行
            while len(self.queue) >= self.max_size:
                logging.debug(f"队列已满，生产者阻塞，当前队列长度{len(self.queue)}")
                self.cv.wait()
            # 走到此处说明已持有锁、且队列非满，可以安全写入数据
            self.queue.append(item)
            logging.debug(f"生产者放入数据 {item}，当前队列长度{len(self.queue)}")
            # 仅唤醒1个等待的消费者，不唤醒其他生产者，避免无意义的性能损耗
            self.cv.notify(n=1)

    def get(self):
        with self.cv:
            # 循环检查条件：队列空时就阻塞等待
            while len(self.queue) == 0:
                logging.debug("队列为空，消费者阻塞等待")
                self.cv.wait()
            # 走到此处说明已持有锁、且队列非空，可以安全读取数据
            item = self.queue.pop(0)
            logging.debug(f"消费者取出数据 {item}，当前队列长度{len(self.queue)}")
            # 仅唤醒1个等待的生产者
            self.cv.notify(n=1)
            return item

# 生产者逻辑：循环生成随机数写入队列
def producer(queue):
    while True:
        item = random.randint(1, 100)
        time.sleep(random.random()*2) # 模拟生产耗时
        queue.put(item)

# 消费者逻辑：循环从队列读取数据
def consumer(queue):
    while True:
        time.sleep(random.random()*3) # 模拟消费耗时
        queue.get()

if __name__ == "__main__":
    # 队列最大容量设为2，更容易触发满/空的阻塞逻辑
    q = BoundedBlockingQueue(max_size=2)
    # 启动2个生产者、3个消费者
    for i in range(2):
        t = threading.Thread(target=producer, args=(q,), name=f"producer{i}")
        t.daemon = True
        t.start()
    for i in range(3):
        t = threading.Thread(target=consumer, args=(q,), name=f"consumer{i}")
        t.daemon = True
        t.start()
    # 运行10秒观察输出
    time.sleep(10)
    logging.debug("演示结束")

为什么Event写不出正确的有界阻塞队列

你自己动手试下就知道，有三个坎根本绕不过去：

竞态问题避不开：Event本身不带锁，比如你先判断队列是空的，正准备调wait()阻塞，这时候刚好生产者往队列里塞了数据调了set()，你这个wait()调用就直接错过了唤醒信号，大概率会永久卡死。而Condition的wait()是和绑定的锁联动的，从检查条件到真正进入等待状态的整个过程是原子的，根本不会出这种问题。
精准唤醒做不到：Event只要一调set()，所有等在这个Event上的线程全会醒，你根本没法实现「只唤醒一个消费者/生产者」的逻辑，最后要么所有线程全醒过来抢锁平白浪费性能，要么就得自己额外维护等待线程计数、唤醒计数，等于自己手搓了个Condition。
动态条件处理不了：Event的标志是全局的，你set()之后如果不手动clear()，后面所有调wait()的线程根本不会阻塞直接就往下走；但你根本找不到一个安全的时机去clear()——刚把标志设成False，可能就有线程因为条件不满足要等，直接就卡错过了之前的有效信号。而Condition的逻辑是每次调wait()就老老实实进等待队列，被唤醒了才会退出，天生就适合配合循环反复检查条件。

内容的提问来源于stack exchange，提问作者Saumya Sengupta