Python 3.8下pyserial-asyncio串口客户端/服务端无法即时通信问题

我在树莓派5的bash环境下使用Python 3.8（无法升级版本），基于pyserial-asyncio开发串口客户端/服务端，此前已成功实现TCP客户端/服务端，这是首次尝试串口异步通信。

服务端代码 #

import asyncio
import serial_asyncio

class UARTProtocol(asyncio.Protocol):
    def __init__(self):
        self.transport = None

    def connection_made(self, transport):
        self.transport = transport
        print('Port opened', transport)

    def data_received(self, data):
        print('Data received:', data.decode())
        # Echo received data back (example)
        self.transport.write(data)
        # Close the connection if 'exit' is received
        if data == b"exit\r":
            self.transport.close()

    def connection_lost(self, exc):
        print('Port closed')
        self.transport = None

    def pause_writing(self):
        print('pause writing')
        print(self.transport.get_write_buffer_size())

    def resume_writing(self):
        print(self.transport.get_write_buffer_size())
        print('resume writing')

async def run_uart_server():
    loop = asyncio.get_running_loop()
    try:
        transport, protocol = await serial_asyncio.create_serial_connection(loop, UARTProtocol, '/dev/ttyAMA2', baudrate=9600)
        print("UART server started.")
        await asyncio.Future()  # Run forever
    except serial.serialutil.SerialException as e:
        print(f"Error: Could not open serial port: {e}")
    finally:
        if transport:
            transport.close()

if __name__ == "__main__":
    asyncio.run(run_uart_server())

客户端代码 #

import asyncio
import serial_asyncio

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")

        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:
                message = input("Enter message to send (or 'exit' to quit): ")
                if message.lower() == 'exit':
                    break
                writer.write((message + '\n').encode())
                # writer.write_eof()
                await writer.drain()
                print(f"Sent: {message}")

        await asyncio.gather(receive_data(), send_data())

    except serial.SerialException as e:
        print(f"Error opening serial port: {e}")
    finally:
      if 'writer' in locals():
        writer.close()
        await writer.wait_closed()
        print("Connection closed.")

if __name__ == "__main__":
    asyncio.run(uart_client('/dev/ttyAMA1', 9600))

问题现象 #

预期客户端输入文本后能立即发送至服务端并打印，但实际服务端要等客户端输入exit关闭连接后，才会一次性打印所有之前输入的文本。尝试添加writer.write_eof()后，服务端能立即收到数据，但客户端无法再提示输入。另外，在bash中执行echo foo > /dev/ttyAMA1时，服务端会立即打印foo，说明问题出在客户端代码。

问题原因及解决方案 #

1. 同步输入阻塞异步事件循环 #

客户端send_data函数中使用的input()是同步阻塞函数，会卡住整个asyncio事件循环，导致await writer.drain()无法及时执行，发送缓冲区的数据一直堆积，直到连接关闭才批量发送。

由于Python 3.8没有asyncio.to_thread()，可以用loop.run_in_executor()将同步的input()放到线程池中执行，避免阻塞事件循环。

2. 换行符不匹配 #

客户端发送的是message + '\n'（LF换行），但服务端判断退出的条件是data == b"exit\r"（CR换行），同时串口数据触发data_received的时机和换行符也有关联，统一换行符能避免不必要的问题。

修改后的客户端send_data函数 #

async def send_data():
    loop = asyncio.get_running_loop()
    while True:
        # 将同步input放到线程执行，避免阻塞事件循环
        message = await loop.run_in_executor(None, input, "Enter message to send (or 'exit' to quit): ")
        if message.lower() == 'exit':
            break
        # 发送CR换行，和服务端的退出判断匹配
        writer.write((message + '\r').encode())
        await writer.drain()
        print(f"Sent: {message}")

可选：服务端兼容多种换行符 #

如果需要服务端兼容LF或CRLF，可以修改data_received中的退出判断：

if data.strip() == b"exit":
    self.transport.close()

这样修改后，客户端输入的消息会立即发送到服务端，服务端能实时打印收到的数据，同时客户端可以正常持续输入。

内容的提问来源于stack exchange，提问作者Ed Morton