3、生产者-消费者模型：多线程环境下的日志写入与消费
聊到日志系统的内存管理，有一个模式你肯定绕不开——生产者-消费者模型。说白了，就是一边有线程拼命写日志（生产者），另一边有线程负责把日志刷到磁盘（消费者）。这两者之间怎么协调，直接决定了你的系统会不会丢数据。
我个人习惯把这个模型比作「流水线」。前端工人（生产者）只管往传送带上放零件，后端工人（消费者）只管从传送带上取零件。传送带本身，就是那个关键的缓冲区。嗯，这里要注意：传送带如果太短，前端工人就得停下来等；如果太长，又占地方。这个度怎么把握，就是我们今天要聊的核心。
3.1 为什么非要用这个模型？
你想想看，如果每个业务线程都直接写磁盘，会怎样？
- 磁盘 I/O 是慢操作，一个 write 调用可能阻塞几十毫秒
- 高并发下，大量线程同时写磁盘，磁盘寻道时间爆炸
- 锁竞争激烈，性能直线下降
我在项目中遇到过这种情况：一个支付系统，高峰期每秒处理 5000 笔交易。每个交易都要写日志，如果直接写磁盘，CPU 大部分时间都在等 I/O，业务延迟直接飙到秒级。后来改成生产者-消费者模型，用内存缓冲区做中转，吞吐量提升了 10 倍以上。
3.2 核心组件：缓冲区
缓冲区是生产者和消费者之间的「共享内存」。Python 里常用的实现方式有几种：
| 实现方式 | 特点 | 适用场景 |
|---|---|---|
queue.Queue |
线程安全，自带阻塞机制 | 大多数通用场景 |
collections.deque + 锁 |
性能更高，需手动管理锁 | 高吞吐、低延迟场景 |
| 环形缓冲区（ring buffer） | 固定大小，无内存分配开销 | 极致性能，嵌入式或高频交易 |
我个人最常用的是 queue.Queue。为什么？因为它内置了锁和条件变量，你不需要自己处理那些容易出错的细节。我曾经手写过基于 deque 的缓冲区，结果因为一个锁没释放，线上服务卡死了 3 分钟。从那以后，除非性能真的不够，否则我绝不自己造轮子。
3.3 代码示例：一个简单的日志生产者-消费者
import threading
import queue
import time
import logging
# 配置日志
logging.basicConfig(level=logging.INFO)
class LogProducer:
"""日志生产者：业务线程调用"""
def __init__(self, buffer: queue.Queue):
self.buffer = buffer
def emit(self, log_msg: str):
# 关键：设置超时，防止无限阻塞
try:
self.buffer.put(log_msg, timeout=0.1)
except queue.Full:
# 缓冲区满了，这里就是丢数据的风险点
logging.warning("缓冲区已满，日志丢失: %s", log_msg[:50])
class LogConsumer:
"""日志消费者：后台线程刷盘"""
def __init__(self, buffer: queue.Queue, batch_size: int = 100):
self.buffer = buffer
self.batch_size = batch_size
self.running = True
def run(self):
while self.running:
batch = []
# 批量取出，减少 I/O 次数
for _ in range(self.batch_size):
try:
msg = self.buffer.get(timeout=0.5)
batch.append(msg)
except queue.Empty:
break
if batch:
self._flush(batch)
def _flush(self, batch):
# 实际写入磁盘
with open("app.log", "a") as f:
for msg in batch:
f.write(msg + "\n")
def stop(self):
self.running = False
# 使用示例
buffer = queue.Queue(maxsize=10000)
producer = LogProducer(buffer)
consumer = LogConsumer(buffer)
consumer_thread = threading.Thread(target=consumer.run, daemon=True)
consumer_thread.start()
# 模拟业务线程写日志
for i in range(1000):
producer.emit(f"日志条目 {i}")
time.sleep(0.001)
consumer.stop()
consumer_thread.join()
关键点：缓冲区大小 maxsize=10000 不是随便设的。它取决于你的业务峰值写入速度和消费者处理速度。一般建议设为「峰值写入速率 × 消费者最大延迟」的 1.5 倍。
3.4 避坑指南：丢数据的三种情况
我曾经在生产环境踩过这些坑，现在分享给你：
- 缓冲区满了直接丢弃——上面的代码里，
put(timeout=0.1)超时后日志就丢了。这是最隐蔽的丢数据方式。 - 消费者崩溃未处理——消费者线程如果异常退出，缓冲区里的日志全没了。我建议加一个
try-except-finally兜底，确保退出前把缓冲区清空。 - 进程被强制杀掉——内存里的数据还没刷盘，进程就挂了。这个无解，只能靠双写或 WAL（预写日志）来保证。
警告：千万不要在生产环境用 queue.Queue 的默认无限大小。内存会爆的。我见过一个同事，线上服务跑了 3 天，内存从 256MB 涨到 8GB，最后 OOM 被 kill 了。原因就是缓冲区无限增长，消费者处理不过来。
3.5 核心逻辑流程图
下面这张图展示了生产者-消费者模型的核心数据流：
小技巧：如果你对性能要求极高，可以试试用 threading.Condition 配合 deque 实现一个无锁（或少锁）的环形缓冲区。但说实话，除非你每秒要处理几十万条日志，否则 queue.Queue 完全够用。别过度优化。
3.6 如何避免丢数据？
嗯，这里要讲点实战经验了。我总结了三层防护：
- 第一层：监控缓冲区水位——当缓冲区使用量超过 80% 时，发出告警。这样你可以在丢数据之前介入。
- 第二层：降级策略——缓冲区快满时，可以临时把日志级别从 DEBUG 提升到 WARN，减少写入量。我在一个电商项目中用过这招，效果很好。
- 第三层：本地文件兜底——如果缓冲区真的满了，把日志写到本地临时文件。等消费者空闲了再合并进去。这相当于给系统加了个「安全气囊」。
说白了，生产者-消费者模型的核心就是「解耦」和「缓冲」。但缓冲本身是一把双刃剑——它带来了性能，也带来了丢数据的风险。你需要根据业务场景，在「性能」和「可靠性」之间找到那个平衡点。
我个人习惯在项目初期就把监控和降级策略写好。因为一旦线上出了丢数据的问题，再回头加这些逻辑，往往已经晚了。你想想看，日志系统本身就是为了排查问题而存在的，如果它自己先丢了数据，那可就尴尬了。