如何在 asyncio 中安全使用 input() 实现串口客户端实时通信-Python教程-PHP中文网

如何在 asyncio 中安全使用 input() 实现串口客户端实时通信

本文解决 pyserial-asyncio 客户端中因阻塞式 input() 导致数据无法及时发送至服务器的问题，核心方案是通过 loop.run_in_executor() 将输入操作移入线程池执行，避免阻塞事件循环。

本文解决 pyserial-asyncio 客户端中因阻塞式 `input()` 导致数据无法及时发送至服务器的问题，核心方案是通过 `loop.run_in_executor()` 将输入操作移入线程池执行，避免阻塞事件循环。

在基于 asyncio 的串口通信（如使用 pyserial-asyncio）开发中，一个常见却容易被忽视的陷阱是：直接在协程中调用同步阻塞函数（如 input()）会完全冻结整个事件循环。这正是你遇到问题的根本原因——客户端虽能提示输入，但 input() 阻塞了主线程，导致 send_data() 协程无法继续执行 writer.write() 和 writer.drain()；而 receive_data() 同样被卡住，造成“所有消息仅在退出时批量打印”的假象。

相比之下，echo foo > /dev/ttyAMA1 能立即触发服务端响应，是因为该命令是纯异步的系统级写入，不涉及 Python 事件循环调度。

✅ 正确做法：将 input() 移出事件循环，交由线程池执行
asyncio 提供了 loop.run_in_executor(None, func, *args) 方法，可将任意同步函数提交至默认线程池（concurrent.futures.ThreadPoolExecutor）异步执行，同时返回一个可等待的 Future。这样既保留了用户交互体验，又不阻塞 I/O 协程。

以下是修复后的客户端完整实现（兼容 Python 3.8+，已适配你的硬件环境）：

import asyncio
import serial_asyncio
import sys

async def uart_client(port, baudrate):
    try:
        reader, writer = await serial_asyncio.open_serial_connection(
            url=port, baudrate=baudrate
        )
        print(f"Connected to {port} at {baudrate} bps")

        # ✅ 异步输入函数：在后台线程中执行 input()
        async def async_input(prompt: str) -> str:
            loop = asyncio.get_running_loop()
            # 注意：Python 3.8+ 推荐使用 get_running_loop() 替代 get_event_loop()
            return await loop.run_in_executor(None, lambda: input(prompt))

        async def receive_data():
            while True:
                try:
                    data = await reader.readline()
                    if data:
                        print(f"Received: {data.decode().strip()}")
                except Exception as e:
                    print(f"Error reading data: {e}")
                    break

        async def send_data():
            while True:
                try:
                    # ⚠️ 不再使用阻塞的 input()，改用异步封装
                    message = await async_input("Enter message to send (or 'exit' to quit): ")
                    if message.lower() == 'exit':
                        break
                    # 添加换行符确保 server.readline() 可正确解析（与 server 的 b"exit\r" 匹配需注意）
                    writer.write((message + '\n').encode())
                    await writer.drain()  # 确保数据刷新到串口缓冲区
                    print(f"Sent: {message}")
                except KeyboardInterrupt:
                    break
                except Exception as e:
                    print(f"Send error: {e}")
                    break

        # 并发运行收发任务
        await asyncio.gather(receive_data(), send_data())

    except serial_asyncio.serial.serialutil.SerialException as e:
        print(f"Error opening serial port: {e}")
    finally:
        if 'writer' in locals() and not writer.is_closing():
            writer.close()
            await writer.wait_closed()
            print("Serial connection closed.")

if __name__ == "__main__":
    # Raspberry Pi 5 示例：客户端使用 /dev/ttyAMA1，服务端监听 /dev/ttyAMA2
    asyncio.run(uart_client('/dev/ttyAMA1', 9600))

? 关键注意事项：

换行符一致性：服务端使用 reader.readline()（依赖 \n 或 \r\n），因此客户端必须发送 message + '\n'（而非 \r）。若服务端协议要求 \r 结尾，请统一为 message + '\r' 并确保 readline() 能识别（readline() 默认以 \n 为界，可考虑改用 readuntil(b'\r')）。
资源清理：务必检查 writer.is_closing() 再调用 close()，避免重复关闭异常。
异常鲁棒性：async_input() 可能抛出 EOFError（如 Ctrl+D），建议在外层 try/except 捕获。
串口权限：确保当前用户有 /dev/ttyAMA1 和 /dev/ttyAMA2 的读写权限（推荐加入 dialout 组：sudo usermod -a -G dialout $USER）。