PyQt结合asyncio websockets与QThread报错求助

错误原因 #

你在主线程创建并设置了event loop，但asyncio.set_event_loop是线程局部生效的——这个设置仅对主线程有效，子线程（WSThread的run方法运行的线程）的线程局部存储里并没有绑定该循环。当handler或broadcast中调用asyncio.create_task时，无法找到当前线程的活跃事件循环，因此抛出RuntimeError: There is no current event loop in thread 'Dummy-3'。

修正后的完整代码 #

1. 改进的Websocket线程类 #

import asyncio
import websockets
from PyQt5.QtCore import QThread, QMutex
from PyQt5.QtWidgets import QDialog, QPushButton, QVBoxLayout

class WSThread(QThread):
    def __init__(self, port, parent=None):
        super(WSThread, self).__init__(parent)
        self.clients = []
        self.port = port
        self.loop = None
        self.clients_mutex = QMutex()  # 锁保护客户端列表，避免并发修改冲突

    def run(self):
        # 子线程内部创建并绑定事件循环，确保线程局部生效
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        server = websockets.serve(self.handler, "", self.port)
        self.loop.run_until_complete(server)
        self.loop.run_forever()

    async def send(self, websocket, message):
        try:
            await websocket.send(message)
        except websockets.ConnectionClosed:
            # 连接关闭时从列表移除失效客户端
            self.clients_mutex.lock()
            if websocket in self.clients:
                self.clients.remove(websocket)
            self.clients_mutex.unlock()

    async def broadcast(self, message):
        # 先复制客户端列表，减少锁持有时间
        self.clients_mutex.lock()
        clients_copy = self.clients.copy()
        self.clients_mutex.unlock()
        
        if clients_copy:
            await asyncio.gather(*[self.send(ws, message) for ws in clients_copy])

    async def handler(self, websocket, path):
        # 添加新客户端到列表
        self.clients_mutex.lock()
        self.clients.append(websocket)
        self.clients_mutex.unlock()
        
        try:
            async for message in websocket:
                await self.broadcast(message)
        finally:
            # 确保连接断开时移除客户端
            self.clients_mutex.lock()
            if websocket in self.clients:
                self.clients.remove(websocket)
            self.clients_mutex.unlock()

2. 改进的对话框类 #

class MyDialog(QDialog):
    def __init__(self, port, parent=None):
        super(MyDialog, self).__init__(parent)
        self.port = port
        self.ws_thread = None

        tb = QPushButton("Start WS")
        tb.clicked.connect(self.start_ws)
        layout = QVBoxLayout()
        layout.addWidget(tb)
        self.setLayout(layout)

    def start_ws(self):
        # 避免重复启动线程
        if self.ws_thread is None or not self.ws_thread.isRunning():
            self.ws_thread = WSThread(self.port)
            self.ws_thread.start()

关键修改点 #

• 事件循环线程绑定：不再从主线程传入循环，改为在子线程的run方法内创建并绑定，确保子线程的asyncio操作能找到对应循环。
• 客户端列表线程安全：添加QMutex保护clients列表，避免并发场景下的列表修改冲突。
• 失效连接清理：在连接关闭时自动移除客户端，避免广播消息时尝试向失效连接发送数据。
• 线程启动防护：添加判断逻辑，防止重复启动websocket线程。

内容的提问来源于stack exchange，提问作者dmjf08