共享厨房：线程之间怎样聊天？

开篇除恐

当人们谈到“线程通信”，往往会立刻想到复杂的锁、信号量、条件变量，甚至还有让人头疼的“死锁”和“竞态条件”。听起来像是要先考取一个并发编程的博士学位才能安全地让两个线程说一句话。其实，线程之间的聊天并没有那么神秘——它们不过是共用同一块厨房（内存），通过约定好的“信号”和“工具”来协调谁什么时候能用哪样东西。只要把共享内存想象成厨房里的案板、调料罐和炉灶，线程就像厨师们，只要大家遵守同样的使用规则（比如谁在切菜时别人别动刀子，谁在炒菜时别人别抢锅），就能和平共处、高效完成菜肴。

白话化

共享内存（Shared Memory）：同一进程的所有线程都能直接读写的内存区域（全局变量、堆分配的对象等）。这是线程之间最快速的通信方式，因为不需要内核拷贝或系统调用——只要一个线程改了值，另一个线程立刻就能看到（前提是看得到的时机满足可见性要求，这会在第九章再细说）。
同步原语（Synchronization Primitives）：用来保证对共享内存的访问是有序的，防止两个线程同时修改同一个变量导致数据混乱。常见的有：
- 互斥锁（Mutex）：像厨房里的“案板使用权”标志——只有拿到锁的线程才能切菜，其他线程必须等它释放锁后才能拿到。
- 条件变量（Condition Variable）：像厨师之间的“唱叫”——“菜已经切好了，谁来炒？”一个线程可以在条件不满足时等待，另一个线程在完成工作后通知等待者。
- 信号量（Semaphore）：更一般的计数器控制工具，可用于限制同时访问某资源的线程数量（比如同时只有两个人能用炉灶）。
- 原子操作（Atomic）：对单个变量的读写在硬件层面保证不可分割，适用于简单的计数或标志位，省去了加锁的开销（在 Python 中可用 threading.Lock 的简单情形或使用 queue.Queue 等）。

这些工具的核心思想都是：先约定好谁什么时候能用哪件东西，然后严格执行约定。只要约定清晰、执行到位，线程之间的“聊天”就会既快又安全。

直觉先行

想象你们俩在同一间厨房准备晚餐： - 你负责切洋葱，我负责炒肉。 - 洋葱切好之前，我不能往锅里放肉，否则会先把肉炒焦；肉炒好之前，你也不应该把洋葱倒进锅里，否则会蒸水导致不香。 - 于是我们约定：
1. 你切完洋葱后，把切好的洋葱放在案板旁的盘子里，并敲一下锅盖作为信号（“洋葱好了！”）。
2. 我听到敲锅盖的声音后，才把肉放进锅里炒；炒完肉后，我也会敲一下案板（“肉好了！”），告诉你可以把洋葱倒进锅里翻炒。
- 如果谁没遵守约定——比如我在你还没敲锅盖时就把肉下锅——那就可能导致“生肉”或“焦肉”，也就是我们说的竞态条件。

在程序里，案板上的盘子就是共享内存（比如一个布尔变量 onion_ready），敲锅盖就是互斥锁或条件变量的通知。通过这些简单的工具，线程就能像厨师一样有条不紊地完成复杂的菜肴。

例子贴身

例子 1（☼ 热身）：用互斥锁保护全局计数器

任务：让 10 条线程各自对同一个全局整型 counter 加 1 万次，最后得到准确的 100 000。

怎么想到的：如果不加保护，多个线程可能同时读取同一个旧值、各自加一后再写回，导致有些加法被丢失（这就是竞态条件）。我们给 counter 配一个互斥锁 lock；每次要改动 counter 时，先把锁锁上（acquire），改完再解锁（release）。这样保证同一时刻只有一条线程能访问 counter。

示例代码（Python）

import threading

lock = threading.Lock()
counter = 0

def worker():
    global counter
    for _ in range(10000):  # 每人加 10000 次
        with lock:          # 自动加锁/解锁（RAII 风格）
            counter += 1

def main():
    threads = []
    for _ in range(10):     # 雇佣 10 个帮手
        t = threading.Thread(target=worker)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()            # 等大家都忙完
    print(f"Final counter = {counter}")  # 期望 100000

if __name__ == "__main__":
    main()

运行结果

Final counter = 100000

要点：
- with lock: 是 Python 的 RAII 风格锁包装器，进入时加锁，离开时自动解锁，即便在块中抛出异常也不会忘记解锁。
- 没有锁的时候，你可能会看到类似 98732、99104 这样的错误结果；加锁后结果恒准确。

例子 2（☼☼ 正经）：生产者‑消费者模型（用条件变量实现“有货则消费，无货则等待”）

任务：一个生产者线程不断往一个有固定大小的缓冲区（比如大小为 5 的环形队列）里放入数字；多个消费者线程从缓冲区取出数字并打印。当缓冲区满时，生产者等待；当缓冲区空时，消费者等待。

怎么想到的：我们用一个列表 buffer 作为环形队列，max_size = 5，buffer 存放实际元素；使用索引 in_idx、out_idx 和计数 size 表示状态。互斥锁 mutex 保护对 buffer、size、in_idx、out_idx 的访问；两个条件变量：not_empty（当 size > 0 时可消费）和 not_full（当 size < max_size 时可生产）。生产者在放入元素后通知 not_empty；消费者在取出元素后通知 not_full。

示例代码（Python）

import threading
import time

class BoundedBuffer:
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.buffer = [None] * capacity
        self.in_idx = 0
        self.out_idx = 0
        self.size = 0
        self.mutex = threading.Lock()
        self.not_empty = threading.Condition(self.mutex)
        self.not_full = threading.Condition(self.mutex)

    def produce(self, item):
        with self.not_full:
            while self.size >= self.capacity:
                self.not_full.wait()
            self.buffer[self.in_idx] = item
            self.in_idx = (self.in_idx + 1) % self.capacity
            self.size += 1
            self.not_empty.notify()

    def consume(self):
        with self.not_empty:
            while self.size <= 0:
                self.not_empty.wait()
            item = self.buffer[self.out_idx]
            self.out_idx = (self.out_idx + 1) % self.capacity
            self.size -= 1
            self.not_full.notify()
            return item

def producer(buf, pid):
    for i in range(20):
        buf.produce(i)
        print(f"Producer {pid} produced {i}")
        time.sleep(0.03)  # 模拟生产间隔

def consumer(buf, cid):
    for _ in range(20):
        item = buf.consume()
        print(f"Consumer {cid} consumed {item}")
        time.sleep(0.05)  # 模拟消费间隔

def main():
    buf = BoundedBuffer(5)
    p1 = threading.Thread(target=producer, args=(buf, 1))
    c1 = threading.Thread(target=consumer, args=(buf, 1))
    c2 = threading.Thread(target=consumer, args=(buf, 2))
    p1.start()
    c1.start()
    c2.start()
    p1.join()
    c1.join()
    c2.join()

if __name__ == "__main__":
    main()

运行结果摘录（生产和消费交替进行，缓冲区永不越界）

Producer 1 produced 0
Consumer 1 consumed 0
Consumer 2 consumed 1
Producer 1 produced 1
...

要点：
- 条件变量的典型用法是：持有互斥锁时检查谓词（比如 size < CAPACITY），若不满足则 wait 释放锁并挂起；被 notify 唤醒后重新获取锁并再次检查谓词。
- 这样既避免了“虚假唤醒”（spurious wakeup）导致的错误，也确保了对共享状态的独占访问。
- 生产者‑消费者模式是许多并发场景的基础（日志系统、任务队列、音视频流缓冲等），掌握它后你就能够构建更复杂的线程协作。

收尾

这一章要带走的东西
- 线程之间最直接的通信方式是共享内存（全局变量、堆对象等），读写速度快但需要同步来防止数据竞争。
- 常用的同步原语包括互斥锁（Lock）、条件变量（Condition）、信号量（Semaphore，可通过 threading.Semaphore 实现）以及原子操作（在 Python 中可通过 queue.Queue 或 threading.Lock 的简单情形模拟），它们各自对应不同的协作场景。
- 使用这些工具的核心思想是：先约定好谁什么时候能访问哪份数据，然后通过加锁/等待/通知严格执行约定。
- 掌握了共享内存与同步的基本模式后，你就能够安全地让多个线程围着同一份“食材”工作，而不会把厨房弄成一团糟。
- 接下来我们将进入线程使用时最常见的“痛点”——竞态条件、死锁和内存可见性问题，并学习如何用工具和习惯来避免它们。

就这样。 下一章我们揭开“抢锅铲的灾难”：到底什么是竞态条件，为什么它会让正确的程序产生错乱的结果？