Python Serial 多线程通信与异常处理实践指南

Python Serial 多线程通信与异常处理实践指南

环境准备：安装 pyserial 库

在开始编写串口通信代码之前，需要确保 Python 环境中已安装pyserial库。该库提供了跨平台的串口访问功能。

安装命令

在终端或命令行中执行以下命令即可完成安装：

pipinstallpyserial

简要说明

• 跨平台支持：pyserial兼容 Windows、Linux 和 macOS，屏蔽了底层操作系统的差异。
• 版本检查：安装完成后，可通过python -m serial --version验证是否安装成功及当前版本号。
• 依赖关系：该库无其他强制依赖，安装后即可直接通过import serial在代码中使用。

架构设计：threading 与 deque 通信

在 Python 串口（Serial）多线程开发中，通常采用“主线程发送 + 独立子线程接收”的架构。为了保证线程安全并高效地传递串口数据，推荐使用collections.deque作为线程间的数据通信缓冲区。

为什么选择 deque？

• 线程安全：deque的append()和popleft()操作在 CPython 中是原子的，无需额外加锁即可在多线程间安全传递数据。
• 高性能：基于双向链表实现，两端插入和弹出元素的时间复杂度均为O(1)O(1)O(1)，非常适合处理高频到达的串口数据流。
• 防溢出：支持设置maxlen参数，当缓冲区满时自动丢弃最旧的数据，防止内存泄漏。

完整代码示例

以下示例展示了如何开启读取线程、主线程发送数据以及优雅地管理串口生命周期。

importserialimportthreadingimporttimefromcollectionsimportdequeclassSerialManager:def__init__(self,port,baudrate=9600,buffer_size=1024):self.port=port self.baudrate=baudrate self.ser=None# 线程安全的接收缓冲区self.rx_buffer=deque(maxlen=buffer_size)self._running=Falseself._rx_thread=Nonedefstart(self):"""初始化串口并启动接收线程"""try:self.ser=serial.Serial(self.port,self.baudrate,timeout=1)self._running=Trueself._rx_thread=threading.Thread(target=self._read_loop,daemon=True)self._rx_thread.start()print(f"串口{self.port}已打开，接收线程已启动")exceptExceptionase:print(f"启动串口失败:{e}")raisedef_read_loop(self):"""子线程：持续读取串口数据并放入 deque"""whileself._running:try:# 核心：使用 in_waiting 避免死循环空转ifself.ser.in_waiting>0:data=self.ser.read(self.ser.in_waiting)self.rx_buffer.append(data)exceptserial.SerialExceptionase:print(f"读取异常:{e}")breakexceptExceptionase:print(f"未知异常:{e}")breakdefsend(self,data):"""主线程：发送数据"""ifself.serandself.ser.is_open:self.ser.write(data)defget_received_data(self):"""主线程：从 deque 获取所有接收到的数据"""data=b''whileself.rx_buffer:data+=self.rx_buffer.popleft()returndatadefstop(self):"""优雅关闭串口与线程"""self._running=Falseifself._rx_thread:self._rx_thread.join(timeout=2)ifself.serandself.ser.is_open:self.ser.close()print("串口已安全关闭")# 使用示例if__name__=="__main__":manager=SerialManager('COM3',9600)try:manager.start()whileTrue:manager.send(b"Hello Serial\n")rx=manager.get_received_data()ifrx:print(f"收到数据:{rx}")time.sleep(1)exceptKeyboardInterrupt:manager.stop()

常见问题与最佳实践

在实际工程中，串口通信极易受到硬件状态、系统权限和代码逻辑的影响。以下是针对常见问题的错误处理代码与最佳实践建议。

1. 参数错误处理

当port、baudrate、bytesize等参数不合法时，pyserial会抛出ValueError。

try:ser=serial.Serial(port='COM3',baudrate=9600,bytesize=8)exceptValueErrorase:print(f"参数配置错误:{e}")# 最佳实践：提供默认值或进行参数合法性预校验

2. 端口未找到 (Port Not Found)

当指定的串口被拔出或名称错误时，会抛出SerialException。

importserial.tools.list_portsdefsafe_open_port(port_name):available_ports=[p.deviceforpinserial.tools.list_ports.comports()]ifport_namenotinavailable_ports:print(f"错误: 端口{port_name}不存在。可用端口:{available_ports}")returnNonereturnserial.Serial(port_name,9600,timeout=1)

3. 权限问题 (Permission Denied)

串口被其他程序占用，或当前用户无访问权限时触发。

try:ser=serial.Serial('COM3',9600)exceptserial.SerialExceptionase:if"Permission denied"instr(e)or"Access is denied"instr(e):print("权限错误: 请检查串口是否被其他程序占用，或以管理员身份运行")else:print(f"串口异常:{e}")

4. 死循环等待 vsin_waiting

绝对禁止在接收线程中使用while True: ser.read(1)这种无阻塞的死循环读取方式，这会导致 CPU 占用率飙升至 100%。

• 错误做法：无休眠的无限读取。
• 正确做法：
  1. 优先使用in_waiting：检查缓冲区是否有数据再读取（如上文完整代码所示）。
  2. 结合timeout参数：在初始化serial.Serial(timeout=1)时设置超时。当没有数据时，read()会在 1 秒后返回空字节b''，从而让出 CPU 时间片，避免死循环。

5.with serial.Serial() as ser:常见报错与正确用法

为什么会报错？

当你使用with serial.Serial() as ser:时，上下文管理器在初始化时就已经自动调用了ser.open()。因此，如果你在代码块内部再次手动调用ser.open()，就会抛出异常，提示端口已经被打开（Port is already open）。

正确的做法

在使用with语法时，直接将串口参数（如端口号、波特率）传入serial.Serial()中即可，不需要再手动调用open()。

importserialtry:# 正确用法：直接在初始化时传入参数，with 语句会自动打开串口withserial.Serial('COM3',9600,timeout=1)asser:print("串口已成功打开")# 发送数据ser.write(b'Hello\n')# 读取数据line=ser.readline()print(f"收到数据:{line}")# 退出 with 代码块后，串口会自动关闭，无需手动调用 ser.close()print("串口已自动关闭")exceptserial.SerialExceptionase:print(f"串口打开失败:{e}")exceptExceptionase:print(f"发生未知错误:{e}")

补充说明：什么时候才需要手动open()？

如果你需要延迟打开串口（例如先创建对象，等用户点击按钮后再连接），你应该使用无参初始化，然后再手动调用open()。此时就不能使用with语法：

# 延迟打开的用法（不使用 with）ser=serial.Serial()ser.port='COM3'ser.baudrate=9600# 此时串口尚未打开ser.open()# 手动打开ser.write(b'Test')ser.close()# 必须手动关闭

6. 其他注意事项

• 资源释放：务必在finally块或上下文管理器中调用ser.close()，防止串口句柄泄露导致下次无法打开。
• 线程退出标志：使用self._running标志位控制子线程退出，避免使用threading.Event以外的强制中断手段，确保串口资源被正确清理。