python asyncio如何读取StdIn并写入StdOut？

Question

我需要异步读取 StdIn 以获取消息(json 由\r\n 终止)并在处理异步后将更新的消息写入 StdOut。
目前我正在同步进行，例如:

class SyncIOStdInOut():
    def write(self, payload: str):
        sys.stdout.write(payload)
        sys.stdout.write('\r\n')
        sys.stdout.flush()

    def read(self) -> str:
        payload=sys.stdin.readline()
        return  payload

如何异步执行相同操作？

Answer 1

这是回声 stdin 的示例至stdout使用 asyncio streams (对于 Unix)。

import asyncio
import sys


async def connect_stdin_stdout():
    loop = asyncio.get_event_loop()
    reader = asyncio.StreamReader()
    protocol = asyncio.StreamReaderProtocol(reader)
    await loop.connect_read_pipe(lambda: protocol, sys.stdin)
    w_transport, w_protocol = await loop.connect_write_pipe(asyncio.streams.FlowControlMixin, sys.stdout)
    writer = asyncio.StreamWriter(w_transport, w_protocol, reader, loop)
    return reader, writer


async def main():
    reader, writer = await connect_stdin_stdout()
    while True:
        res = await reader.read(100)
        if not res:
            break
        writer.write(res)


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

作为即用型解决方案，您可以使用 aioconsole图书馆。它实现了类似的方法，但也为 input 提供了额外有用的异步等价物。 , print , exec和 code.interact :

from aioconsole import get_standard_streams

async def main():
    reader, writer = await get_standard_streams()

更新:
让我们试着弄清楚函数 connect_stdin_stdout作品。

获取当前事件循环:

loop = asyncio.get_event_loop()

创建 StreamReader实例。

reader = asyncio.StreamReader()

一般情况下， StreamReader/StreamWriter类不打算被直接实例化，并且只能作为 open_connection() 等函数的结果使用。和 start_server() . StreamReader为某些数据流提供缓冲的异步接口(interface)。某些源(库代码)调用它的函数，例如 feed_data , feed_eof , 数据被缓冲并且可以使用 documented 读取接口(interface)协程 read() , readline() ， 等等。

创建 StreamReaderProtocol实例。

protocol = asyncio.StreamReaderProtocol(reader)

此类派生自 asyncio.Protocol和 FlowControlMixin并有助于在 Protocol 之间进行调整和 StreamReader .它覆盖了这样的 Protocol方法为 data_received , eof_received并调用 StreamReader方法 feed_data .

注册标准输入流stdin在事件循环中。

await loop.connect_read_pipe(lambda: protocol, sys.stdin)

connect_read_pipe函数采用 pipe参数一个类似文件的对象。 stdin是一个类似文件的对象。从现在开始，从 stdin 读取的所有数据将落入 StreamReaderProtocol然后传入 StreamReader

注册标准输出流stdout在事件循环中。

w_transport, w_protocol = await loop.connect_write_pipe(FlowControlMixin, sys.stdout)

在 connect_write_pipe您需要传递一个协议(protocol)工厂，该工厂创建协议(protocol)实例来实现 StreamWriter.drain() 的流控制逻辑.此逻辑在类 FlowControlMixin 中实现.还有 StreamReaderProtocol继承自它。

创建 StreamWriter实例。

writer = asyncio.StreamWriter(w_transport, w_protocol, reader, loop)

此类使用函数 write() 转发传递给它的数据, writelines()等到基础 transport . protocol用于支持 drain()函数等待底层传输已刷新其内部缓冲区并可再次写入。 reader是可选参数，可以是 None ，也用于支持 drain()函数，在此函数开始时检查是否为阅读器设置了异常，例如，由于连接丢失(与套接字和双向连接相关)，然后 drain()也会抛出异常。
您可以阅读有关 StreamWriter 的更多信息和 drain()功能在这个伟大的 answer .
更新 2:
使用 \r\n 读取行分隔符 readuntil可以使用