使用python asyncio从套接字读取时如何避免阻塞？

Question

我正在尝试使用 python asyncio 进行一些实验以改进该领域，出于自学目的，我正在尝试连接到 redis，发送一些命令并读取响应，这可能属于通用的“读取流”来自某个来源的数据”。我无法解决的问题是如何分块读取数据，因为服务器和客户端之间的连接没有关闭，并且终止序列\r\n 可能会多次遇到。如果我在没有更多数据时等待，那么调用当然会阻塞，直到收到其他数据。

class Client:
    def __init__(self, loop, host='127.0.0.1', port=6379):
        self.host = host
        self.port = port
        self.reader = None
        self.writer = None
        self.loop = loop

    @asyncio.coroutine
    def _connect(self):
        self.reader, self.writer = yield from asyncio.open_connection(
            self.host, self.port, loop=self.loop)

    async def read(self, b=4096):
        resp = b''
        while True:
            chunk = await self.reader.read(b)
            if chunk:
                resp += chunk
            else:
                break
        return resp

假设我想以 2 个字节为单位读取响应（是的，这很愚蠢，但这只是为了学习目的）所以：

loop = asyncio.get_event_loop()
client = Client(loop)
..... sends some commands here ....
resp = await client.read(2)

我不知道在不知道服务器响应长度的情况下，当响应长于从套接字读取的字节时，代码仍然是安全的。

Answer 1

我最近遇到了类似的问题。我的解决方案是继续阅读，直到阅读给定的字符（或一组字符）。这与人们说完对讲机后说“结束”的理念相同。等待响应说它已经完成谈话更容易。

虽然我之前没有使用过 asyncio 模块，但我相信以下代码应该可以解决您的问题，假设输入源以变量中指示的任何字符（或字符串）结束响应end_signal.

class Client:
    def __init__(self, loop, host='127.0.0.1', port=6379):
        self.host = host
        self.port = port
        self.reader = None
        self.writer = None
        self.loop = loop

    @asyncio.coroutine
    def _connect(self):
        self.reader, self.writer = yield from asyncio.open_connection(
            self.host, self.port, loop=self.loop)

    async def read(self, b=4096, end_signal = "10101101110111110"):
        resp = b''
        while True:
            chunk = await self.reader.read(b)
            resp += chunk
            if resp[-1*len(end_signal):] == end_signal:
                break
        return resp