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,baudrate9600,buffer_size1024):self.portport self.baudratebaudrate self.serNone# 线程安全的接收缓冲区self.rx_bufferdeque(maxlenbuffer_size)self._runningFalseself._rx_threadNonedefstart(self):初始化串口并启动接收线程try:self.serserial.Serial(self.port,self.baudrate,timeout1)self._runningTrueself._rx_threadthreading.Thread(targetself._read_loop,daemonTrue)self._rx_thread.start()print(f串口{self.port}已打开接收线程已启动)exceptExceptionase:print(f启动串口失败:{e})raisedef_read_loop(self):子线程持续读取串口数据并放入 dequewhileself._running:try:# 核心使用 in_waiting 避免死循环空转ifself.ser.in_waiting0:dataself.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 获取所有接收到的数据databwhileself.rx_buffer:dataself.rx_buffer.popleft()returndatadefstop(self):优雅关闭串口与线程self._runningFalseifself._rx_thread:self._rx_thread.join(timeout2)ifself.serandself.ser.is_open:self.ser.close()print(串口已安全关闭)# 使用示例if__name____main__:managerSerialManager(COM3,9600)try:manager.start()whileTrue:manager.send(bHello Serial\n)rxmanager.get_received_data()ifrx:print(f收到数据:{rx})time.sleep(1)exceptKeyboardInterrupt:manager.stop()常见问题与最佳实践在实际工程中串口通信极易受到硬件状态、系统权限和代码逻辑的影响。以下是针对常见问题的错误处理代码与最佳实践建议。1. 参数错误处理当port、baudrate、bytesize等参数不合法时pyserial会抛出ValueError。try:serserial.Serial(portCOM3,baudrate9600,bytesize8)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,timeout1)3. 权限问题 (Permission Denied)串口被其他程序占用或当前用户无访问权限时触发。try:serserial.Serial(COM3,9600)exceptserial.SerialExceptionase:ifPermission deniedinstr(e)orAccess is deniedinstr(e):print(权限错误: 请检查串口是否被其他程序占用或以管理员身份运行)else:print(f串口异常:{e})4. 死循环等待 vsin_waiting绝对禁止在接收线程中使用while True: ser.read(1)这种无阻塞的死循环读取方式这会导致 CPU 占用率飙升至 100%。错误做法无休眠的无限读取。正确做法优先使用in_waiting检查缓冲区是否有数据再读取如上文完整代码所示。结合timeout参数在初始化serial.Serial(timeout1)时设置超时。当没有数据时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,timeout1)asser:print(串口已成功打开)# 发送数据ser.write(bHello\n)# 读取数据lineser.readline()print(f收到数据:{line})# 退出 with 代码块后串口会自动关闭无需手动调用 ser.close()print(串口已自动关闭)exceptserial.SerialExceptionase:print(f串口打开失败:{e})exceptExceptionase:print(f发生未知错误:{e})补充说明什么时候才需要手动open()如果你需要延迟打开串口例如先创建对象等用户点击按钮后再连接你应该使用无参初始化然后再手动调用open()。此时就不能使用with语法# 延迟打开的用法不使用 withserserial.Serial()ser.portCOM3ser.baudrate9600# 此时串口尚未打开ser.open()# 手动打开ser.write(bTest)ser.close()# 必须手动关闭6. 其他注意事项资源释放务必在finally块或上下文管理器中调用ser.close()防止串口句柄泄露导致下次无法打开。线程退出标志使用self._running标志位控制子线程退出避免使用threading.Event以外的强制中断手段确保串口资源被正确清理。