27、面向对象与并发：线程安全的对象设计，锁与条件变量，生产者-消费者模式

并发编程，说白了就是让程序同时干多件事。但对象一旦被多个线程同时访问，就容易出乱子。我早年做后台服务时，就吃过这个亏——一个共享的计数器，两个线程同时加一，结果丢数据了。嗯，从那以后，我对线程安全就特别上心。

一、线程安全的对象设计原则

一个对象要线程安全，核心就一句话：多个线程同时访问它，不需要外部加锁，行为也是正确的。怎么做到？我总结了几条经验。

1. 封装可变状态

把可能被并发修改的成员变量藏起来，只通过同步方法暴露。你想想看，如果所有字段都是 public 的，别人随便改，你怎么保证安全？

2. 使用不可变对象

如果一个对象创建后状态再也不变，那它天然就是线程安全的。我个人习惯，能设计成不可变的类，绝不设计成可变的。比如 String、Integer 这些，随便多少个线程读，都不会出问题。

3. 加锁保护临界区

如果必须修改共享状态，那就加锁。锁的本质，就是让同一时刻只有一个线程能进入临界区。

二、锁与条件变量

锁是并发编程的基础工具。Python 里最常用的是 threading.Lock 和 threading.RLock。我刚开始用的时候，经常忘记释放锁，后来养成了用 with 语句的习惯，再也没漏过。

基本锁的使用

import threading

lock = threading.Lock()
shared_counter = 0

def increment():
    global shared_counter
    with lock:
        shared_counter += 1

# 启动10个线程
threads = [threading.Thread(target=increment) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(shared_counter)  # 结果一定是10

你看，用 with lock 包裹，进入时自动获取锁，退出时自动释放。省心。

条件变量：让线程学会等待

有时候，一个线程需要等某个条件成立才能继续。比如生产者生产了数据，消费者才能消费。这时候就需要条件变量 threading.Condition。

条件变量内部其实有一把锁。调用 wait() 会释放锁并阻塞，直到被 notify() 或 notify_all() 唤醒。唤醒后，它会重新获取锁再继续执行。

三、生产者-消费者模式

这是并发编程里最经典的模型。生产者负责生产数据，消费者负责处理数据。中间用队列做缓冲，解耦两者。

我当年做一个日志收集系统，就是用这个模式。多个生产者线程写日志，一个消费者线程批量写入磁盘。效率高，还不会丢数据。

用队列实现

Python 的 queue.Queue 本身就是线程安全的，内部已经处理好了锁和条件变量。直接用就行。

import threading
import queue
import time
import random

# 共享队列，最大容量5
q = queue.Queue(maxsize=5)

def producer(name):
    for i in range(10):
        item = f"{name}-{i}"
        q.put(item)  # 如果队列满了，会自动阻塞
        print(f"生产者 {name} 生产了 {item}")
        time.sleep(random.random())

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

# 启动2个生产者，3个消费者
for i in range(2):
    threading.Thread(target=producer, args=(f"P{i}",), daemon=True).start()
for i in range(3):
    threading.Thread(target=consumer, args=(f"C{i}",), daemon=True).start()

# 等待所有生产完成
time.sleep(5)
print("主线程结束")

你看，代码很简洁。生产者只管 put，消费者只管 get，队列自己处理同步。这就是面向对象封装的好处——你不用关心锁怎么加，条件变量怎么用，队列内部都帮你搞定了。

自己实现一个简易版

为了让你理解原理，我写一个用条件变量实现的简易队列。嗯，这里要注意，实际开发直接用 queue.Queue 就好，但理解底层有助于你排查问题。

import threading

class SimpleQueue:
    def __init__(self, maxsize=0):
        self.maxsize = maxsize
        self.items = []
        self.mutex = threading.Lock()
        self.not_full = threading.Condition(self.mutex)
        self.not_empty = threading.Condition(self.mutex)

    def put(self, item):
        with self.not_full:
            while self.maxsize > 0 and len(self.items) >= self.maxsize:
                self.not_full.wait()  # 队列满了，等待
            self.items.append(item)
            self.not_empty.notify()   # 通知消费者

    def get(self):
        with self.not_empty:
            while len(self.items) == 0:
                self.not_empty.wait()  # 队列空了，等待
            item = self.items.pop(0)
            self.not_full.notify()     # 通知生产者
            return item

核心逻辑就两个：满了等，空了等。生产者发现队列满了，就在 not_full 上等；消费者发现队列空了，就在 not_empty 上等。生产一个，通知消费者；消费一个，通知生产者。

四、知识体系图

下面这张图总结了本章的核心知识点。你可以看到，线程安全设计、锁与条件变量、生产者-消费者模式，这三者层层递进。

五、避坑总结

最后，我把自己踩过的坑列出来，你写代码时多留个心眼。

常见问题	原因	解决方案
忘记释放锁	手动 lock() 后没有 unlock()	用 with 语句自动管理锁
条件变量虚假唤醒	wait() 返回后条件可能不成立	把 wait() 放在 while 循环里
死锁	多个锁获取顺序不一致	固定锁的获取顺序，或使用 RLock
队列无界导致内存溢出	生产者速度远快于消费者	设置 maxsize 限制队列长度

好了，这一章就到这里。线程安全的设计思想，说白了就是「管好你的共享状态」。锁和条件变量是工具，生产者-消费者是经典模式。把这些用熟了，并发编程就没那么可怕了。

---

公众号:蓝海资料掘金营，微信deep3321