python - Websockets，异步处理用户输入。获取连接关闭错误 1006

Question

我正在尝试运行 websocket客户端不断地从终端接收用户输入，并将其发送到服务器。
只要该人定期输入内容，它就可以正常工作。但是如果用户超过大约 20 秒未能发送消息，则会抛出以下错误消息。

websockets.exceptions.ConnectionClosedError: code = 1006 (connection closed abnormally [internal]), no reason

我怀疑这可能是因为 input()等待用户输入的函数是一个阻塞函数。也许这会阻止客户端 ping 服务器？
我试图包装 input()在异步函数中等待用户输入的函数。但我仍然遇到同样的问题。
我该如何解决这个问题？
这是我使用的代码。
代码
客户

import asyncio
import websockets

async def producer():
    return input("Enter something: ")

async def echo_loop():
    uri = f"wss://192.168.1.2:5432"
    async with websockets.connect(uri, ssl=None) as websocket:
        while True:
            # Get user input
            # msg = input("Enter something: ")
            msg = await producer()

            # Send message to the server
            await websocket.send(msg)
            print(f"> {msg}")

            # Get feedback from server
            feedback = await websocket.recv()
            print(f"< {feedback}")


asyncio.get_event_loop().run_until_complete(echo_loop())
asyncio.get_event_loop().run_forever()

服务器

import asyncio
import websockets

async def echo(websocket, path):
    async for msg in websocket:
        # Send Feedback of message to client
        feedback = f"RECEIVED: {msg}!"
        await websocket.send(feedback)
        print(f"> {feedback}")

start_server = websockets.serve(echo, host=None, port=5432, ssl=None)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

补充说明:
根据 documentation FAQ我知道 ConnectionClosedError: code = 1006有时会在出现网络问题时发生，例如:

两台计算机之间的互联网连接中断。

如果代理有问题。

如果 HTTP 负载平衡器出现问题

但是，这不是导致它的 Internet 或代理问题。
我还查看了有关堆栈溢出的其他一些相关问题，但没有一个能回答我的问题:

Python Websockets-8.1 ConnectionClosedError

Error 1006 on websocket

WebSocket connection is closed: code = 1006

日志
以下是客户端上的日志，从进行某些输入的那一刻到引发异常之前:

client > Frame(fin=True, opcode=1, data=b'entering some text here after a long delay', rsv1=False, rsv2=False, rsv3=False)
> entering some text here after a long delay
client - event = data_received(<10 bytes>)
client < Frame(fin=True, opcode=9, data=b'\xb4Y\xbd\xa2', rsv1=False, rsv2=False, rsv3=False)
client - received ping, sending pong: b459bda2
client > Frame(fin=True, opcode=10, data=b'\xb4Y\xbd\xa2', rsv1=False, rsv2=False, rsv3=False)
client > Frame(fin=True, opcode=9, data=b'\xbe\x16\x06\xe7', rsv1=False, rsv2=False, rsv3=False)
client - event = connection_lost([Errno 32] Broken pipe)
client - state = CLOSED
client x code = 1006, reason = [no reason]
client - aborted pending ping: be1606e7
client ! failing CLOSED WebSocket connection with code 1006
Error in data transfer

这是来自服务器的日志。

server > Frame(fin=True, opcode=9, data=b'\xb4Y\xbd\xa2', rsv1=False, rsv2=False, rsv3=False)
server ! timed out waiting for pong
server ! failing OPEN WebSocket connection with code 1011
server - state = CLOSING
server > Frame(fin=True, opcode=8, data=b'\x03\xf3', rsv1=False, rsv2=False, rsv3=False)
server x half-closing TCP connection
server ! timed out waiting for TCP close
server x closing TCP connection
server - event = connection_lost(None)
server - state = CLOSED
server x code = 1006, reason = [no reason]
server - aborted pending ping: b459bda2
Error in connection handler

我怀疑 server ! timed out waiting for pong线是重要的。到目前为止，我的理解是由于等待或用户输入，客户端无法向服务器发送 ping。

Answer 1

我自己想通了。
问题确实是input()客户端上的功能。正如我所怀疑的那样，它阻止了程序对进入的 ping 做任何事情。
我最初认为将它包裹在 async 中协程函数就足够了。但事实证明，如果其中包含 IO 阻塞代码，它仍然不会异步运行。
我通过包装 input() 解决了这个问题asyncio 内部的函数的 run_in_executor()方法，使其异步运行。
这是我为使其工作所做的更改:

async def producer():
    return await asyncio.get_event_loop().run_in_executor(None, lambda: input("Enter something: "))