如何在 websockets 中使用多线程？

Question

我正在尝试使用多线程的 websocket (https://websockets.readthedocs.io/en/stable/)。我想要的是在我的程序继续运行时继续获取数据，但是，当我像下面的代码那样做时，服务器没有接收到来自客户端的任何连接。我之前在主线程上运行过它，它运行良好。

async def hello(websocket, path):
    while True:
        data = await websocket.recv()
        print(data)
        await websocket.send(data)


def between_callback():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    ws_server = websockets.serve(hello, '192.168.0.115', 8899)
    
    loop.run_until_complete(ws_server)
    loop.close()

if __name__ == "__main__":
    _thread = threading.Thread(target=between_callback)
    _thread.start()

  # Do something in main thread

Answer 1

查看websockets 示例程序，您似乎缺少一条语句loop.run_forever()，这似乎确实有所不同（我还将IP 地址更改为“localhost”以进行测试）：

def between_callback():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    ws_server = websockets.serve(hello, 'localhost', 8899)

    loop.run_until_complete(ws_server)
    loop.run_forever() # this is missing
    loop.close()

演示程序

注意更新的hello 函数，它避免了 websockets.exceptions.ConnectionClosedOK: code = 1000 (OK), no reason 异常终止，我怀疑这是由于守护线程。

import websockets
import threading
import asyncio

async def hello(websocket, path):
    async for data in websocket:
        print(f"Received: '{data}'")
        await websocket.send(data)

def between_callback():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    ws_server = websockets.serve(hello, 'localhost', 8899)

    loop.run_until_complete(ws_server)
    loop.run_forever() # this is missing
    loop.close()

async def send_receive_message(uri):
    async with websockets.connect(uri) as websocket:
        await websocket.send('This is some text.')
        reply = await websocket.recv()
        print(f"The reply is: '{reply}'")

def client():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(send_receive_message('ws://localhost:8899'))
    loop.close()

if __name__ == "__main__":
    # daemon server thread:
    server = threading.Thread(target=between_callback, daemon=True)
    server.start()
    client = threading.Thread(target=client)
    client.start()
    client.join()

打印：

Received: 'This is some text.'
The reply is: 'This is some text.'

Answer 2

通常在这些服务器/客户端程序中，您在主线程中接受连接，并且每个传入连接都接收一个其他线程。线程本身可能只是一个回显函数，但这种设计的巨大优势在于您可以一次处理多个客户端，而无需等待另一个客户端收到它的答案。一个很好的方法是创建一个给定大小（比如 20）的线程池，如果它不为空，则从该池中为每个传入连接分配一个线程。这样您就可以在一定程度上控制正在使用的资源量。