我正在嘗試使用 asyncio 和 Unix 域套接字在 Python 中構(gòu)建一個(gè)玩具內(nèi)存 Redis 服務(wù)器。baz我的最小示例只返回每個(gè)請(qǐng)求的值：import asyncioclass 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()當(dāng)我使用以下腳本使用客戶端庫(kù)測(cè)試兩個(gè)連續(xù)的客戶端請(qǐng)求時(shí)，它可以redis工作：import timeimport redisr = redis.Redis(unix_socket_path="/tmp/redis.sock")r.get("foo")time.sleep(1)r.get("bar")但是，如果我刪除time.sleep(1)，有時(shí)它會(huì)起作用，有時(shí)第二個(gè)請(qǐng)求會(huì)失敗，并出現(xiàn)以下任一情況：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 pipeDuring 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.看來(lái)我的實(shí)現(xiàn)缺少客戶端庫(kù)期望的一些關(guān)鍵行為（可能是由于它是異步的）。我缺少什么？
1 回答
拉風(fēng)的咖菲貓
TA貢獻(xiàn)1995條經(jīng)驗(yàn) 獲得超2個(gè)贊
write_eof()如果您想在每次請(qǐng)求后關(guān)閉套接字，則需要使用
緩沖的寫(xiě)入數(shù)據(jù)刷新后，關(guān)閉流的寫(xiě)入端。
您的代碼稍加修改將如下所示：
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()
通常，您不會(huì)在每次請(qǐng)求后關(guān)閉套接字。
以下示例僅用于說(shuō)明目的，旨在表明套接字不需要關(guān)閉。當(dāng)然，您總是會(huì)讀取一行，然后根據(jù) Redis 協(xié)議解釋數(shù)據(jù)。我們知道這里發(fā)送了兩個(gè) GET 命令（每行 5 行，包含 2 個(gè)元素的數(shù)組的指示符，字符串的指示符，字符串值“GET”，以及字符串指示符和相應(yīng)的值，即鍵）
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()
在客戶端發(fā)送是這樣完成的：
print(r.get("foo"))
print(r.get("bar"))
time.sleep(1)
最后一次time.sleep是為了確?？蛻舳瞬粫?huì)立即退出。
控制臺(tái)上的輸出是：
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'
請(qǐng)注意，start僅輸出一次，這表明我們可以處理多個(gè)請(qǐng)求，而不必立即關(guān)閉套接字。
添加回答
舉報(bào)
0/150
提交
取消