树莓派Python串口监听丢包求助：多线程场景下消息丢失

问题背景 #

在树莓派上通过Python实现多线程任务（串口监听回复、IO监控、Web服务器等），已实现多线程运行，但串口监听过程中频繁丢失接收的消息。

测试发送代码（模拟设备每秒发消息） #

import time
from serial_config import serial_port2

counter = 0
aux = 0

while True:
    while aux < 10:
        # 修正原代码语法错误：移除多余的f"
        straux = "{AAF0T000L0I0M0?00000" + str(aux) + "}"
        serial_port2.write(straux.encode())
        print(straux)
        time.sleep(1)
        counter += 1
        aux += 1
    aux = 0

原树莓派监听代码（存在问题） #

import serial
import threading

# ... 其他初始化代码 ...

#### monitores serial port ###########################################
serial_port1 = serial.Serial(
        port='/dev/ttyUSB0',
        baudrate = 9600,
        parity=serial.PARITY_NONE,
        stopbits=serial.STOPBITS_ONE,
        bytesize=serial.EIGHTBITS,
        timeout=1
)

def Monitoring_SerialPort_thr():
    global serial_message_received
    while(True):
        serial_port1.nonblocking()  # 错误：反复切换非阻塞模式会导致读取异常
        serial_message_received = serial_port1.read(100)
        serial_message_received =  # 语法错误：空赋值
        message_validation(serial_message_received)
  
if __name__ == "__main__":
    Init()
    
    Flashing_LED_thread = threading.Thread(target = Flashing_LED_thr)
    Running_Time_thread = threading.Thread(target = Running_Time_thr)
    Monitoring_SerialPort_thread = threading.Thread(target = Monitoring_SerialPort_thr)
    Monitoring_ADC_thread = threading.Thread(target = Monitoring_ADC_thr)
    Monitoring_IO_thread = threading.Thread(target = Monitoring_IO_thr)
    Save_Config_File_thread = threading.Thread(target = Save_Config_File_thr)
    Save_Config_STATUS_thread = threading.Thread(target = Save_Config_STATUS_thr)
    Web_Server_thread = threading.Thread(target = Web_Server_thr)

    Flashing_LED_thread.start()
    Running_Time_thread.start()
    Monitoring_SerialPort_thread.start()
    Monitoring_ADC_thread.start()
    Monitoring_IO_thread.start()
    Save_Config_File_thread.start()
    Save_Config_STATUS_thread.start()
    Web_Server_thread.start()
          
    Flashing_LED_thread.join()
    Monitoring_SerialPort_thread.join()
    Monitoring_ADC_thread.join()
    Monitoring_IO_thread.join()
    Running_Time_thread.join()
    Save_Config_File_thread.join()
    Save_Config_STATUS_thread.join()
    Web_Server_thread.join()

问题分析 #

1. 语法错误：原监听代码存在空赋值、错误调用nonblocking()的问题，导致读取逻辑异常。
2. 流式数据处理不当：串口是流式传输，read(100)可能只读取部分消息，或一次读取多条消息，直接传递给验证函数会导致消息截断或丢失。
3. 线程阻塞与竞争：全局变量serial_message_received存在线程安全问题，且消息验证/回复逻辑可能阻塞串口读取线程，导致后续消息无法及时读取。

解决方案与修正代码 #

核心优化点 #

• 用缓冲区累积串口数据，按消息界定符（{}）分割完整消息
• 分离串口读取与消息处理线程，避免读取阻塞
• 使用线程安全队列传递消息，解决全局变量竞争问题
• 调整串口超时配置，平衡读取及时性与线程占用

修正后的监听代码 #

import serial
import threading
import queue

# 线程安全队列：用于传递串口消息，避免全局变量竞争
serial_msg_queue = queue.Queue()

serial_port1 = serial.Serial(
    port='/dev/ttyUSB0',
    baudrate=9600,
    parity=serial.PARITY_NONE,
    stopbits=serial.STOPBITS_ONE,
    bytesize=serial.EIGHTBITS,
    timeout=0.1  # 短超时，避免读取线程长时间阻塞
)

def message_validation(message):
    """消息验证与回复逻辑"""
    try:
        msg_str = message.decode().strip()
        if msg_str.startswith('{') and msg_str.endswith('}'):
            print(f"收到有效消息: {msg_str}")
            # 示例回复：返回ACK+原消息
            serial_port1.write(f"ACK{msg_str}".encode())
        else:
            print(f"无效消息格式: {msg_str}")
    except UnicodeDecodeError:
        print(f"无法解码的二进制数据: {message}")

def serial_read_thread():
    """专门负责读取串口数据的线程"""
    buffer = b''
    while True:
        # 读取当前串口所有可用字节，无数据时读1字节触发超时
        read_bytes = serial_port1.read(serial_port1.in_waiting or 1)
        if read_bytes:
            buffer += read_bytes
            # 按消息界定符分割：每条消息以}结尾
            while b'}' in buffer:
                # 拆分出完整消息和剩余缓冲区数据
                full_msg, buffer = buffer.split(b'}', 1)
                full_msg += b'}'  # 补回被分割掉的界定符
                # 将完整消息放入队列
                serial_msg_queue.put(full_msg)
        # 让出CPU时间，避免占用过多资源
        threading.Event().wait(0.01)

def serial_process_thread():
    """专门负责处理消息验证与回复的线程"""
    while True:
        # 从队列取消息，队列为空时阻塞
        msg = serial_msg_queue.get()
        message_validation(msg)
        # 标记消息处理完成
        serial_msg_queue.task_done()

if __name__ == "__main__":
    Init()  # 保留原初始化函数

    # 创建所有线程
    thread_list = [
        threading.Thread(target=Flashing_LED_thr),
        threading.Thread(target=Running_Time_thr),
        threading.Thread(target=serial_read_thread),
        threading.Thread(target=serial_process_thread),
        threading.Thread(target=Monitoring_ADC_thr),
        threading.Thread(target=Monitoring_IO_thr),
        threading.Thread(target=Save_Config_File_thr),
        threading.Thread(target=Save_Config_STATUS_thr),
        threading.Thread(target=Web_Server_thr)
    ]

    # 启动所有线程
    for thread in thread_list:
        thread.start()

    # 等待所有线程结束
    for thread in thread_list:
        thread.join()

额外建议 #

• 检查串口硬件连接：确保USB转串口模块接触良好，供电稳定
• 调整串口缓冲区大小：可通过serial_port1.set_buffer_size(rx_size=4096)增大接收缓冲区
• 添加日志记录：记录所有接收和发送的消息，方便排查丢帧问题

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