3、线程间通信：Queue队列的使用、Event事件机制、Condition条件变量

线程间通信，说白了就是多个线程之间怎么「打招呼」和「传纸条」。

我刚开始写多线程代码时，总觉得这玩意儿挺简单——不就是共享变量嘛。结果呢？竞态条件、死锁、活锁，各种问题轮番上阵，搞得我焦头烂额。后来我才明白，线程通信这件事，光靠共享变量是远远不够的。

Python 的 threading 模块给了我们三件法宝：Queue、Event、Condition。今天咱们就一个一个聊透。

3.1 Queue：最安全的「快递员」

先说说 Queue。它本质上是一个线程安全的 FIFO 队列。你想想看，如果多个线程同时往一个列表里塞数据，不加锁的话，数据大概率会乱掉。但 Queue 内部已经帮你做好了锁管理，你只管往里放、往外取就行。

核心要点：Queue 是生产者-消费者模式的最佳拍档。

我在项目中遇到过这样一个场景：爬虫程序需要从多个页面抓取数据，然后统一写入数据库。如果每个线程直接写数据库，连接池肯定扛不住。我的做法是：用 Queue 作为缓冲，生产者线程抓数据，消费者线程批量写入。

import threading
import queue
import time

# 创建一个队列，最大容量为10
q = queue.Queue(maxsize=10)

def producer(name):
    """生产者：往队列里放数据"""
    for i in range(5):
        item = f"{name}-{i}"
        q.put(item)
        print(f"[生产者] {name} 放入了 {item}")
        time.sleep(0.5)

def consumer(name):
    """消费者：从队列里取数据"""
    while True:
        item = q.get()
        if item is None:  # 哨兵值，表示结束
            break
        print(f"[消费者] {name} 取出了 {item}")
        q.task_done()  # 通知队列任务完成

# 启动生产者线程
producers = []
for i in range(2):
    t = threading.Thread(target=producer, args=(f"P{i}",))
    producers.append(t)
    t.start()

# 启动消费者线程
consumers = []
for i in range(2):
    t = threading.Thread(target=consumer, args=(f"C{i}",))
    consumers.append(t)
    t.start()

# 等待所有生产者完成
for t in producers:
    t.join()

# 放入哨兵值，通知消费者退出
for _ in consumers:
    q.put(None)

# 等待所有消费者完成
for t in consumers:
    t.join()

print("所有任务完成！")

小技巧：使用 q.join() 可以阻塞主线程，直到队列中的所有任务都被处理完毕。配合 task_done() 使用，效果非常好。

Queue 还提供了几个实用的方法：

方法	说明	阻塞行为
`put(item, block=True, timeout=None)`	放入数据	队列满时阻塞，直到有空位
`get(block=True, timeout=None)`	取出数据	队列空时阻塞，直到有数据
`qsize()`	返回队列大小	非阻塞
`empty()`	判断队列是否为空	非阻塞
`full()`	判断队列是否已满	非阻塞

注意：我曾经犯过一个错误——在消费者线程里直接调用 q.get(block=False)，结果队列一空就抛出 queue.Empty 异常。后来我改用 block=True 配合 timeout，问题就解决了。

3.2 Event：简单粗暴的「信号灯」

Event 机制，说白了就是一个全局的布尔标志位。它只有两个状态：set()（绿灯）和 clear()（红灯）。线程可以调用 wait() 等待绿灯亮起。

我个人习惯用 Event 来做「一次性通知」的场景。比如：主线程等待所有子线程初始化完成后再开始工作。

import threading
import time

# 创建事件对象
init_event = threading.Event()

def worker(name):
    print(f"[{name}] 正在初始化...")
    time.sleep(2)  # 模拟初始化耗时
    print(f"[{name}] 初始化完成")
    init_event.set()  # 通知主线程

def task(name):
    print(f"[{name}] 等待初始化完成...")
    init_event.wait()  # 阻塞，直到事件被设置
    print(f"[{name}] 开始执行任务")

# 启动工作线程
t1 = threading.Thread(target=worker, args=("Worker",))
t2 = threading.Thread(target=task, args=("Task1",))
t3 = threading.Thread(target=task, args=("Task2",))

t1.start()
t2.start()
t3.start()

t1.join()
t2.join()
t3.join()

print("所有线程执行完毕")

Event 的典型应用场景：

服务启动等待：主线程等待所有子线程初始化完成
优雅关闭：设置关闭事件，所有线程检测到后自行退出
定时任务：配合 wait(timeout) 实现超时等待

避坑指南：我曾经用 Event 来做「多次触发」的场景，结果发现 Event 一旦被 set，后续的 wait 会立即返回。如果你需要多次触发，请用 Condition 或者自己封装一个计数器。

3.3 Condition：最灵活的「协调员」

Condition 条件变量，是这三者中最强大的一个。它结合了锁和等待/通知机制，可以实现非常复杂的线程协调逻辑。

你想想看，Queue 只能做「放-取」这种简单模式，Event 只能做「一次通知」。但 Condition 可以做到：线程 A 等待某个条件成立，线程 B 在条件满足时通知线程 A。

import threading
import time

class BoundedBuffer:
    """有界缓冲区，使用 Condition 实现"""
    def __init__(self, capacity):
        self.capacity = capacity
        self.buffer = []
        self.cond = threading.Condition()

    def put(self, item):
        with self.cond:
            while len(self.buffer) >= self.capacity:
                print("缓冲区已满，生产者等待...")
                self.cond.wait()  # 等待消费者通知
            self.buffer.append(item)
            print(f"生产者放入: {item}, 当前大小: {len(self.buffer)}")
            self.cond.notify()  # 通知消费者

    def get(self):
        with self.cond:
            while len(self.buffer) == 0:
                print("缓冲区为空，消费者等待...")
                self.cond.wait()  # 等待生产者通知
            item = self.buffer.pop(0)
            print(f"消费者取出: {item}, 当前大小: {len(self.buffer)}")
            self.cond.notify()  # 通知生产者
            return item

# 测试代码
buffer = BoundedBuffer(3)

def producer():
    for i in range(6):
        buffer.put(f"数据-{i}")
        time.sleep(0.3)

def consumer():
    for _ in range(6):
        item = buffer.get()
        time.sleep(0.5)

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

关键点：使用 Condition 时，wait() 必须在 with self.cond 内部调用。它会释放锁并阻塞，直到被 notify() 唤醒后重新获取锁。

Condition 的常用方法：

acquire() / release()：获取/释放底层锁
wait(timeout=None)：等待条件成立
notify(n=1)：唤醒 n 个等待线程
notify_all()：唤醒所有等待线程

注意：我曾经在 notify() 之后忘记释放锁，导致其他线程永远无法获取锁。记住：notify() 只是唤醒线程，但被唤醒的线程需要等到当前线程释放锁后才能继续执行。

3.4 三者对比与选型建议

说了这么多，到底什么时候用哪个？我总结了一张表：

场景	推荐方案	原因
生产者-消费者模式	Queue	内置锁管理，简单安全
一次性通知（如初始化完成）	Event	轻量级，语义清晰
复杂条件等待（如缓冲区满/空）	Condition	灵活，支持多次等待和通知
超时等待	Event 或 Condition	两者都支持 timeout 参数
多线程协作完成复杂任务	Condition + 自定义状态	可以精确控制每个线程的执行时机

嗯，这里要注意：不要为了炫技而用 Condition。能用 Queue 解决的问题，就别折腾 Condition。我见过不少代码，明明用 Queue 三行就能搞定，非要用 Condition 写二十行，结果还容易出 bug。

3.5 知识体系图

最后，我用一张图来总结本章的核心内容：

好了，线程间通信的三种方式就聊到这里。记住：没有银弹，只有最合适的工具。你在项目中遇到什么奇葩的线程通信问题，欢迎随时交流。

📚 课程目录

3、线程间通信：Queue队列的使用、Event事件机制、Condition条件变量

3.1 Queue：最安全的「快递员」

3.2 Event：简单粗暴的「信号灯」

3.3 Condition：最灵活的「协调员」

3.4 三者对比与选型建议

3.5 知识体系图