PyQt/PySide线程中集成异步Bleak库的问题求解

你的问题根源在于每次调用asyncio.run()都会创建并销毁一个新的asyncio事件循环，同一个线程中重复执行此操作会导致异常或异步任务无法正常运行。正确的做法是在工作线程中维护一个持久的asyncio事件循环，所有Bleak异步操作都基于这个循环执行。

具体实现步骤 #

1. 重构CoreWorker线程，维护持久事件循环 #

修改workerCore.py，在线程启动时创建并绑定唯一的事件循环，所有异步任务都通过该循环执行：

import queueCmds  as qc
from threading import Thread
import asyncio
from bleak import BleakScanner, BleakClient
from PySide6.QtCore import Signal, QObject

CHARACTERISTIC_UUID = "...."

class CoreWorker(Thread, QObject):
    # 定义信号，用于将BLE数据传递到UI线程
    sigBLEDevicesFound = Signal(list)
    sigBLEReadValue = Signal(bytes)

    def __init__(self, app, parent=None):
        Thread.__init__(self, parent)
        QObject.__init__(self, parent)
        self.app = app

    def run(self):
        # 创建并绑定当前线程的asyncio事件循环
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        
        try:
            while True:
                rxd = self.app.queueCore.get()
                if rxd.cmd == qc.CoreQC.SCAN_BLE:
                    # 用持久循环执行扫描任务
                    devices = self.loop.run_until_complete(self.scan_devices())
                    # 发送信号到UI线程更新数据
                    self.sigBLEDevicesFound.emit([(d.name, d.address) for d in devices])
                elif rxd.cmd == qc.CoreQC.CONNECT_AND_READ:
                    # 假设rxd携带目标设备地址
                    value = self.loop.run_until_complete(self.connect_and_read(rxd.device_address))
                    self.sigBLEReadValue.emit(value)
                # 标记队列任务完成
                self.app.queueCore.task_done()
        except Exception as e:
            print(f"Worker线程错误: {e}")
        finally:
            # 线程退出时关闭循环
            self.loop.close()

    async def scan_devices(self):
        devices = await BleakScanner.discover()
        for i, device in enumerate(devices):
            print(f"{i}: {device.name} ({device.address})")
        return devices

    async def connect_and_read(self, device_address):
        async with BleakClient(device_address) as client:
            print(f"已连接到 {device_address}")
            value = await client.read_gatt_char(CHARACTERISTIC_UUID)
            print(f"读取到的值: {value}")
            return value

2. 在UI线程中绑定信号，处理异步结果 #

修改AppWindow类，绑定Worker的信号，确保UI更新操作在主线程执行：

import os
from pathlib import Path
import sys

from PySide6.QtCore import QObject, Slot, Signal
from PySide6.QtQml import QQmlApplicationEngine
from PySide6.QtWidgets import QApplication 

from workerCore import CoreWorker
from queue import Queue
from threading import Thread

import queueCmds  as qc

class AppWindow(QObject):
    sigPairingPopUp = Signal()    

    def __init__(self):
        super().__init__(None)
        self.queueCore = Queue()
        self.workerCore = CoreWorker(self)
        # 绑定Worker信号到UI处理函数
        self.workerCore.sigBLEDevicesFound.connect(self.handleBLEDevices)
        self.workerCore.sigBLEReadValue.connect(self.handleBLEValue)
        # 启动Worker线程
        self.workerCore.start()   

    @Slot()
    def scanBLE(self):
        sd = qc.QueueData()
        sd.cmd = qc.CoreQC.SCAN_BLE
        self.queueCore.put(sd)
    
    def handleBLEDevices(self, device_list):
        # 在此处理扫描到的设备，例如更新QML界面
        print("UI线程收到BLE设备列表:", device_list)
    
    def handleBLEValue(self, value):
        # 处理读取到的BLE特征值
        print("UI线程收到BLE数值:", value)

if __name__ == "__main__":
    main = QApplication(sys.argv)
    engine = QQmlApplicationEngine()
    appWin = AppWindow()
    engine.rootContext().setContextProperty("backend", appWin)
    engine.load(os.fspath(Path(__file__).resolve().parent / "qml/main.qml"))
    if not engine.rootObjects():
        sys.exit(-1)
    sys.exit(main.exec())

关键说明 #

• 持久事件循环：Worker线程启动时创建唯一事件循环，所有异步任务通过loop.run_until_complete()执行，避免asyncio.run()重复创建/销毁循环的问题。
• 线程安全UI更新：通过Qt的Signal机制将异步结果传递到UI线程，确保所有UI操作在主线程执行，符合Qt线程模型要求。
• 队列任务管理：保留原线程安全队列传递命令，Worker线程持续监听队列，处理同步命令并触发异步操作。

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