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

多线程编程里，最让人头疼的问题是什么？我个人觉得，不是锁，而是线程之间怎么「好好说话」。你想想看，多个线程同时跑，各干各的活，但有时候它们需要互相打个招呼、传个数据、或者等对方干完活再动手。这就是线程间通信要解决的问题。

Python 给我们提供了三种主要的通信工具：Condition、Event 和 Queue。它们各有各的适用场景，用错了地方，代码就会变得又慢又难调试。我在项目中踩过不少坑，今天就把这些经验分享给你。

4.1 Condition：让线程学会等待和通知

Condition 条件变量，说白了就是一把「高级锁」。它不光能保护共享资源，还能让线程在某个条件不满足时主动等待，等条件满足了再被唤醒。

我举个例子。假设你写一个生产者-消费者程序，生产者往列表里放数据，消费者从列表里取数据。如果列表为空，消费者就得等着。用 Condition 来实现，代码会非常优雅。

import threading
import time
import random

class SharedBuffer:
    def __init__(self, max_size=5):
        self.buffer = []
        self.max_size = max_size
        self.condition = threading.Condition()
    
    def produce(self, item):
        with self.condition:
            # 如果缓冲区满了，就等待
            while len(self.buffer) >= self.max_size:
                print(f"缓冲区满了，生产者等待...")
                self.condition.wait()
            self.buffer.append(item)
            print(f"生产者生产了: {item}, 缓冲区: {self.buffer}")
            # 通知消费者可以取了
            self.condition.notify()
    
    def consume(self):
        with self.condition:
            # 如果缓冲区空了，就等待
            while len(self.buffer) == 0:
                print(f"缓冲区空了，消费者等待...")
                self.condition.wait()
            item = self.buffer.pop(0)
            print(f"消费者消费了: {item}, 缓冲区: {self.buffer}")
            # 通知生产者可以生产了
            self.condition.notify()
            return item

def producer(buffer):
    for i in range(10):
        time.sleep(random.uniform(0.1, 0.5))
        buffer.produce(f"商品-{i}")

def consumer(buffer):
    for i in range(10):
        time.sleep(random.uniform(0.2, 0.6))
        buffer.consume()

buffer = SharedBuffer()
t1 = threading.Thread(target=producer, args=(buffer,))
t2 = threading.Thread(target=consumer, args=(buffer,))
t1.start()
t2.start()
t1.join()
t2.join()
print("所有任务完成")

4.2 Event：简单粗暴的信号灯

Event 事件，比 Condition 简单得多。它就像一个红绿灯，只有两种状态：绿灯（set）和红灯（clear）。线程可以等红灯变绿，也可以把红灯变成绿灯。

我习惯用 Event 来做「一次性」的同步。比如，主线程等所有子线程准备好再开始干活，或者某个线程等另一个线程完成初始化后再继续。

import threading
import time

def worker(event, worker_id):
    print(f"工作线程 {worker_id} 正在准备...")
    time.sleep(worker_id)  # 模拟不同的准备时间
    print(f"工作线程 {worker_id} 准备就绪")
    event.set()  # 通知主线程

def main():
    event = threading.Event()
    threads = []
    
    for i in range(3):
        t = threading.Thread(target=worker, args=(event, i))
        t.start()
        threads.append(t)
    
    # 等待所有线程准备就绪
    event.wait()
    print("所有线程准备就绪，开始执行主任务")
    
    for t in threads:
        t.join()

if __name__ == "__main__":
    main()

4.3 Queue：最安全的线程间数据通道

Queue 队列，是我在实际项目中最常用的线程通信方式。它内部已经实现了锁机制，你不需要自己操心线程安全问题。说白了，就是「往队列里放数据，从队列里取数据」，就这么简单。

Python 的 queue.Queue 提供了三种队列：

队列类型	特点	适用场景
Queue（FIFO）	先进先出	生产者-消费者模式
LifoQueue（LIFO）	后进先出	栈式处理
PriorityQueue	优先级队列	任务优先级调度

import threading
import queue
import time
import random

def producer(q, name):
    for i in range(5):
        item = f"{name}-任务-{i}"
        time.sleep(random.uniform(0.1, 0.3))
        q.put(item)
        print(f"生产者 {name} 放入: {item}")
    # 发送结束信号
    q.put(None)

def consumer(q, name):
    while True:
        item = q.get()
        if item is None:
            q.task_done()
            break
        print(f"消费者 {name} 取出: {item}")
        time.sleep(random.uniform(0.2, 0.4))
        q.task_done()

q = queue.Queue(maxsize=3)
producers = []
consumers = []

for i in range(2):
    p = threading.Thread(target=producer, args=(q, f"P{i}"))
    producers.append(p)
    p.start()

for i in range(2):
    c = threading.Thread(target=consumer, args=(q, f"C{i}"))
    consumers.append(c)
    c.start()

for p in producers:
    p.join()

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

4.4 三种通信方式的对比

你可能会问，到底该用哪个？我根据自己的经验，给你一个选择指南：

通信方式	复杂度	适用场景	我的推荐指数
Condition	高	复杂的条件等待、多条件判断	⭐⭐⭐
Event	低	一次性同步、信号通知	⭐⭐⭐⭐
Queue	中	数据传递、任务调度	⭐⭐⭐⭐⭐

4.5 知识体系总览

下面这张图，帮你理清这三种通信方式的关系和适用场景：

嗯，到这里，三种线程间通信方式就讲完了。我个人最常用的是 Queue，因为它最安全、最省心。Event 适合做简单的信号灯。Condition 虽然功能强大，但容易写出 bug，我一般只在万不得已时才用它。

记住一点：线程间通信的核心是「同步」和「数据传递」。选对工具，你的多线程代码就会既高效又稳定。

可可精品资料库