PySerial调用read()读取超10000字节时程序冻结如何解决？

读取15000字节触发程序冻结和数据量本身无直接关联，是默认配置与错误读取逻辑导致的必然结果：

• PySerial默认开启永久阻塞读取模式，ser.read(n)会持续挂起直到凑齐n个字节才返回。只要对端发送存在字节间隔、实际发送总长度略小于指定读取长度，调用就会无限等待，表现为程序无响应。
• 操作系统串口驱动、USB转串口芯片的默认输入缓冲区大小普遍在4096~16384字节区间，未主动调整时，数据量接近阈值如果没及时读走，会直接触发IO阻塞甚至丢包，这也是测试时10000字节可读、15000字节必卡的典型阈值表现。
• 此前尝试的方案本身存在逻辑缺陷：刷新缓冲区会直接丢弃未读取的数据，无法完成完整接收；逐字节读取会产生巨量系统IO调用，读取效率远低于高速串口的发送速率，很快就会堵满缓冲区；反复开关串口会直接丢弃传输过程中的数据包，不可能实现大容量数据的稳定接收。

先升级PySerial到最新稳定版，避免旧版本的缓冲区bug：

pip install -U pyserial

1. 初始化阶段修正基础配置 #

初始化串口时必须设置合理超时，同时调大上层接收缓冲区，从根源避免永久阻塞：

import serial

ser = serial.Serial(
    port="你的实际串口号",
    baudrate=115200, # 替换为实际使用的波特率
    timeout=0.2, # 核心配置：读超时设为0.2秒，收不到新数据就直接返回，不会永久卡
    write_timeout=0.2,
    inter_byte_timeout=0.1 # 字节间超时，两个字节间隔超过0.1秒即判定当前传输段结束
)
# 调大PySerial层接收缓冲区到64KB，适配大容量数据流
ser.set_buffer_size(rx_size=65536, tx_size=4096)

注意：绝对不要保留默认的timeout=None配置，这是触发永久阻塞的最直接原因。

2. 替换固定长度读取逻辑，用非阻塞循环拼包 #

禁止直接调用ser.read(15000)这种指定大长度的阻塞读，改为每次读取固定小块数据，循环拼接直到满足退出条件，全程不会触发永久阻塞：

def recv_large_serial_data(ser, expected_len=None, end_mark=None):
    """
    大容量串口数据非阻塞读取函数
    :param ser: 初始化完成的serial对象
    :param expected_len: 已知数据总长度时传入，凑够长度自动返回
    :param end_mark: 约定的帧尾标识时传入，读到帧尾自动返回
    """
    recv_buf = bytearray()
    while True:
        # 每次最多读取1024字节，超时就返回空值，不会挂起
        chunk = ser.read(1024)
        recv_buf.extend(chunk)
        # 满足长度要求退出
        if expected_len and len(recv_buf) >= expected_len:
            break
        # 读到帧尾退出
        if end_mark and recv_buf.endswith(end_mark):
            break
        # 缓冲区已读空、暂时无新数据时退出
        if len(chunk) == 0:
            break
    return bytes(recv_buf)

该逻辑对10KB到数MB级别的数据流都兼容，单次读操作最长等待时间就是设置的timeout值，不会卡住主程序。

3. 超高波特率场景的优化方案 #

如果使用921600及以上的高波特率传输，主线程轮询可能因为调度不及时堵满缓冲区，直接使用PySerial自带的后台读线程，把读取操作放到独立线程执行，完全不阻塞主业务逻辑：

from serial.threaded import ReaderThread, Protocol

class SerialRecvProtocol(Protocol):
    def __init__(self):
        self.data_buf = bytearray()
    def connection_made(self, transport):
        self.transport = transport
    def data_received(self, data):
        # 后台线程自动接收数据存入缓冲区，无需主线程轮询
        self.data_buf.extend(data)
    def fetch_all_data(self):
        # 主线程需要取数据时调用即可，无阻塞
        res = bytes(self.data_buf)
        self.data_buf.clear()
        return res

# 启动后台读线程
reader = ReaderThread(ser, SerialRecvProtocol)
reader.start()
_, recv_proto = reader.connect()

# 业务逻辑中取数据示例
# current_data = recv_proto.fetch_all_data()

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