如何从其他线程安全关闭Python Websocket服务器？

我编写了一个WebSocket服务器类，其中的stop()方法会在按下Control+C时被调用（从另一个线程访问）。当前调用stop()后，脚本无法正常结束，也打印不出‘Server closed’；只有客户端在调用stop()后发送消息，脚本才会真正退出并打印该信息。代码如下：

import asyncio
import websockets

DOMAIN = 'localhost'
PORT = 1111

class EchoServer:
    loop = None

    def stop(self): # METHOD CALLED FROM ANOTHER THREAD
        self.loop.stop()

    def process_message(self, message):
        print(message)
        return message

    async def handle(self, ws_client):
        print('Listening')
        async for message in ws_client:
            await ws_client.send(message)

    async def main(self):
        start_server = websockets.serve(self.handle, DOMAIN, PORT)
        asyncio.ensure_future(start_server)

    def start(self):
        self.loop = asyncio.get_event_loop()
        self.loop.create_task(self.main())
        print('Starting server...')
        self.loop.run_forever()

        print('Server closed')

• loop.stop()仅会停止事件循环，但WebSocket服务器的监听任务、已建立的客户端连接仍处于活跃状态，事件循环还有未完成的任务，因此不会直接退出。
• 只有当客户端发消息时，handle方法里的async for会因连接状态变化触发退出，相关任务完成后，事件循环才会真正结束。

要安全停止服务器，需要：1. 关闭WebSocket服务器实例，停止接受新连接；2. 等待现有连接处理完成；3. 确保跨线程调用asyncio方法的安全性。修改后的完整代码如下：

import asyncio
import websockets

DOMAIN = 'localhost'
PORT = 1111

class EchoServer:
    loop = None
    server = None  # 保存WebSocket服务器实例

    def stop(self):  # 从另一个线程调用
        # 用线程安全的方式调用异步关闭逻辑
        self.loop.call_soon_threadsafe(self._async_stop)

    def _async_stop(self):
        # 关闭服务器，停止接受新连接
        if self.server:
            self.server.close()
            # 创建任务等待服务器完全关闭
            self.loop.create_task(self._wait_server_closed())

    async def _wait_server_closed(self):
        await self.server.wait_closed()
        # 服务器关闭后再停止事件循环
        self.loop.stop()

    def process_message(self, message):
        print(message)
        return message

    async def handle(self, ws_client):
        print('Listening')
        try:
            async for message in ws_client:
                await ws_client.send(message)
        except websockets.exceptions.ConnectionClosed:
            print('Client disconnected')

    async def main(self):
        start_server = websockets.serve(self.handle, DOMAIN, PORT)
        self.server = await start_server  # 保存启动后的服务器实例

    def start(self):
        self.loop = asyncio.get_event_loop()
        self.loop.run_until_complete(self.main())  # 确保服务器启动完成
        print('Starting server...')
        self.loop.run_forever()

        print('Server closed')

• 添加server属性保存服务器实例，用于后续主动关闭
• stop()方法中使用call_soon_threadsafe调用异步关闭逻辑，避免跨线程操作asyncio的线程安全问题
• _async_stop先关闭服务器，再等待服务器完全关闭后才停止事件循环，确保所有任务正常收尾
• main()中改用await start_server获取服务器实例，保证服务器启动完成后再进入事件循环
• handle方法中捕获ConnectionClosed异常，优雅处理客户端断开场景

修改后调用stop()时，服务器会先停止接受新连接，等待现有连接处理完毕，再停止事件循环，脚本就能正常退出并打印‘Server closed’。

内容的提问来源于stack exchange，提问作者Oliver Mohr Bonometti