4、实时数据流处理：使用Python生成模拟遥测数据、生产者-消费者模式、队列（Queue）与多线程基础
好，咱们进入第四章。这一章，说白了就是解决一个核心问题：数据怎么源源不断地流进来，同时系统还不卡死？
我在做第一个遥测项目时，犯过一个低级错误——直接用单线程循环去收数据、存数据、画图。结果呢？数据一多，界面直接冻住，鼠标都点不动。嗯，从那以后我就明白了，实时系统必须用多线程+队列。
4.1 为什么需要生产者-消费者模式？
你想想看，遥测数据采集和数据处理是两个节奏完全不同的任务。
- 生产者：传感器、模拟器，它们按固定频率往外吐数据。比如每秒100条。
- 消费者：数据清洗、存储、可视化，这些操作耗时不确定。有时候快，有时候慢。
如果让生产者直接调用消费者，那生产者就得等消费者干完活才能继续。这就好比流水线上，包装工动作慢了，前面的传送带就得停——整个产线效率全废。
生产者-消费者模式，就是在这两者之间加一个缓冲区。生产者只管往缓冲区里丢数据，消费者只管从缓冲区里取数据。两者互不等待，各干各的。
核心思想：解耦生产与消费，用队列做缓冲，让系统能应对突发流量。
4.2 Python中的Queue：线程安全的缓冲区
Python标准库里的 queue.Queue，就是专门干这个的。它内部实现了锁机制，多个线程同时往里放、往外取，不会乱套。
我个人习惯用 Queue(maxsize=0)，不设上限。但在生产环境中，我建议你设一个合理的上限，比如 maxsize=1000。为什么？防止内存被撑爆。我曾经遇到过模拟器疯狂发包，队列积压了几十万条，内存直接飙到几个G……
| 方法 | 说明 | 阻塞行为 |
|---|---|---|
put(item, block=True, timeout=None) |
放入数据 | 队列满时阻塞，直到有空位或超时 |
get(block=True, timeout=None) |
取出数据 | 队列空时阻塞，直到有数据或超时 |
qsize() |
当前队列长度 | 近似值，不保证精确 |
task_done() |
通知队列任务完成 | 配合 join() 使用 |
小技巧：调试时可以用 queue.get_nowait() 和 queue.put_nowait()，它们不会阻塞，队列空或满时会直接抛出异常。方便你快速定位问题。
4.3 多线程基础：别让主线程闲着
Python的 threading 模块，用起来很简单。但有几个坑，我得提前跟你说。
第一个坑：GIL（全局解释器锁）
Python的多线程，在CPU密集型任务上其实是个假并行。因为GIL同一时刻只允许一个线程执行字节码。但咱们做的是I/O密集型任务——网络收发、文件读写、界面刷新——这些操作会释放GIL，所以多线程完全够用。
第二个坑：线程安全
多个线程同时修改同一个变量，结果不可预测。解决办法就是加锁，或者直接用 Queue——它内部已经帮你加好了。
来看一个最简单的多线程例子：
import threading
import time
def worker(name):
for i in range(5):
print(f"线程 {name} 正在工作...")
time.sleep(0.5)
# 创建线程
t1 = threading.Thread(target=worker, args=("A",))
t2 = threading.Thread(target=worker, args=("B",))
# 启动线程
t1.start()
t2.start()
# 等待线程结束
t1.join()
t2.join()
print("所有线程执行完毕")嗯，这里要注意：join() 是必须的。如果不加，主线程可能直接退出，子线程还没跑完就被强制结束了。
4.4 实战：模拟遥测数据的生产者-消费者
好，理论说完了，咱们直接上代码。这个例子会生成模拟的遥测数据——温度、压力、振动——然后通过队列传给消费者去处理。
import threading
import queue
import time
import random
import json
# 全局队列
data_queue = queue.Queue(maxsize=1000)
def generate_telemetry():
"""生产者：生成模拟遥测数据"""
while True:
data = {
"timestamp": time.time(),
"temperature": round(random.uniform(20.0, 80.0), 2),
"pressure": round(random.uniform(100.0, 200.0), 2),
"vibration": round(random.uniform(0.0, 5.0), 3)
}
try:
data_queue.put(data, timeout=1)
print(f"[生产] {data['timestamp']:.2f} 温度:{data['temperature']}")
except queue.Full:
print("[警告] 队列满了，丢弃一条数据")
time.sleep(0.1) # 模拟10Hz采集频率
def process_telemetry():
"""消费者：处理遥测数据"""
while True:
try:
data = data_queue.get(timeout=2)
# 模拟数据处理：比如计算平均值、存储到数据库
processed = {
"avg_temp": (data["temperature"] + random.uniform(-1, 1)),
"max_pressure": data["pressure"] * 1.1,
"status": "normal" if data["vibration"] < 3.0 else "warning"
}
print(f"[消费] 处理结果: {json.dumps(processed)}")
data_queue.task_done()
except queue.Empty:
print("[消费] 队列空了，等待数据...")
time.sleep(0.5)
# 启动生产者和消费者线程
producer = threading.Thread(target=generate_telemetry, daemon=True)
consumer = threading.Thread(target=process_telemetry, daemon=True)
producer.start()
consumer.start()
# 主线程保持运行
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
print("程序退出")避坑指南：我曾经把 daemon=True 漏掉了，结果程序退出时子线程还在跑，导致一堆僵尸线程。记住：守护线程会在主线程退出时自动结束，非守护线程会阻塞进程退出。
4.5 知识体系总览
这一章的内容，我用一张SVG图给你串起来。你看完应该能明白整个数据流的走向。
4.6 几个关键点再强调一下
- 队列大小要合理：太小容易丢数据，太大容易爆内存。根据你的数据量和处理速度来调。
- 超时机制不能省：
put(timeout=1)和get(timeout=2)能防止线程死等。 - 守护线程要设好：
daemon=True让子线程随主线程退出而结束，避免僵尸线程。 - 日志输出要带标记：生产者和消费者的日志加上 [生产] [消费] 前缀，方便排查问题。
我的经验：在实际项目中，我还会加一个监控线程，定期打印队列长度。如果队列一直增长，说明消费者处理速度跟不上生产者，需要扩容或者优化消费逻辑。
好了，这一章就到这里。你把这个生产者-消费者的架子搭好，后面做可视化、做报警，都往这个管道里接就行。数据流通了，整个系统就活了。