线程只读场景下用全局变量/字典传递变量是否为不良实践？

一、用字典/全局变量的潜在问题 #

即使主线程只读取不修改，用字典或全局变量存储线程数据仍存在明显风险：

• 可见性问题：多CPU核心环境下，线程写入的数据可能被缓存到核心本地内存，主线程无法及时获取最新值，导致打印的是过期数据。
• 一致性隐患：当前代码是单个键值的赋值，看似原子，但如果后续需要同时读取多个关联值（比如同一批次的电压和电流），字典无法保证这两个值的一致性——可能主线程读到的电压是新值，电流却是旧值。
• 扩展性差：如果后续新增线程或数据类型，手动维护字典的同步逻辑很容易出错，代码复杂度会急剧上升。

二、最佳选择：使用Queue #

queue.Queue是Python标准库专为线程间通信设计的工具，自带线程安全的同步机制，完美解决上述问题：

• 自动处理内存可见性：写入操作会强制刷新内存，保证主线程能读取到最新数据。
• 保证数据一致性：可以把一组关联数据（比如电压和电流）打包成元组、列表或对象存入队列，主线程一次取出就能拿到同一批次的完整数据。
• 无需手动同步：队列内部已经实现了锁机制，不用额外编写threading.Lock这类同步代码，简洁又可靠。

三、修改后的示例代码 #

把原代码中的字典替换为maxsize=1的Queue（只保留最新数据）：

import serial
import struct
import threading
import readkrohne
import keyboard
import time
from queue import Queue

Krohnehigh = readkrohne.comport('COM4')
Krohnelow = readkrohne.comport('COM6')

CONST_SER_PORT = 'COM11'
serDataq = serial.Serial(
    port=CONST_SER_PORT,
    timeout=0.5
)

serDataq.reset_input_buffer()

numofchannel = 2
numofbyteperscan = 2 * numofchannel

def scan(output_queue):
    serDataq.write(b"start\r")
    while True:
        try:
            i = serDataq.in_waiting
            if (i // numofbyteperscan) > 0:
                response = serDataq.read(i - i % numofbyteperscan)
                count = (i - i % numofbyteperscan) // 2
                response2 = bytearray(response)
                Channel = struct.unpack("<" + "h" * count, response2)
                voltage = round((Channel[0]/32768*10), 3)
                current = round((Channel[1]/32768*10)*(1000/498.4), 3)
                # 队列满时移除旧数据，只保留最新值
                if not output_queue.empty():
                    output_queue.get_nowait()
                output_queue.put((voltage, current))
        except serial.SerialException as e:
            print(f"串口错误: {e}")
        except Exception as e:
            print(f"未知错误: {e}")

def kscan(krohne_queue):
    while True:
        high = Krohnehigh.FlowRate()
        low = Krohnelow.FlowRate()
        # 队列满时移除旧数据
        if not krohne_queue.empty():
            krohne_queue.get_nowait()
        krohne_queue.put((high, low))
        time.sleep(0.03)

# 初始化队列，最大容量1，只存最新数据
output_queue = Queue(maxsize=1)
krohne_queue = Queue(maxsize=1)
a = 0

t1 = threading.Thread(target=scan, args=(output_queue,))
t1.daemon = True
t1.start()

t2 = threading.Thread(target=kscan, args=(krohne_queue,))
t2.daemon = True
t2.start()

while True:
    if keyboard.is_pressed('x'):
        serDataq.write(b"stop\r")
        time.sleep(1)
        serDataq.close()
        print('Good-Bye')
        break

    a += 1
    # 读取队列数据，超时0秒避免阻塞
    krohne_data = krohne_queue.get_nowait() if not krohne_queue.empty() else (0, 0)
    output_data = output_queue.get_nowait() if not output_queue.empty() else (0, 0)
    print(f'{krohne_data[0]:.3f} {krohne_data[1]:.3f} {a:.3f} {output_data[0]:.3f} {output_data[1]:.3f}', end='\r')
    time.sleep(0.05)

补充说明 #

• 用maxsize=1的队列，每次写入前清空旧数据，确保主线程拿到的永远是最新的仪器读数。
• 替换了原代码中宽泛的except:，改为捕获具体异常，方便排查问题。
• 主线程读取队列时用get_nowait()并判断队列是否为空，避免因队列无数据而阻塞。

内容的提问来源于stack exchange，提问作者Isaac Zaiek