如何在类中使用 Python 的 websockets 和 asyncio 以及现有的事件循环

Question

我目前正在尝试将 websocket 接口集成到我的程序中。为此，我使用了 https://websockets.readthedocs.io/en/stable/intro.html 模块以及 asyncio。我目前正在努力在一个专用类中实现 websocket 功能，该类将 websocket 任务设置为在与 MyDriver 类中运行的任务相同的事件循环中并发运行。

main.py

from myDriver import MyDriver
from webSocketServer import WebSocketServer

async def main():

   # loop = asyncio.get_event_loop()
   driver = MyDriver()
   ws = WebSocketServer()
   await driver.drive()
   # The following does not integrate properly with the above. The msgHandler is not ran 
   await websockets.serve(lambda websocket, path: ws.msgHandler(websocket, path), "localhost", 5678)

asyncio.run(main())

这里的 lambda 是为了摆脱来自类的 self 参数。

webSocketServer.py

import asyncio
import websockets

class WebSocketServer:

    def __init__(self):
        print('Init')

    async def msgHandler(self, websocket, path):
        self.sendTask = asyncio.create_task(self.sendHandler(websocket, path))
        self.receiveTask = asyncio.create_task(self.receiveHandler(websocket, path))

        await asyncio.wait([self.sendTask, self.receiveTask], return_when=asyncio.FIRST_COMPLETED)

    async def sendHandler(self, websocket, path):
        while True:
            await asyncio.sleep(2)
            message = producer()
            await websocket.send(message)

    async def receiveHandler(self, websocket, path):
        async for message in websocket:
            await self.printMsg()

    async def printMsg(self, msg):
        await asyncio.sleep(0.1)
        print(msg)

    def producer():
        return 'Hi !'

我的实现基于 websockets 入门页面上提供的示例。他们使用loop.run_until_complete(server) 和loop.run_forever() API。我还尝试通过将loop in 参数传递给WebSocketServer(loop) 的构造函数并在那里执行websockets.serve(lambda websocket, path: ws.msgHandler(websocket, path), "localhost", 5678) 来使用这些，但随后我收到错误RuntimeError: This event loop is already running。我还看了loop.create_task()，它以协程为参数。

有没有人看到我可以正确集成在与我的其他任务相同的事件循环中运行的 websocket 服务器的方法？谢谢！

Answer 1

他们使用loop.run_until_complete(server) 和loop.run_forever() API。

将它们转换为asyncio.run() 很好，这是一种启动异步代码的更可靠的方式。只是您的转换不完整，缺少run_forever() 部分。需要像run_forever() 这样的东西，因为websockets.serve() 并不意味着“完成后服务并退出”，它意味着“开始服务并返回一个句柄给服务器”。这个句柄（几乎）会立即返回，如果您此时从main 返回，就像您的代码一样，程序将在处理单个连接之前退出。

您需要做的是在 main 的末尾添加另一个 await，只要程序需要运行，您就可以在其中等待。一种方法是await asyncio.Event().wait()，它将永远等待，并且与loop.run_forever() 等效。而且由于您可以使用服务器对象，因此您可以使用它的wait_closed() 方法来等待，只要服务器正在运行（即协程通过调用server.close() 有意地关闭它），这可能是您想要的。

也就是说，需要将main()的最后一行修改为：

   server = await websockets.serve(lambda websocket, path: ws.msgHandler(websocket, path), "localhost", 5678)
   await server.wait_closed()