asyncio.start_unix_server 和 redis 的套接字错误

Dus*_*ram 2 python redis python-3.x redis-py python-asyncio

我正在尝试使用 asyncio 和 Unix 域套接字在 Python 中构建一个玩具内存 Redis 服务器。

我的最小示例只返回baz每个请求的值:

import asyncio


class RedisServer:
    def __init__(self):
        self.server_address = "/tmp/redis.sock"

    async def handle_req(self, reader, writer):
        await reader.readline()
        writer.write(b"$3\r\nbaz\r\n")
        await writer.drain()
        writer.close()
        await writer.wait_closed()

    async def main(self):
        server = await asyncio.start_unix_server(self.handle_req, self.server_address)
        async with server:
            await server.serve_forever()

    def run(self):
        asyncio.run(self.main())


RedisServer().run()
Run Code Online (Sandbox Code Playgroud)

当我测试有两个连续的客户端请求redis客户端库与下面的脚本,它的工作原理:

import time
import redis


r = redis.Redis(unix_socket_path="/tmp/redis.sock")

r.get("foo")
time.sleep(1)
r.get("bar")
Run Code Online (Sandbox Code Playgroud)

但是,如果我删除time.sleep(1),有时它会起作用,有时第二个请求会失败,并且会出现以下任一情况:

Traceback (most recent call last):
  File "/tmp/env/lib/python3.8/site-packages/redis/connection.py", line 706, in send_packed_command
    sendall(self._sock, item)
  File "/tmp/env/lib/python3.8/site-packages/redis/_compat.py", line 9, in sendall
    return sock.sendall(*args, **kwargs)
BrokenPipeError: [Errno 32] Broken pipe

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "test.py", line 9, in <module>
    r.get("bar")
  File "/tmp/env/lib/python3.8/site-packages/redis/client.py", line 1606, in get
    return self.execute_command('GET', name)
  File "/tmp/env/lib/python3.8/site-packages/redis/client.py", line 900, in execute_command
    conn.send_command(*args)
  File "/tmp/env/lib/python3.8/site-packages/redis/connection.py", line 725, in send_command
    self.send_packed_command(self.pack_command(*args),
  File "/tmp/env/lib/python3.8/site-packages/redis/connection.py", line 717, in send_packed_command
    raise ConnectionError("Error %s while writing to socket. %s." %
redis.exceptions.ConnectionError: Error 32 while writing to socket. Broken pipe.
Run Code Online (Sandbox Code Playgroud)

或者:

Traceback (most recent call last):
  File "test.py", line 9, in <module>
    r.get("bar")
  File "/tmp/env/lib/python3.8/site-packages/redis/client.py", line 1606, in get
    return self.execute_command('GET', name)
  File "/tmp/env/lib/python3.8/site-packages/redis/client.py", line 901, in execute_command
    return self.parse_response(conn, command_name, **options)
  File "/tmp/env/lib/python3.8/site-packages/redis/client.py", line 915, in parse_response
    response = connection.read_response()
  File "/tmp/env/lib/python3.8/site-packages/redis/connection.py", line 739, in read_response
    response = self._parser.read_response()
  File "/tmp/env/lib/python3.8/site-packages/redis/connection.py", line 324, in read_response
    raw = self._buffer.readline()
  File "/tmp/env/lib/python3.8/site-packages/redis/connection.py", line 256, in readline
    self._read_from_socket()
  File "/tmp/env/lib/python3.8/site-packages/redis/connection.py", line 201, in _read_from_socket
    raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
redis.exceptions.ConnectionError: Connection closed by server.
Run Code Online (Sandbox Code Playgroud)

似乎我的实现缺少客户端库期望的一些关键行为(可能是因为它是异步的)。我错过了什么?

Ste*_*cht 7

如果您希望每个请求后关闭套接字,你需要使用write_eof(),这

缓冲的写入数据刷新后关闭流的写入端。

参见 docs.python.org -> write_eof

您的代码稍作修改,如下所示:

async def handle_req(self, reader, writer):
    await reader.readline()
    writer.write(b"$3\r\nbaz\r\n")
    await writer.drain()
    writer.write_eof()
    writer.close()
    await writer.wait_closed()
Run Code Online (Sandbox Code Playgroud)

通常,您不会在每次请求后关闭套接字。

以下示例仅用于说明目的,旨在表明不需要关闭套接字。当然,您总是会读取一行,然后根据 Redis 协议解释数据。我们知道这里发送了两个 GET 命令(每行 5 行,包含 2 个元素的数组指示符,字符串指示符,字符串值 'GET' 和字符串指示符和相应的值,即键)

async def handle_req(self, reader, writer):
    print("start")
    for i in range(0, 2):
        for x in range(0, 5):
            print(await reader.readline())
        writer.write(b"$3\r\nbaz\r\n")
        await writer.drain()
    writer.write_eof()
    writer.close()
    await writer.wait_closed()
Run Code Online (Sandbox Code Playgroud)

在客户端发送是这样完成的:

print(r.get("foo"))
print(r.get("bar"))
time.sleep(1)
Run Code Online (Sandbox Code Playgroud)

最后一次.sleep 是为了确保客户端不会立即退出。

控制台上的输出是:

start
b'*2\r\n'
b'$3\r\n'
b'GET\r\n'
b'$3\r\n'
b'foo\r\n'
b'*2\r\n'
b'$3\r\n'
b'GET\r\n'
b'$3\r\n'
b'bar\r\n'
Run Code Online (Sandbox Code Playgroud)

请注意,start它只输出一次,这表明我们可以处理多个请求而不必立即关闭套接字。