4. 同步与异步通信模式：阻塞式读写、非阻塞式读写（timeout机制）、线程安全读写、事件驱动（serial.threaded）

串口通信，说白了就是两个设备之间通过一根线（或几根线）来回倒腾数据。但怎么倒腾，这里面的门道可不少。你想想看，如果CPU一直死等着数据过来，那其他活儿还干不干了？反过来，如果CPU把数据扔出去就不管了，那数据丢了怎么办？

这一章，我们就来聊聊这些通信模式。我个人习惯把同步/异步、阻塞/非阻塞这些概念放在一起讲，因为它们在实际项目中是纠缠在一起的。搞懂了这些，你写出来的串口驱动才算是真正能上生产线的。

4.1 阻塞式读写：简单粗暴，但得悠着点

阻塞式读写，就是最直白的方式。你调用read()，程序就停在那里，直到读到足够的数据或者文件关闭。同样，write()会等到数据全部写入缓冲区才返回。

import serial

ser = serial.Serial('/dev/ttyUSB0', 115200, timeout=None)  # timeout=None 表示阻塞模式
data = ser.read(10)  # 这里会一直等，直到读到10个字节
print(f"读到数据: {data.hex()}")

我在项目中遇到过一个问题：用阻塞模式读一个传感器，结果传感器偶尔会掉线，程序就永远卡在read()那里了。后来加了看门狗才解决。所以，阻塞模式虽然简单，但用在关键路径上要小心。

4.2 非阻塞式读写与timeout机制：给程序一个“耐心值”

非阻塞模式，说白了就是给读写操作加一个“耐心值”。你告诉串口：“我给你0.5秒，有数据就给我，没有你就回来告诉我。”这样程序就不会死等了。

在PySerial中，设置timeout参数就能实现非阻塞效果：

import serial

# timeout=0.5 表示最多等0.5秒
ser = serial.Serial('/dev/ttyUSB0', 115200, timeout=0.5)

# 尝试读10个字节，但最多等0.5秒
data = ser.read(10)

if len(data) == 0:
    print("超时了，没读到数据")
elif len(data) < 10:
    print(f"只读到了{len(data)}个字节，可能数据还没到齐")
else:
    print(f"完整读到10个字节: {data.hex()}")

这里有个细节要注意：timeout的单位是秒，可以是浮点数。设置成0就是立即返回，不管有没有数据。我个人习惯设成0.1到1.0之间，具体看你的应用场景。

timeout值	行为	典型场景
None	阻塞模式，一直等到数据到齐	数据确定性高的场景
0	非阻塞模式，立即返回	轮询查询、UI响应
0.1 ~ 1.0	超时模式，等指定时间后返回	工业通信、传感器读取

4.3 线程安全读写：别让数据打架

多线程环境下，串口读写会出问题。你想想看，如果两个线程同时往串口写数据，那数据就混在一起了。同样，一个线程在读，另一个线程在写，也可能导致数据错乱。

解决这个问题，我一般用两种方式：

1. 加锁（Lock）：用Python的threading.Lock保护读写操作。
2. 专用读写线程：一个线程专门读，一个线程专门写，通过队列传递数据。

先看加锁的方式：

import serial
import threading

class SafeSerial:
    def __init__(self, port, baudrate):
        self.ser = serial.Serial(port, baudrate, timeout=0.5)
        self._lock = threading.Lock()
    
    def read(self, size):
        with self._lock:
            return self.ser.read(size)
    
    def write(self, data):
        with self._lock:
            return self.ser.write(data)
    
    def close(self):
        with self._lock:
            self.ser.close()

这种方式简单直接，但有个问题：如果读操作阻塞了，写操作也得等着。所以，我更喜欢第二种方式——用队列解耦：

import serial
import threading
import queue

class SerialWorker:
    def __init__(self, port, baudrate):
        self.ser = serial.Serial(port, baudrate, timeout=0.1)
        self.tx_queue = queue.Queue()
        self.rx_queue = queue.Queue()
        self._running = True
        
        self._tx_thread = threading.Thread(target=self._tx_loop, daemon=True)
        self._rx_thread = threading.Thread(target=self._rx_loop, daemon=True)
        self._tx_thread.start()
        self._rx_thread.start()
    
    def _tx_loop(self):
        while self._running:
            try:
                data = self.tx_queue.get(timeout=0.1)
                self.ser.write(data)
            except queue.Empty:
                continue
    
    def _rx_loop(self):
        while self._running:
            data = self.ser.read(1024)
            if data:
                self.rx_queue.put(data)
    
    def send(self, data):
        self.tx_queue.put(data)
    
    def receive(self, timeout=0.5):
        try:
            return self.rx_queue.get(timeout=timeout)
        except queue.Empty:
            return None
    
    def close(self):
        self._running = False
        self._tx_thread.join(timeout=1)
        self._rx_thread.join(timeout=1)
        self.ser.close()

4.4 事件驱动：serial.threaded 的妙用

PySerial从3.0版本开始，提供了一个叫serial.threaded的模块。这东西说白了就是一个事件驱动的框架，帮你把串口读写、协议解析、事件回调都封装好了。

我刚开始用的时候也觉得有点绕，但用顺手之后，发现它特别适合做工业通信协议的处理。比如Modbus、自定义帧协议这些。

来看一个例子：

import serial
import serial.threaded

class MyProtocol(serial.threaded.LineReader):
    TERMINATOR = b'\n'
    
    def connection_made(self, transport):
        """连接建立时调用"""
        super().connection_made(transport)
        print("串口已打开，开始通信")
    
    def handle_line(self, line):
        """收到一行数据时调用"""
        print(f"收到数据: {line.decode('utf-8', errors='ignore')}")
        # 这里可以做协议解析
        if line.startswith(b'AT'):
            self.transport.write(b'OK\n')
    
    def connection_lost(self, exc):
        """连接断开时调用"""
        if exc:
            print(f"连接异常断开: {exc}")
        else:
            print("连接正常关闭")

# 使用
ser = serial.Serial('/dev/ttyUSB0', 115200, timeout=0.1)
with serial.threaded.ReaderThread(ser, MyProtocol) as protocol:
    protocol.write(b'AT\r\n')
    time.sleep(1)

这个框架的核心思想是：你只需要实现几个回调方法，框架会自动处理线程、读写、数据分割这些事情。我个人觉得，它特别适合处理那些有固定帧格式的协议。

4.5 知识体系总览

下面这张图，把本章讲到的几种通信模式的关系画出来了。你可以看到，从最简单的阻塞模式，到最复杂的事件驱动，每种模式都有它的适用场景。

4.6 如何选择？我的经验总结

说了这么多，到底该用哪种模式？我根据自己的项目经验，给你几个判断标准：

• 简单数据采集：用阻塞模式+timeout，代码简单，容易调试。
• 多设备通信：用非阻塞模式+轮询，避免一个设备卡死影响其他设备。
• 多线程环境：必须用线程安全方式，加锁或队列二选一。
• 复杂协议处理：用serial.threaded，省心省力。