如何在PyQt5程序中结合使用Trio异步框架？

要让Trio与PyQt5在同一线程协同运行，核心是利用PyQt的信号机制实现Trio所需的两个关键函数：

• run_sync_soon_threadsafe：通过PyQt的元对象系统，以线程安全的方式将Trio任务投递到PyQt事件循环
• done_callback：直接调用即可，因为两者运行在同一线程，无需额外线程同步处理

以下是修改后的完整示例代码：

import sys
import trio
from PyQt5.QtWidgets import QApplication, QWidget, QLabel, QVBoxLayout
from PyQt5.QtCore import QMetaObject, Qt, Q_ARG

class MainWindow(QWidget):
    def __init__(self):
        super().__init__()
        self.init_ui()
        # 启动Trio guest run，绑定PyQt适配的回调函数
        trio.lowlevel.start_guest_run(
            self.trio_main,
            run_sync_soon_threadsafe=self.run_sync_soon_threadsafe,
            done_callback=self.done_callback
        )

    def init_ui(self):
        self.setWindowTitle("PyQt5 + Trio Demo")
        self.label = QLabel("等待更新...", self)
        layout = QVBoxLayout()
        layout.addWidget(self.label)
        self.setLayout(layout)
        self.show()

    def run_sync_soon_threadsafe(self, func):
        # 用PyQt的队列连接方式实现线程安全的任务调度
        QMetaObject.invokeMethod(
            self,
            "_execute_trio_task",
            Qt.QueuedConnection,
            Q_ARG(object, func)
        )

    def _execute_trio_task(self, func):
        # 实际执行Trio投递的任务
        func()

    def done_callback(self):
        # Trio事件循环结束后的清理回调
        print("Trio guest run 已终止")

    async def update_label_timer(self):
        count = 0
        while True:
            # 同一线程下可直接更新UI组件
            self.label.setText(f"已更新 {count} 次")
            count += 1
            await trio.sleep(1)

    async def trio_main(self):
        # Trio主任务入口，可同时启动UI更新和WebSocket连接
        async with trio.Nursery() as nursery:
            nursery.start_soon(self.update_label_timer)
            # 在此添加WebSocket连接任务示例
            # nursery.start_soon(self.websocket_client)

    async def websocket_client(self):
        # 这里编写Trio WebSocket连接逻辑
        # 示例框架：
        # async with trio_websocket.connect_websocket(
        #     trio.hazmat.current_async_token(),
        #     "ws://your-server-address",
        #     "/target-path"
        # ) as ws:
        #     while True:
        #         received_msg = await ws.get_message()
        #         print(f"服务器消息: {received_msg}")
        #         await ws.send_message("客户端响应")
        pass

if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = MainWindow()
    sys.exit(app.exec_())

关键细节说明 #

1. 任务调度适配：run_sync_soon_threadsafe通过QMetaObject.invokeMethod将Trio任务放入PyQt事件队列，确保任务在UI线程执行，避免线程安全问题。
2. 同一线程运行：Trio的guest run直接附着在PyQt的事件循环上，所有异步任务（包括UI更新和网络操作）都在主线程执行，无需跨线程通信。
3. 扩展性：在trio_main的Nursery中可同时启动多个异步任务，比如UI定时器和WebSocket客户端，Trio会自动调度这些任务的执行。

内容的提问来源于stack exchange，提问作者Игорь Платонов