4、线程间通信：Condition条件变量、Event事件、Queue队列

多线程编程里，最让人头疼的问题是什么？

我个人觉得，不是锁，而是线程之间怎么「说话」。你想想看，一个线程下载数据，另一个线程处理数据，第三个线程写文件。它们之间怎么配合？怎么通知对方「我干完了，你上」？

这就是线程间通信要解决的问题。Python 给了我们三件法宝：Condition、Event 和 Queue。我这些年做项目，几乎每个多线程场景都会用到它们。

4.1 Condition：精准的条件等待

Condition，说白了就是「条件变量」。它比普通的锁多了一个功能：可以让线程在某个条件不满足时主动等待，等条件满足了再被唤醒。

我在项目中遇到过这样一个场景：一个线程负责从网络读取数据，另一个线程负责把数据写入文件。读线程必须等数据缓冲区有足够的数据才能开始写，而写线程写完后又需要通知读线程继续读。

用 Condition 实现起来很优雅：

import threading
import time

class DataBuffer:
    def __init__(self):
        self.buffer = []
        self.cond = threading.Condition()
    
    def produce(self, data):
        with self.cond:
            self.buffer.append(data)
            print(f"生产数据: {data}")
            # 通知等待的消费者
            self.cond.notify()
    
    def consume(self):
        with self.cond:
            while len(self.buffer) == 0:
                print("缓冲区为空，等待...")
                self.cond.wait()
            data = self.buffer.pop(0)
            print(f"消费数据: {data}")
            return data

# 使用示例
buffer = DataBuffer()

def producer():
    for i in range(5):
        buffer.produce(f"data-{i}")
        time.sleep(0.5)

def consumer():
    for _ in range(5):
        data = buffer.consume()
        time.sleep(1)

t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t1.start()
t2.start()
t1.join()
t2.join()

核心要点：

wait() 会释放锁并阻塞，直到被 notify() 唤醒
唤醒后会自动重新获取锁，继续执行
一定要在 with self.cond 的上下文中使用
用 while 循环检查条件，而不是 if（防止虚假唤醒）

我曾经踩过的坑：

有一次我用 if 判断条件，结果在极端情况下出现了虚假唤醒——线程被唤醒了但条件其实还没满足。程序跑起来偶尔会报错，查了两天才找到原因。记住：永远用 while 循环检查条件！

4.2 Event：简单粗暴的信号机制

Event 比 Condition 简单得多。它就是一个「信号灯」：要么亮着（set），要么灭着（clear）。线程可以等信号灯亮，也可以自己把灯点亮或熄灭。

我习惯用 Event 来做「一次性通知」或者「开关控制」。比如：

主线程通知工作线程「可以开始干活了」
工作线程通知主线程「我已经准备好了」
控制所有线程「停止运行」

import threading
import time

# 创建事件
start_event = threading.Event()
stop_event = threading.Event()

def worker(name):
    print(f"线程 {name} 等待开始信号...")
    start_event.wait()  # 等待事件被设置
    print(f"线程 {name} 开始工作")
    
    while not stop_event.is_set():
        print(f"线程 {name} 正在处理...")
        time.sleep(1)
    
    print(f"线程 {name} 收到停止信号，退出")

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

time.sleep(2)
print("主线程：发送开始信号")
start_event.set()  # 设置事件，所有等待的线程被唤醒

time.sleep(5)
print("主线程：发送停止信号")
stop_event.set()  # 设置停止事件

for t in threads:
    t.join()

小技巧：

Event 的 wait() 可以设置超时时间：event.wait(timeout=5)。这样线程不会无限等待，超时后会自动返回。我在做心跳检测时经常用这个特性。

4.3 Queue：最推荐的线程通信方式

说实话，如果让我选一种线程通信方式，我首选 Queue。为什么？因为它把「锁」和「条件变量」都封装好了，你直接用就行，几乎不会出错。

Queue 本质上是一个线程安全的队列。它支持：

put(item)：往队列放数据，如果队列满了就阻塞
get()：从队列取数据，如果队列空了就阻塞
task_done()：标记一个任务已完成
join()：等待所有任务完成

来看一个实际的文件处理场景：

import threading
import queue
import time

# 创建队列
file_queue = queue.Queue(maxsize=10)
result_queue = queue.Queue()

def file_reader(file_list):
    """模拟读取文件"""
    for file_name in file_list:
        print(f"读取文件: {file_name}")
        # 模拟读取耗时
        time.sleep(0.5)
        content = f"内容来自 {file_name}"
        file_queue.put(content)
        print(f"文件 {file_name} 已放入队列")
    
    # 发送结束信号
    file_queue.put(None)

def file_writer():
    """模拟写入文件"""
    while True:
        content = file_queue.get()
        if content is None:
            # 收到结束信号
            file_queue.task_done()
            break
        
        print(f"写入内容: {content[:20]}...")
        # 模拟写入耗时
        time.sleep(0.3)
        result_queue.put(f"已写入: {content}")
        file_queue.task_done()

# 启动线程
files = [f"data_{i}.txt" for i in range(5)]
reader = threading.Thread(target=file_reader, args=(files,))
writer = threading.Thread(target=file_writer)

reader.start()
writer.start()

reader.join()
writer.join()

# 输出结果
while not result_queue.empty():
    print(result_queue.get())

Queue 的优势：

特性	说明
线程安全	内部自动加锁，无需手动管理
阻塞控制	队列满时 put 阻塞，空时 get 阻塞
超时机制	支持 timeout 参数，避免死等
任务追踪	task_done + join 实现精确控制

4.4 三种方式怎么选？

我个人的经验是这样的：

Queue：90% 的场景用它就够了。生产者-消费者模式，数据传递，任务分发，统统用 Queue
Event：适合「一次性通知」或「开关控制」。比如通知线程启动、停止
Condition：需要精细控制等待条件时用。比如资源池管理、复杂的协作场景

避坑指南：

我曾经在一个项目里用 Condition 实现了一个复杂的协作逻辑，结果代码越写越乱。后来改成 Queue + Event 的组合，代码清晰多了。记住：能用 Queue 就别折腾 Condition，简单才是王道。

嗯，这三种方式各有各的用武之地。我个人建议：先从 Queue 入手，它最安全、最易用。等遇到更复杂的协作需求时，再考虑 Event 和 Condition。

记住一个原则：能用简单方案解决的问题，就别搞复杂。多线程编程已经够头疼了，别再给自己加戏。