多进程自更新对象阻塞问题：高刷新率显示如何实时获取硬件最新值

看起来你碰到了多进程同步里的经典坑——慢节奏的硬件读取进程把高刷新率的显示进程给拖慢了。核心问题应该是你当前的共享状态（管道）用了阻塞式的交互逻辑，导致显示进程每次都得等硬件进程更新完才能拿到数据，而不是直接抓取最新的已更新值。我给你几个实际可落地的解决思路：

一、用非阻塞队列丢弃旧值，只取最新数据 #

最直接的方案是用multiprocessing.Queue，让硬件进程不断把最新的传感器数据塞进队列，显示进程每次刷新时，先把队列里所有旧数据清空，只取最后一条（也就是当前最新的硬件值）。这样显示进程完全不受硬件更新速度的限制，想刷多快就刷多快。

代码示例： #

import multiprocessing
import board
import busio
import adafruit_lis2mdl
import adafruit_lsm303_accel
import time

class AccMag():
    def init(self):
        self.i2c = busio.I2C(board.SCL, board.SDA)
        self.mag = adafruit_lis2mdl.LIS2MDL(self.i2c)
        self.accel = adafruit_lsm303_accel.LSM303_Accel(self.i2c)

    def read_sensors(self):
        # 读取最新的传感器数据
        mag_val = self.mag.magnetic
        accel_val = self.accel.acceleration
        return mag_val, accel_val

# 硬件读取进程的工作函数
def hardware_worker(queue):
    acc_mag = AccMag()
    acc_mag.init()
    while True:
        # 读取传感器数据，塞进队列
        data = acc_mag.read_sensors()
        queue.put(data)
        # 硬件本身需要的更新间隔，可根据实际调整
        time.sleep(0.1)

# 显示进程的工作函数
def display_worker(queue):
    while True:
        # 清空队列里的旧数据，只保留最后一条最新的
        latest_data = None
        while not queue.empty():
            latest_data = queue.get(block=False)
        
        if latest_data is not None:
            mag_val, accel_val = latest_data
            # 这里执行你的显示更新逻辑
            print(f"显示更新：磁强计={mag_val}, 加速度计={accel_val}")
        
        # 模拟高刷新率的显示循环（比如60帧/秒）
        time.sleep(1/60)

if __name__ == "__main__":
    # 创建队列，设置最大长度避免内存溢出（可选）
    sensor_queue = multiprocessing.Queue(maxsize=10)
    
    # 启动两个进程
    hardware_proc = multiprocessing.Process(target=hardware_worker, args=(sensor_queue,))
    display_proc = multiprocessing.Process(target=display_worker, args=(sensor_queue,))
    
    hardware_proc.start()
    display_proc.start()
    
    hardware_proc.join()
    display_proc.join()

二、用共享内存+轻量锁实现快速读写 #

如果你不想用队列，也可以直接用共享内存变量（比如multiprocessing.Value/Array）配合锁，确保硬件进程更新数据和显示进程读取数据时的原子性，但要注意锁的粒度必须极小——只在读写共享变量的瞬间持有锁，绝对不要在锁里做硬件读取或显示渲染的操作。

代码示例： #

import multiprocessing
import ctypes
import time
import board
import busio
import adafruit_lis2mdl
import adafruit_lsm303_accel

class AccMag():
    def init(self):
        self.i2c = busio.I2C(board.SCL, board.SDA)
        self.mag = adafruit_lis2mdl.LIS2MDL(self.i2c)
        self.accel = adafruit_lsm303_accel.LSM303_Accel(self.i2c)

    def read_sensors(self):
        mag_val = self.mag.magnetic
        accel_val = self.accel.acceleration
        return mag_val, accel_val

def hardware_worker(shared_mag, shared_accel, lock):
    acc_mag = AccMag()
    acc_mag.init()
    while True:
        mag_val, accel_val = acc_mag.read_sensors()
        # 加锁更新共享变量，更新完立刻释放
        with lock:
            shared_mag[0] = mag_val[0]
            shared_mag[1] = mag_val[1]
            shared_mag[2] = mag_val[2]
            shared_accel[0] = accel_val[0]
            shared_accel[1] = accel_val[1]
            shared_accel[2] = accel_val[2]
        # 硬件更新间隔
        time.sleep(0.1)

def display_worker(shared_mag, shared_accel, lock):
    while True:
        current_mag = [0,0,0]
        current_accel = [0,0,0]
        # 加锁读取共享变量，读完立刻释放
        with lock:
            current_mag[0] = shared_mag[0]
            current_mag[1] = shared_mag[1]
            current_mag[2] = shared_mag[2]
            current_accel[0] = shared_accel[0]
            current_accel[1] = shared_accel[1]
            current_accel[2] = shared_accel[2]
        
        # 执行显示更新
        print(f"显示更新：磁强计={current_mag}, 加速度计={current_accel}")
        # 高刷新率间隔
        time.sleep(1/60)

if __name__ == "__main__":
    # 创建共享数组（存储三维数据）
    shared_mag = multiprocessing.Array(ctypes.c_float, 3)
    shared_accel = multiprocessing.Array(ctypes.c_float, 3)
    # 创建锁
    data_lock = multiprocessing.Lock()
    
    hardware_proc = multiprocessing.Process(target=hardware_worker, args=(shared_mag, shared_accel, data_lock))
    display_proc = multiprocessing.Process(target=display_worker, args=(shared_mag, shared_accel, data_lock))
    
    hardware_proc.start()
    display_proc.start()
    
    hardware_proc.join()
    display_proc.join()

三、排查你当前阻塞的核心原因 #

你提到用了管道实现共享状态，大概率是显示进程用了pipe.recv()这种阻塞式调用——它会一直等硬件进程发数据，自然就被硬件的更新速度卡住了。如果非要用管道，得改成非阻塞的pipe.poll()先检查有没有数据，有就全部读完取最后一条，和队列的思路类似，但队列已经帮你封装好了这个逻辑，没必要自己造轮子。

另外要注意：硬件读取的循环里不要有过长的阻塞操作，显示进程的循环里也绝对不能有任何会拖慢帧率的阻塞代码（比如同步等待硬件数据）。

备注：内容来源于stack exchange，提问作者Sarazil