Python串口数据采集丢包求助：每4-15分钟丢失30-45秒数据

嘿，我之前做多串口数据采集的时候也踩过几乎一模一样的坑！这种周期性的30-45秒数据丢失，大概率是单线程阻塞或者资源竞争导致的，咱们一步步来解决它！

一、核心问题分析 #

从你的代码片段来看，你应该是用单线程同时处理两个串口的采集任务吧？单线程下，当你在读取其中一个串口的数据时（比如等待timeout或者用readline()阻塞），另一个串口的 incoming 数据会堆积在硬件缓冲区里——一旦缓冲区满了，后续的数据就会直接丢失，这就是为什么你会每隔几分钟就出现一次数据断档。

二、具体解决方案 #

1. 用多进程隔离每个串口的采集任务 #

每个串口单独分配一个进程来处理，这样它们之间完全独立，不会因为一个串口的阻塞影响另一个的采集。推荐用多进程而非多线程，因为Python的GIL锁会限制多线程的并行性能，多进程能充分利用CPU资源，避免采集延迟。

2. 优化串口读取策略，避免不必要的阻塞 #

• 把timeout设置得小一点（比如0.1秒），别用2秒这么长的超时，否则每次空读取都会阻塞很久
• 用ser.in_waiting检查是否有可用数据，再一次性读取所有数据，而不是逐行读取（除非你的数据格式是严格换行分隔的）
• 采集进程只负责读数据，别在里面做解析、写入这类耗时操作，把这些工作交给专门的进程处理

3. 异步写入文件，避免采集被IO阻塞 #

单独开一个进程负责写入文件，采集进程只需要把数据放到进程间队列里就行。这样采集操作不会因为文件IO的延迟被打断，也能保证数据及时写入磁盘，不会因为程序卡顿丢失内存中的数据。

三、完整代码示例 #

import serial
import multiprocessing
import time

def collect_serial_data(port, baudrate, data_queue):
    """单个串口的数据采集进程"""
    # 配置串口，缩小timeout减少阻塞时间
    ser = serial.Serial(
        port=port,
        baudrate=baudrate,
        timeout=0.1,
        parity=serial.PARITY_NONE,
        stopbits=serial.STOPBITS_ONE,
        bytesize=serial.EIGHTBITS
    )
    
    try:
        while True:
            # 先检查是否有可用数据，避免空读取阻塞
            if ser.in_waiting > 0:
                # 一次性读取所有缓冲区数据，防止溢出
                raw_data = ser.read(ser.in_waiting)
                # 带上时间戳和端口标识，方便后续排查问题
                data_queue.put((time.time(), port, raw_data))
            # 短暂休眠降低CPU占用，避免忙等
            time.sleep(0.001)
    except Exception as e:
        print(f"串口 {port} 发生错误: {str(e)}")
    finally:
        ser.close()

def write_data_to_file(data_queue, output_filename):
    """专门的文件写入进程"""
    # 用二进制追加模式，避免编码格式冲突
    with open(output_filename, 'ab') as f:
        while True:
            if not data_queue.empty():
                timestamp, port, raw_data = data_queue.get()
                # 格式化时间戳，方便日志回溯
                time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(timestamp))
                # 把原始数据转成字符串（忽略编码错误），加上端口和时间戳
                log_line = f"[{time_str}] [{port}] {raw_data.decode('utf-8', errors='ignore')}\n".encode('utf-8')
                f.write(log_line)
                # 强制刷新缓冲区，确保数据立即写入磁盘
                f.flush()
            time.sleep(0.001)

if __name__ == '__main__':
    # 获取两个串口的端口号
    gps_port = input("Enter GPS comport as 'COM_': ")
    second_port = input("Enter second comport as 'COM_': ")
    
    # 创建进程间通信队列，设置最大容量防止内存溢出
    data_queue = multiprocessing.Queue(maxsize=1000)
    
    # 启动GPS串口采集进程
    gps_process = multiprocessing.Process(
        target=collect_serial_data,
        args=(gps_port, 38400, data_queue)
    )
    # 启动第二个串口采集进程（替换成你的设备波特率）
    second_process = multiprocessing.Process(
        target=collect_serial_data,
        args=(second_port, 你的波特率, data_queue)
    )
    
    # 启动文件写入进程
    write_process = multiprocessing.Process(
        target=write_data_to_file,
        args=(data_queue, "serial_log.txt")
    )
    
    # 启动所有进程
    gps_process.start()
    second_process.start()
    write_process.start()
    
    # 等待进程结束（按Ctrl+C终止）
    try:
        gps_process.join()
        second_process.join()
        write_process.join()
    except KeyboardInterrupt:
        print("\n正在停止采集...")
        # 终止所有进程
        gps_process.terminate()
        second_process.terminate()
        write_process.terminate()
        print("已停止所有进程")

四、额外排查建议 #

• 检查硬件问题：换一根屏蔽性好的串口线，确保设备供电稳定，远离电磁干扰源（比如电机、路由器），有时候硬件故障才是丢数据的根源
• 查看系统串口日志：Windows可以在设备管理器里查看串口属性的事件日志，Linux用dmesg | grep tty查看有没有串口溢出（overrun）错误，如果有说明硬件或驱动存在问题
• 测试单串口采集：先单独测试每个串口的采集，确认单串口不会丢数据，再测试多串口，这样能快速排除单个设备的问题

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