Python结合asyncio实现异步串口非阻塞读写

传统的串口读写通常基于同步阻塞调用，当程序执行 ser.read() 时，线程会一直挂起等待数据到达，或者依赖超时轮询与多线程模型。在同时处理网络通信、多串口设备或图形界面的场景下，多线程会引入线程同步、锁竞争以及上下文切换的开销。借助 Python 标准库的 asyncio 及其串口异步扩展 pyserial-asyncio，我们可以将底层操作系统串口文件描述符挂载到事件循环中，利用事件通知机制实现单线程内的全双工非阻塞读写。

一、为什么串口读写需要接入 asyncio

在标准 pyserial 库中，无论是轮询读还是指定超时时间读取，本质上都在占用操作系统线程的 CPU 时间切片或陷入内核阻塞。如果系统需要管理 5 个甚至 10 个传感器串口，常规做法通常是为每个串口启动 1 到 2 个专用线程进行收发，这不仅让错误捕获和资源释放变得繁琐，而且极易因全局解释器锁（GIL）及跨线程队列引入不必要的延迟。

异步 I/O 的核心优势在于将“等待数据到来”的控制权移交给底层系统的多路复用机制（如 Linux 上的 epoll 或 select）。当串口没有数据准备好时，当前协程主动让出执行权，事件循环可以立即去调度发送任务、处理网络请求或执行计算；一旦硬件缓冲区有字节进入，事件循环唤醒对应的读取协程提取数据。

图 1：阻塞多线程模式与基于 asyncio 事件循环的单线程协作调度对比

二、环境准备与依赖安装

Python 标准库的 asyncio 自身并不直接提供串口硬件驱动，因此需要使用成熟的异步串口库 pyserial-asyncio。它底层基于 pyserial，并为 asyncio 封装了基于 Protocol/Transport 以及 Stream (Reader/Writer) 的高级 API。

安装命令如下：

pip install pyserial pyserial-asyncio

需要注意平台差异：在 Linux 与 macOS 上，串口以标准字符设备（如 /dev/ttyUSB0 或 /dev/cu.usbserial-*）存在，底层完全兼容 POSIX 标准的非阻塞文件描述符；在 Windows 系统上，串口为 COM 端口，asyncio 底层依赖重叠 I/O（Overlapped I/O），在 Python 3.8 及以上版本中，通常建议确保事件循环采用默认的 ProactorEventLoop。

三、快速实现：通过 open_serial_connection 建立异步流

serial_asyncio.open_serial_connection 是最直观也最符合现代 asyncio 编程习惯的 API。它直接返回一组 (StreamReader, StreamWriter) 对象，使我们像操作网络 Socket 一样读写串口。

图 2：基于 Stream API 的异步串口建立连接与非阻塞交互步骤

下面是一个标准的读写示例代码。我们建立连接后，向串口发送一条探测指令，并以异步方式等待行数据返回：

import asyncio
import serial_asyncio

async def main():
    # 建立串口异步读写流
    reader, writer = await serial_asyncio.open_serial_connection(
        url='/dev/ttyUSB0',   # Windows 上使用 'COM3'
        baudrate=115200
    )

    print("串口已成功打开，准备发送指令...")
    
    # 写入数据（写入操作进入缓冲区）
    writer.write(b"PING\n")
    # drain() 是关键：等待缓冲区数据全部刷入底层硬件
    await writer.drain()
    print("指令已发送，等待设备响应...")

    try:
        # 异步非阻塞等待接收一行数据，设置 5 秒超时防止挂死
        line = await asyncio.wait_for(reader.readline(), timeout=5.0)
        print(f"收到回复: {line.decode('utf-8', errors='replace').strip()}")
    except asyncio.TimeoutError:
        print("等待响应超时，设备未返回数据")
    finally:
        # 关闭连接并等待底层传输彻底释放
        writer.close()
        await writer.wait_closed()
        print("串口已安全关闭")

if __name__ == '__main__':
    asyncio.run(main())

四、全双工并发模式：独立读写任务

实际的工业设备或传感器交互通常不是一问一答的同步模式。设备往往会以 50Hz 或 100Hz 的频率不间断上报遥测数据，而上位机也会随机下发控制命令。此时我们需要将“读取”与“写入”解耦成两个长期并发运行的协程。

图 3：读写分离与异步队列驱动的全双工非阻塞架构

实现该架构的代码非常结构化：利用一个 asyncio.Queue 暂存待发送的数据帧，发送协程通过 await queue.get() 获取数据并发送；读取协程则持续运行 await reader.read(1024) 或 readline()：

import asyncio
import serial_asyncio

async def reader_loop(reader: asyncio.StreamReader):
    """独立的数据接收协程"""
    print("[RX] 接收循环启动")
    try:
        while True:
            # 持续非阻塞读取数据（收到多少读多少）
            data = await reader.read(1024)
            if not data:
                # 收到空字节通常意味着底层设备断开连接
                print("[RX] 串口已被对端关闭")
                break
            print(f"[RX 收到] {data.hex()} ({data})")
    except asyncio.CancelledError:
        print("[RX] 接收循环收到取消信号，准备退出")
    except Exception as e:
        print(f"[RX 异常] {e}")

async def writer_loop(writer: asyncio.StreamWriter, send_queue: asyncio.Queue):
    """独立的数据发送协程，从队列中消费数据"""
    print("[TX] 发送循环启动")
    try:
        while True:
            # 等待业务层投递发送命令
            cmd = await send_queue.get()
            writer.write(cmd)
            await writer.drain()
            print(f"[TX 已发送] {cmd}")
            send_queue.task_done()
    except asyncio.CancelledError:
        print("[TX] 发送循环已取消")
    except Exception as e:
        print(f"[TX 异常] {e}")

async def business_logic(send_queue: asyncio.Queue):
    """模拟业务层周期性或按需下发指令"""
    for i in range(3):
        await asyncio.sleep(2.0)
        msg = f"CMD_SEQ_{i}\n".encode('utf-8')
        await send_queue.put(msg)

async def main():
    reader, writer = await serial_asyncio.open_serial_connection(
        url='/dev/ttyUSB0',
        baudrate=115200
    )
    
    send_queue = asyncio.Queue()

    # 将收、发、业务放入同一个任务组并发运行
    reader_task = asyncio.create_task(reader_loop(reader))
    writer_task = asyncio.create_task(writer_loop(writer, send_queue))
    biz_task = asyncio.create_task(business_logic(send_queue))

    # 等待业务逻辑运行完成
    await biz_task
    # 等待队列中的全部指令发送完毕
    await send_queue.join()

    # 取消后台长任务并平稳回收
    reader_task.cancel()
    writer_task.cancel()
    await asyncio.gather(reader_task, writer_task, return_exceptions=True)

    writer.close()
    await writer.wait_closed()
    print("系统已平稳停机")

if __name__ == '__main__':
    asyncio.run(main())

五、常见避坑指南与工程实践

将串口接入异步生态虽然带来了高效与整洁，但在实际工程调试中，有几个细节需要特别警惕：

• 粘包与断包处理：串口是流式（Stream-based）传输，而非面向消息报文（Message-based）。reader.read(n) 返回的数据块大小由系统缓冲区和波特率决定，不能假定单次读取刚好对应一条完整的业务指令。若通信协议基于换行符，可直接使用 reader.readline()；若为固定帧头+长度+校验和的二进制协议（如 Modbus、自定义航模协议），必须在协程内实现帧拼装状态机。
• 热插拔与设备掉线检测：当物理 USB 转串口线被意外拔出时，底层文件描述符通常会抛出 SerialException 或 OSError，或持续返回长度为 0 的字节块。因此在 reader_loop 中，如果检测到 not data 或捕获到 I/O 错误，必须主动退出循环并触发重连机制。
• 禁止在协程中调用同步 time.sleep()：协程之间通过合作式调度运行在同一线程中。一旦在协程内部执行了同步阻塞的 time.sleep() 或耗时极长的大循环，整个事件循环都会被冻结，所有串口的接收缓冲区都可能因为未能及时清空而发生溢出（Buffer Overrun）。耗时计算应移交 asyncio.to_thread() 或 loop.run_in_executor()。

六、全文总结与选型决策

为了帮助开发者根据实际场景评估是否应引入 asyncio 串口方案，下表梳理了常见实现方案的核心特征与适用边界：

图 4：Python 串口通信方案全维度对比与选型决策

结合 asyncio 以后，串口不再是一个割裂的阻塞源，而是可以作为标准异步数据流自然融入整体应用架构。无论是把采集到的遥测信息直接 await websocket.send() 推送给前端，还是通过异步队列安全排队下发控制指令，都能够在单线程内保证极高执行效率与代码的直观优雅。