如何解决PyQt独立QThread中WebSocket跨线程消息发送失败问题

我开发了一个PyQt GUI应用，同时在独立QThread中用Uvicorn运行FastAPI Web服务器，服务器包含WebSocket端点。我想通过主线程界面的按钮，向这个WebSocket发送消息。

尝试用Queue机制实现，但点击按钮向队列添加内容后，队列始终为空，代码如下：

import sys
from queue import Queue

import uvicorn
from fastapi import FastAPI, WebSocket
from PySide2 import QtCore, QtWidgets

app = FastAPI()

queue = Queue()


@app.websocket("/")
async def websocket_endpoint(websocket: WebSocket):

    await websocket.accept()

    if not queue.empty():
        msg = queue.get()
        await websocket.send_text(msg)

    while True:
        incoming = await websocket.receive_text()
        print("Received Message from Client: ", incoming)


class UvicornThread(QtCore.QThread):
    def run(self):
        return uvicorn.run(app)


class MainWindow(QtWidgets.QWidget):
    def __init__(self):
        super().__init__()
        self.uvicornThread = UvicornThread()
        self.uvicornThread.start()

        self.mainLayout = QtWidgets.QVBoxLayout()
        self.setLayout(self.mainLayout)

        self.button = QtWidgets.QPushButton("Send Message")
        self.mainLayout.addWidget(self.button)
        self.button.clicked.connect(lambda: queue.put("Sending Message To Client..."))


if __name__ == "__main__":
    qapp = QtWidgets.QApplication(sys.argv)
    win = MainWindow()
    win.show()
    sys.exit(qapp.exec_())

问题根源 #

1. 队列仅初始化时检查一次：WebSocket端点只在连接建立时检查队列是否有消息，之后就进入等待客户端消息的死循环，后续按钮添加的消息完全没机会被处理。
2. 同步队列不适配异步环境：标准queue.Queue是线程安全的，但在FastAPI的异步上下文里直接使用，无法配合事件循环持续监听队列变化。

修复方案 #

改用异步安全的asyncio.Queue，并在WebSocket端点中启动独立异步任务持续监听队列；同时处理Qt主线程（同步）调用异步队列方法的线程安全问题。

修改后的完整代码：

import sys
import asyncio
from fastapi import FastAPI, WebSocket
from PySide2 import QtCore, QtWidgets
import uvicorn

app = FastAPI()
# 使用异步队列适配FastAPI的异步环境
async_queue = asyncio.Queue()
# 保存FastAPI的事件循环引用，用于跨线程调用
loop = None


@app.websocket("/")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    print("Client connected")

    # 定义异步任务：持续监听队列，有消息就发送给客户端
    async def send_from_queue():
        while True:
            msg = await async_queue.get()
            await websocket.send_text(msg)
            async_queue.task_done()

    # 启动监听队列的任务
    asyncio.create_task(send_from_queue())

    # 持续接收客户端消息
    while True:
        incoming = await websocket.receive_text()
        print("Received Message from Client: ", incoming)


class UvicornThread(QtCore.QThread):
    def run(self):
        global loop
        # 获取FastAPI运行的事件循环
        loop = asyncio.get_event_loop()
        uvicorn.run(app, host="127.0.0.1", port=8000)


class MainWindow(QtWidgets.QWidget):
    def __init__(self):
        super().__init__()
        self.uvicornThread = UvicornThread()
        self.uvicornThread.start()

        self.mainLayout = QtWidgets.QVBoxLayout()
        self.setLayout(self.mainLayout)

        self.button = QtWidgets.QPushButton("Send Message")
        self.mainLayout.addWidget(self.button)
        self.button.clicked.connect(self.send_msg_to_websocket)

    def send_msg_to_websocket(self):
        msg = "Sending Message To Client..."
        # 跨线程调用异步队列的put方法，确保线程安全
        if loop is not None:
            asyncio.run_coroutine_threadsafe(async_queue.put(msg), loop)
            print("Message added to queue")


if __name__ == "__main__":
    qapp = QtWidgets.QApplication(sys.argv)
    win = MainWindow()
    win.show()
    sys.exit(qapp.exec_())

关键修改点 #

• 替换queue.Queue为asyncio.Queue，适配FastAPI的异步运行环境
• 在WebSocket端点中启动独立的异步任务send_from_queue，持续监听队列并发送消息
• 保存FastAPI的事件循环引用，在Qt主线程中用asyncio.run_coroutine_threadsafe安全调用异步队列的put方法
• 将按钮点击的lambda表达式改为独立方法，提升代码可读性和可维护性

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