使用pyserial处理双队列时CPU占用过高问题排查

问题背景 #

通过串口进行数据读写，并将数据存入多进程队列。理论上现代CPU处理字节速度远快于57600波特率，但实际单核心CPU占用100%：树莓派上仅达4kB/s（约32kbps），笔记本满核心占用时才到约8kB/s。

串口初始化代码：

ser1 = serial.Serial('./reader', 57600, timeout=0.00001)

核心处理函数：

print("servicing port!")
while(1):
  if(ser1.in_waiting):
    newbytes = ser1.read()
    tx.put(newbytes)
  time.sleep(0.00001)
  try:
    rxbyte=rx.get(block=False)
    ser1.write(rxbyte)
  except QueueEmpty:
    time.sleep(0.00001)
    continue

尝试过相关方案但无改善：

def serviceport():
  print("servicing port!")
  while(1):
    newbytes = ser1.read(1)
    tx.put(newbytes)
    if(ser1.in_waiting):
      newbytes=ser1.read()
      tx.put(newbytes)
    try:
      rxbyte=rx.get(block=False)
      ser1.write(rxbyte)
    except QueueEmpty:
      time.sleep(0.00001)
      continue

增加sleep时间能降低CPU占用，但数据速率会进一步下降。

核心问题分析 #

1. 逐字节读写的巨大开销：每次read()/read(1)只读1字节，write()也仅写1字节，串口驱动和系统调用的开销远大于数据本身的处理，频繁的系统调用直接拉满CPU。
2. 忙等待与过短sleep：time.sleep(0.00001)（10微秒）在多数系统中精度不足，实际可能变成忙等待循环，CPU一直在空转检查状态，而非真正休眠。
3. 多进程队列的同步开销：每字节都调用tx.put()/rx.get()，跨进程队列的锁竞争会额外消耗CPU资源。
4. 串口timeout设置不合理：timeout=0.00001会让read()几乎立刻返回，导致循环内不断发起无效的读请求。

解决方案 #

1. 批量读写串口数据 #

不要逐字节处理，一次性读取缓冲区中所有可用数据，减少系统调用次数：

print("servicing port!")
while True:
    # 批量读取所有等待的数据
    if ser1.in_waiting > 0:
        new_bytes = ser1.read(ser1.in_waiting)
        tx.put(new_bytes)  # 批量放入队列，降低锁竞争频率

    # 批量处理写队列
    try:
        while True:
            rx_bytes = rx.get(block=False)
            ser1.write(rx_bytes)
    except QueueEmpty:
        pass

    # 用合理的休眠减少空转，比如1ms
    time.sleep(0.001)

2. 优化sleep策略 #

• 去掉循环内多次sleep，只在循环末尾做一次合理休眠，避免忙等待。
• 不要使用小于1ms的sleep，多数系统定时器精度在1ms左右，过短的sleep会被系统忽略，变成无效空转。

3. 调整串口timeout参数 #

把timeout设置为更合理的值，比如timeout=0.001（1ms），减少无效读请求的频率：

ser1 = serial.Serial('./reader', 57600, timeout=0.001)

4. 分离读写线程/进程 #

把读操作和写操作分开到不同线程，避免读写互相阻塞，减少单线程内的状态切换开销：

import threading

def read_serial():
    while True:
        if ser1.in_waiting > 0:
            data = ser1.read(ser1.in_waiting)
            tx.put(data)
        time.sleep(0.001)

def write_serial():
    while True:
        try:
            data = rx.get(block=True)  # 用阻塞等待，彻底避免空转
            ser1.write(data)
        except Exception as e:
            pass

# 启动两个后台线程
threading.Thread(target=read_serial, daemon=True).start()
threading.Thread(target=write_serial, daemon=True).start()

# 主线程保持运行
while True:
    time.sleep(1)

5. 减少多进程队列调用频率 #

尽量批量传递数据，比如积累到128字节再放入队列，或者直接传递字节串而非单个字节，降低跨进程同步的锁开销。

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