实时动画场景下的生产者-消费者线程时序问题排查

• 自定义锁逻辑混乱：Pipeline初始化时直接抢占flag_get_lock和consumer_lock，后续方法重复执行acquire操作，导致线程死锁，动画线程被阻塞无法更新画面。
• 共享变量读写不安全：producer_get_flag直接读取未加锁的self.flag，多线程环境下会出现数据不一致问题。
• 串口线程逻辑错误：存在拼写错误（pipline→pipeline），且每次读取串口数据后等待标记位，导致线程阻塞或串口数据堆积。
• 动画线程逻辑矛盾：设置标记位为1后，反而判断标记位为1时使用旧数据，逻辑完全颠倒。

核心优化点 #

• 抛弃复杂的自定义锁，改用queue.Queue实现线程间数据传递，利用其内置锁机制避免死锁和线程安全问题。
• 串口线程持续读取解析数据，将最新数据放入固定长度队列（队列满时自动丢弃旧数据，只保留最新值）。
• 动画线程每帧更新时主动从队列获取最新数据，无新数据则沿用旧值，无需依赖标记位同步。
• 添加异常处理，避免串口读取异常导致线程崩溃。

修复后的完整代码 #

import serial
import struct
import pygame
from pygame.locals import QUIT, K_ESCAPE
import concurrent.futures
import queue

# 配置参数
TARGET_FPS = 60
TIME_STEP = 1.0 / TARGET_FPS

# 请提前初始化以下全局资源
# world = ...  # Box2D世界对象
# screen = ...  # Pygame窗口对象
# clock = pygame.time.Clock()
# robot1 = ...  # 机器人刚体对象

def readport(data_queue):
    try:
        with serial.Serial(
            'COM3',
            baudrate=3000000,
            bytesize=8,
            timeout=None,
            stopbits=serial.STOPBITS_ONE,
            parity=serial.PARITY_NONE,
            rtscts=True,
            dsrdtr=False
        ) as s:
            print('Opened', s.name)
            while True:
                # 读取16字节完整数据
                res = s.read(16)
                if len(res) != 16:
                    continue
                # 解析浮点数
                ypos = struct.unpack('f', res[0:4])[0]
                xpos = struct.unpack('f', res[4:8])[0]
                yamp = struct.unpack('f', res[8:12])[0]
                xamp = struct.unpack('f', res[12:16])[0]
                
                # 数据合法性校验
                if xamp != 5.0:
                    print('异常数据:', [xpos, ypos, xamp, yamp])
                    break
                
                # 存入队列，满则替换旧数据
                try:
                    data_queue.put([xpos, ypos], block=False)
                except queue.Full:
                    data_queue.get(block=False)
                    data_queue.put([xpos, ypos], block=False)
    except Exception as e:
        print('串口读取错误:', str(e))

def animation(data_queue):
    running = True
    prevx, prevy = 0, 0
    while running:
        # 处理Pygame退出事件
        for event in pygame.event.get():
            if event.type == QUIT or (event.type == KEYDOWN and event.key == K_ESCAPE):
                running = False
        
        # 获取最新位置数据
        try:
            xpos, ypos = data_queue.get(block=False)
            prevx, prevy = xpos, ypos
        except queue.Empty:
            xpos, ypos = prevx, prevy
        
        # 更新机器人位置与物理模拟
        robot1.position = b2Vec2(xpos, ypos + 12)
        world.Step(TIME_STEP, 10, 10)
        
        # 绘制画面
        screen.fill((255, 255, 255))
        for body in world.bodies:
            for fixture in body.fixtures:
                fixture.shape.draw(body, fixture)
        
        # 刷新屏幕并控制帧率
        pygame.display.flip()
        clock.tick(TARGET_FPS)
    
    pygame.quit()
    print('Done!')

if __name__ == '__main__':
    # 初始化队列，最大长度1，仅保留最新数据
    data_queue = queue.Queue(maxsize=1)
    
    # 启动线程池运行任务
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        executor.submit(readport, data_queue)
        executor.submit(animation, data_queue)

关键认知纠正 #

• 优先使用标准库同步原语：queue.Queue已内置线程安全机制，无需手动实现锁逻辑，大幅降低死锁风险。
• 适配数据频率差异：串口读取（100us/次）远快于动画更新（~16ms/次），让动画线程主动取最新数据，而非让串口线程等待动画标记，避免数据堆积或线程阻塞。
• 共享变量必须线程安全：所有跨线程的数据传递必须通过同步机制完成，禁止无保护的直接读写。
• 异常处理不可少：串口通信易出现数据不完整、端口断开等异常，捕获异常可避免单个线程崩溃导致整个程序挂掉。

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