4. 线程间通信：Condition条件变量、Event事件对象、Queue队列通信、Semaphore信号量

线程间通信，说白了就是让多个线程能「好好说话」。

我刚开始做并发编程时，总觉得线程之间各干各的就行。后来线上出了个诡异的Bug——两个线程同时修改共享数据，结果数据全乱了。嗯，从那以后我才明白：线程之间必须有一套成熟的通信机制。

Python里提供了四种核心工具：Condition、Event、Queue、Semaphore。咱们一个一个聊。

4.1 Condition：条件变量

Condition，我习惯叫它「条件变量」。它解决的是这样一个场景：一个线程在等某个条件成立，另一个线程负责把条件「搞成立」。

举个例子。生产者-消费者模型里，消费者必须等生产者生产了东西才能消费。用Condition实现就很自然：

import threading
import time

class MessageQueue:
    def __init__(self):
        self.cond = threading.Condition()
        self.msg = None

    def produce(self, msg):
        with self.cond:
            while self.msg is not None:
                self.cond.wait()  # 等待消费者取走消息
            self.msg = msg
            print(f"生产: {msg}")
            self.cond.notify()  # 通知消费者

    def consume(self):
        with self.cond:
            while self.msg is None:
                self.cond.wait()  # 等待生产者生产
            result = self.msg
            self.msg = None
            print(f"消费: {result}")
            self.cond.notify()  # 通知生产者
            return result

4.2 Event：事件对象

Event 比 Condition 更轻量。它就是一个「开关」——要么开（set），要么关（clear）。

你想想看，如果只是通知「可以开始了」，不需要传递数据，用 Event 就够。我经常用它来做「启动信号」：

import threading
import time

start_event = threading.Event()

def worker(name):
    print(f"{name} 等待开始信号...")
    start_event.wait()  # 阻塞，直到事件被设置
    print(f"{name} 开始工作！")

# 启动多个工作线程
threads = [threading.Thread(target=worker, args=(f"Worker-{i}",)) for i in range(3)]
for t in threads:
    t.start()

time.sleep(2)
print("主线程发出开始信号")
start_event.set()  # 所有等待的线程同时被唤醒

4.3 Queue：队列通信

Queue 是我个人最推荐的线程间通信方式。它把「数据传递」和「线程安全」打包在一起，你几乎不需要自己加锁。

Python 的 queue.Queue 内部已经实现了 Condition 和锁。你只管 put 和 get 就行：

import threading
import queue
import time

q = queue.Queue(maxsize=5)

def producer():
    for i in range(10):
        q.put(i)
        print(f"生产: {i}")
        time.sleep(0.5)

def consumer():
    while True:
        item = q.get()
        print(f"消费: {item}")
        q.task_done()  # 告诉队列任务完成

# 启动生产者和消费者
threading.Thread(target=producer, daemon=True).start()
threading.Thread(target=consumer, daemon=True).start()

# 等待所有任务完成
q.join()
print("所有任务处理完毕")

我在项目中遇到过一个问题：生产者生产太快，消费者处理太慢。如果不限制队列大小，内存会一直涨。用 maxsize 限制后，put() 会自动阻塞，完美解决了背压问题。

4.4 Semaphore：信号量

Semaphore，说白了就是一个「计数器锁」。它控制同时访问某个资源的线程数量。

举个例子。数据库连接池只有5个连接，同时只能有5个线程使用。用 Semaphore 就很合适：

import threading
import time

# 最多允许3个线程同时访问
sem = threading.Semaphore(3)

def access_db(thread_id):
    with sem:  # 获取信号量，如果已满则阻塞
        print(f"线程 {thread_id} 正在访问数据库...")
        time.sleep(2)
        print(f"线程 {thread_id} 释放数据库连接")

# 启动10个线程
threads = [threading.Thread(target=access_db, args=(i,)) for i in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

4.5 四种机制对比

机制	适用场景	特点	我常用的地方
Condition	复杂的条件等待	需要手动 wait/notify	生产者-消费者（需要精确控制时）
Event	一次性信号通知	轻量，适合「开始/停止」信号	线程启动同步、服务关闭通知
Queue	数据传递	自带线程安全，最推荐	任务队列、日志收集
Semaphore	限制并发数	计数器锁，控制资源访问	连接池、限流

4.6 知识体系图

下面这张图帮你理清这四种机制的关系和选择逻辑：

嗯，这四种机制基本覆盖了日常开发中90%的线程间通信场景。我个人最常用的是 Queue，因为它最省心。但遇到需要精细控制的情况，Condition 和 Semaphore 也各有千秋。

想要更多资料，联系可可