3、数据流与队列：生产者-消费者模式、queue.Queue的使用、多线程数据采集与动画刷新的解耦

上一章我们让采集线程和动画线程各自跑起来了。但你很快会发现一个问题——两个线程直接共享一个列表，动不动就报错，或者数据读到一半被改了。我当年第一次做实时数据可视化时就踩过这个坑：采集线程往列表里写，动画线程从列表里读，结果界面卡死不说，数据还乱成一锅粥。

问题的根源在于——多线程直接操作共享数据，没有协调机制。说白了，就是两个人在同一张纸上写字，一个写一个擦，谁也看不清。今天我们要聊的，就是怎么用队列把这个乱局理顺。

3.1 生产者-消费者模式：一个经典的设计思路

先别急着写代码。我们想想现实中的场景：面包店的后厨做面包，前台卖面包。后厨做好了就放到货架上，前台从货架上拿给顾客。后厨和前台不需要直接沟通，货架就是中间缓冲区。

这就是生产者-消费者模式的核心思想。在我们的场景里：

• 生产者：数据采集线程，负责从传感器/API/文件读取数据
• 消费者：动画刷新线程，负责把数据绘制到界面上
• 货架：一个线程安全的队列，存放待处理的数据

我习惯把这个模式画成一条流水线：

你看，生产者只管往队列里放数据，消费者只管从队列里取数据。两个线程不再直接操作同一个列表，而是通过队列这个中间人打交道。这就是解耦——一个改了，另一个不用跟着改。

3.2 queue.Queue：Python自带的线程安全队列

Python标准库里的queue.Queue，就是为我们这种场景量身定做的。它内部已经处理好了锁机制，你不需要自己写threading.Lock。我刚开始用的时候还担心性能，后来压测发现，每秒几千次put/get完全没问题。

常用的方法就三个：

方法	作用	阻塞行为
put(item)	往队列尾部放一个数据	队列满时阻塞，直到有空位
get()	从队列头部取一个数据	队列空时阻塞，直到有数据
qsize()	返回当前队列长度	非阻塞，近似值

还有一个很实用的参数——maxsize。设置队列最大长度，可以防止生产者生产太快，把内存撑爆。我一般设成100或200，够用又不占太多内存。

3.3 实战：用队列改造我们的采集-动画系统

好，理论说完了，我们直接上代码。这是改造后的数据采集线程：

import queue
import threading
import time
import random

# 创建一个队列，最大长度100
data_queue = queue.Queue(maxsize=100)

def data_collector():
    """生产者：数据采集线程"""
    for i in range(100):
        # 模拟采集数据
        value = random.uniform(20.0, 30.0)
        timestamp = time.time()
        data_point = {'time': timestamp, 'value': value}
        
        # 放入队列
        data_queue.put(data_point)
        print(f"[采集] 放入数据: {value:.2f}, 队列大小: {data_queue.qsize()}")
        
        # 模拟采集间隔
        time.sleep(0.1)
    
    # 发送结束信号
    data_queue.put(None)
    print("[采集] 采集完成，发送结束信号")

这是改造后的动画刷新线程：

def animation_updater():
    """消费者：动画刷新线程"""
    while True:
        # 从队列取数据（阻塞等待）
        data_point = data_queue.get()
        
        # 检查结束信号
        if data_point is None:
            print("[动画] 收到结束信号，退出")
            break
        
        # 处理数据并刷新动画
        value = data_point['value']
        timestamp = data_point['time']
        print(f"[动画] 绘制数据: {value:.2f} @ {timestamp:.2f}")
        
        # 模拟动画绘制耗时
        time.sleep(0.05)
        
        # 标记任务完成（用于join）
        data_queue.task_done()

启动两个线程：

# 创建线程
t1 = threading.Thread(target=data_collector)
t2 = threading.Thread(target=animation_updater)

# 启动
t1.start()
t2.start()

# 等待采集线程结束
t1.join()

# 等待队列中的所有数据被消费完
data_queue.join()

print("所有数据采集和动画刷新完成")

3.4 避坑指南：我曾经踩过的坑

用队列看起来简单，但有几个细节不注意，照样翻车。我一个个说：

1. 忘记处理结束信号：消费者线程会一直阻塞在get()上，程序退不了。我习惯用None作为哨兵值，生产者放完数据后放一个None，消费者收到就退出。
2. 队列满导致生产者阻塞：如果消费者处理太慢，队列满了，生产者会被卡住。这时候可以设maxsize大一点，或者用put_nowait()加异常处理。
3. 数据积压太多：动画刷新是实时的，如果队列里积压了几百条数据，动画会一直追赶，看起来像快进。我一般会检查qsize()，如果超过阈值就丢弃旧数据。

3.5 解耦带来的好处

用了队列之后，你会发现代码结构清晰了很多。采集线程和动画线程各干各的，互不依赖。想改采集频率？改采集线程就行，动画那边不用动。想换动画库？改动画线程就行，采集那边不用管。

我去年做一个工业监控项目，传感器数据每秒来1000条，动画只需要每秒刷新30帧。用队列做缓冲，采集线程只管往队列里塞，动画线程按自己的节奏取。后来客户要求增加数据存储功能，我加了一个消费者线程从队列里取数据写数据库，完全不影响原有的采集和动画逻辑。

这就是解耦的魅力——每个线程只关心自己的事，队列是它们之间的契约。

3.6 本章小结

今天我们聊了三个核心内容：

• 生产者-消费者模式：用队列解耦数据采集和动画刷新
• queue.Queue的使用：put、get、task_done、join的配合
• 多线程数据采集与动画刷新的解耦：各自独立，通过队列协作