📚 课程目录第4章线程间通信

4. 线程间通信

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

我刚开始做多线程开发时，总觉得每个线程各干各的就行。后来线上出了个bug，两个线程同时修改共享数据，数据全乱套了。嗯，从那以后我才真正重视起线程通信来。

Python里常用的通信工具有三个：Condition、Event、Queue。它们各有各的适用场景。我一个个说。

4.1 Condition条件变量

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

举个例子。生产者消费者模型里，生产者往队列里放数据，消费者从队列里取数据。消费者发现队列空了，就得等着。生产者放完数据，得通知消费者「别等了，有货了」。

Condition就是干这个的。

核心用法：

acquire() / release()：获取/释放底层锁
wait()：释放锁并等待通知
notify() / notify_all()：唤醒等待的线程

import threading
import time

class MessageQueue:
    def __init__(self):
        self.items = []
        self.cond = threading.Condition()
    
    def put(self, item):
        with self.cond:
            self.items.append(item)
            print(f"生产了: {item}")
            self.cond.notify()  # 通知消费者
    
    def get(self):
        with self.cond:
            while not self.items:  # 这里要用while，不能用if
                print("队列空了，等待中...")
                self.cond.wait()
            item = self.items.pop(0)
            print(f"消费了: {item}")
            return item

def producer(q):
    for i in range(5):
        q.put(f"消息-{i}")
        time.sleep(0.5)

def consumer(q):
    for _ in range(5):
        q.get()
        time.sleep(1)

q = MessageQueue()
t1 = threading.Thread(target=producer, args=(q,))
t2 = threading.Thread(target=consumer, args=(q,))
t1.start()
t2.start()
t1.join()
t2.join()

避坑指南：

我曾经在项目里用if判断条件，结果出现了「虚假唤醒」——线程被唤醒了，但条件其实还没满足。后来改成while循环才解决。记住：wait()返回后一定要重新检查条件。

4.2 Event事件对象

Event比Condition更轻量。它就像一个信号旗：要么举着（set），要么放下（clear）。

我常用Event来做「一次性通知」。比如主线程等所有子线程启动完毕，再发号施令。

import threading
import time

start_event = threading.Event()
ready_events = []

def worker(worker_id):
    print(f"工人{worker_id} 已就位")
    ready_events[worker_id].set()  # 报告就绪
    start_event.wait()             # 等待开工信号
    print(f"工人{worker_id} 开始干活")

# 创建5个工人
ready_events = [threading.Event() for _ in range(5)]
threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

# 等待所有工人就绪
for evt in ready_events:
    evt.wait()
print("所有人就绪，发令！")
start_event.set()

for t in threads:
    t.join()

我的经验：

Event适合「一次性触发」的场景。如果你需要多次通知，或者需要传递数据，那还是用Condition或Queue更合适。

4.3 Queue队列通信

Queue是我用得最多的线程通信方式。它把锁、条件变量都封装好了，你只管往里放、往外取就行。

Python的queue.Queue是线程安全的。它内部用了Condition来实现阻塞和通知。你想想看，这比你自己手写Condition要省心多少？

import threading
import queue
import time
import random

q = queue.Queue(maxsize=5)

def producer():
    for i in range(10):
        item = f"产品-{i}"
        q.put(item)  # 队列满了会自动阻塞
        print(f"生产: {item}")
        time.sleep(random.uniform(0.1, 0.5))

def consumer(name):
    while True:
        item = q.get()  # 队列空了会自动阻塞
        print(f"{name} 消费: {item}")
        q.task_done()   # 通知队列任务完成
        time.sleep(random.uniform(0.2, 0.6))

# 启动1个生产者，2个消费者
threading.Thread(target=producer, daemon=True).start()
threading.Thread(target=consumer, args=("消费者A",), daemon=True).start()
threading.Thread(target=consumer, args=("消费者B",), daemon=True).start()

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

Queue的三种变体：

类型	特点	适用场景
`queue.Queue`	先进先出（FIFO）	最常用，生产者消费者模型
`queue.LifoQueue`	后进先出（LIFO）	类似栈，适合深度优先搜索
`queue.PriorityQueue`	优先级队列	任务有优先级时使用

避坑指南：

我曾经在项目里忘记调用task_done()，结果q.join()一直阻塞，整个程序卡死了。记住：get()之后一定要task_done()，除非你用的是get_nowait()。

4.4 三种通信方式对比

特性	Condition	Event	Queue
复杂度	高	低	中
数据传递	需自行管理	不支持	内置支持
多次通知	支持	需手动clear	自动支持
线程安全	需配合锁	是	是
推荐场景	复杂条件等待	一次性信号	生产者消费者

我个人习惯是：能上Queue就上Queue。它把锁、条件变量、数据传递都封装好了，不容易出错。只有遇到特别复杂的条件判断，我才会手写Condition。

选型建议：

只是通知「可以开始了」→ 用Event
需要传递数据 → 用Queue
条件判断复杂，需要精细控制 → 用Condition

嗯，这三种方式基本覆盖了日常开发中90%的线程通信场景。我个人建议：先从Queue入手，它最不容易出错。等你对多线程有了更深的理解，再尝试Condition的精细控制。