用 asyncio.to_thread 包裹 serial.read() 是最稳方案,因 serial.Serial 为同步阻塞 IO;需设 timeout 防卡死,write() 无需 await;推荐后台 task 持续读取并写入 asyncio.Queue,避免轮询或过时库 pyserial-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 拿。
使用场景:上位机接收传感器心跳包、解析不定长协议帧、多任务并发消费串口数据流。
立即学习“Python免费学习笔记(深入)”;
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac,无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启:辅助功能 + 屏幕录制权限
- 初始化队列:
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 的异常传播、取消逻辑有兼容问题。
性能 / 兼容性影响:在 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.Eventset)通知主逻辑重连 - 写入前检查
serial_port.is_open,避免write()报ValueError: I/O operation on closed file - 重连逻辑别写在 reader task 里——新串口对象要重新创建,老对象必须 close(),否则 Linux 下可能残留
/dev/ttyUSB0占用
真正麻烦的是硬件层的不可靠:没有标准方式区分“暂时无数据”和“设备已消失”,得靠应用层加心跳、超时计数、重试退避,这部分没法靠 asyncio 自动解决。

















