Python如何实现非阻塞的串口通信_使用asyncio扩展串口读取功能-Python教程-PHP中文网

用 asyncio.to_thread 包裹 serial.read() 是最稳方案，因 serial.Serial 为同步阻塞 IO；需设 timeout 防卡死，write() 无需 await；推荐后台 task 持续读取并写入 asyncio.Queue，避免轮询或过时库 pyserial-asyncio。

python如何实现非阻塞的串口通信_使用asyncio扩展串口读取功能

串口读取卡主线程？用 `asyncio.to_thread` 包一层最稳

Python 标准库的 serial.Serial 是纯同步阻塞 IO，直接扔进 asyncio 里会拖垮整个事件循环。别碰 loop.run_in_executor 手动写线程池——除非你真需要细粒度控制超时或取消，否则 asyncio.to_thread（Python 3.9+）是更干净的选择。

常见错误现象：await serial_port.read(1) 报错 TypeError: object bytes can't be used in 'await' expression，因为 read() 不是协程。

把阻塞调用包进 asyncio.to_thread(serial_port.read, 1)，它自动在后台线程执行并返回可 await 的 Future
读取前确保串口已打开且 timeout=0 或设为较小值（如 0.05），否则 to_thread 仍会卡住线程
不要对 serial_port.write() 做 await 包装——它本身不阻塞（除非缓冲区满且 write_timeout 非零），直接调用即可

想监听串口数据到达就触发回调？用 `asyncio.Queue` 当中转站

轮询 in_waiting 再 read() 很容易漏字节或重复读；用回调机制又得自己管理线程安全。折中方案：起一个后台 task 持续读、往 asyncio.Queue 里塞数据，业务逻辑只管从 queue 拿。

使用场景：上位机接收传感器心跳包、解析不定长协议帧、多任务并发消费串口数据流。

初始化队列：queue = asyncio.Queue(maxsize=100)，maxsize 防止内存暴涨

后台读取 task 示例：

async def reader_task(serial_port, queue):
    while True:
        try:
            data = await asyncio.to_thread(serial_port.read, 1)
            if data:
                await queue.put(data)
        except serial.SerialException:
            break

业务侧用 await queue.get() 或 async for item in queue:（需配合 asyncio.Queue 的迭代器支持）

`pyserial-asyncio` 看起来很香？先看清它不维护的事实

GitHub 上搜 pyserial-asyncio，第一个结果是 2018 年归档的项目，最后 commit 在 Python 3.6 时代。它依赖过时的 asyncio.StreamReader/StreamWriter 接口，和现代 asyncio 的异常传播、取消逻辑有兼容问题。

macOS UI 自动化技能包 —— 用 Python 控制你的 Mac

让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限

下载

性能 / 兼容性影响：在 Python 3.11+ 下可能触发 RuntimeWarning: coroutine 'XXX' was never awaited，或在连接断开时无法正确清理资源。

别 pip install pyserial-asyncio，它不是官方维护，也不是 pyserial 的一部分
pyserial 官方明确表示「不提供原生 async 支持」，文档里写的替代方案就是 to_thread 或 executor
如果硬要用第三方库，serial.tools.miniterm 里的异步部分可参考，但别直接依赖——它只是调试工具，非生产级

超时和断连怎么处理？别只靠 `try/except`

串口设备拔掉、USB 转串口芯片掉线、权限丢失，这些都会让 serial.Serial 抛出 serial.SerialException。但单纯捕获异常不够——后台读取 task 可能已死，queue 停滞，而主逻辑还在等 queue.get()。

容易踩的坑：用 asyncio.wait_for(queue.get(), timeout=5) 后没检查是否真的拿到数据，导致后续解析空字节或旧缓存。

在 reader task 里 catch SerialException 后主动 break，再发个信号（比如往另一个 asyncio.Event set）通知主逻辑重连
写入前检查 serial_port.is_open，避免 write() 报 ValueError: I/O operation on closed file
重连逻辑别写在 reader task 里——新串口对象要重新创建，老对象必须 close()，否则 Linux 下可能残留 /dev/ttyUSB0 占用

真正麻烦的是硬件层的不可靠：没有标准方式区分“暂时无数据”和“设备已消失”，得靠应用层加心跳、超时计数、重试退避，这部分没法靠 asyncio 自动解决。