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

我需要异步读取 StdIn 以获取消息（由 \r\n 终止的 json），并在处理异步后将更新的消息写入 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

如何异步执行相同的操作？

这是一个回声的例子stdin to stdout using 异步流（对于 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)
        await writer.drain()


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

作为即用型解决方案，您可以使用控制台图书馆。它实现了类似的方法，但还提供了其他有用的异步等效方法input, print, exec and code.interact:

from aioconsole import get_standard_streams

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

Update:

让我们试着弄清楚这个函数是如何实现的connect_stdin_stdout works.

1. 获取当前事件循环：

loop = asyncio.get_event_loop()

1. Create StreamReader实例。

reader = asyncio.StreamReader()

通常，StreamReader/StreamWriter类不打算直接实例化，而只能用作诸如以下函数的结果open_connection() and start_server(). StreamReader为某些数据流提供缓冲异步接口。一些源代码（库代码）调用其函数，例如feed_data, feed_eof，数据被缓冲并且可以使用读取有记录的接口协程read(), readline()， 等等。

1. Create StreamReaderProtocol实例。

protocol = asyncio.StreamReaderProtocol(reader)

该类派生于asyncio.Protocol and FlowControlMixin并有助于适应之间Protocol and StreamReader。它会覆盖这样的Protocol方法如data_received, eof_received并打电话StreamReader方法feed_data.

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

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

The connect_read_pipe函数作为pipe参数一个类似文件的对象。stdin是一个类似文件的对象。从现在开始，所有数据都从stdin将陷入StreamReaderProtocol然后传入StreamReader

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

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

In connect_write_pipe您需要传递一个协议工厂来创建实现流控制逻辑的协议实例StreamWriter.drain()。这个逻辑是在类中实现的FlowControlMixin. Also StreamReaderProtocol继承自它。

1. Create StreamWriter实例。

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

此类使用函数转发传递给它的数据write(), writelines()等等到底层transport.

protocol用于支持drain()函数等待底层传输刷新其内部缓冲区并可再次写入的时刻。

reader是一个可选参数，可以是None，它也用于支持drain()函数，在此函数开始时，检查是否为读取器设置了异常，例如，由于连接丢失（与套接字和双向连接相关），然后drain()也会抛出异常。

您可以阅读更多有关StreamWriter and drain()在这个伟大的功能中answer.

更新2：

读取行\r\n分隔器阅读直到可以使用