From baa82c895da7396b8faf580987b298abc10e5db9 Mon Sep 17 00:00:00 2001 From: wangfq Date: Tue, 7 Jul 2026 14:25:49 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20MqttClient=20=E9=87=8D=E6=9E=84=E4=B8=BA?= =?UTF-8?q?=20QObject=20=E2=80=94=20publish=20=E9=80=9A=E8=BF=87=20Qt=20?= =?UTF-8?q?=E4=BF=A1=E5=8F=B7=E6=8E=92=E9=98=9F=E9=81=BF=E5=85=8D=20C=20?= =?UTF-8?q?=E5=B1=82=E5=B4=A9=E6=BA=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 问题: paho-mqtt publish() C 扩展在 Windows 上从 GUI 线程调用时静默崩溃。 修复: - MqttClient 改为 QObject 子类 - publish() 通过 Qt Signal(Qt.QueuedConnection) 排队，_do_publish() 始终在主线程事件循环中执行 paho 调用 - 状态/消息通知改用 Qt Signal 跨线程传递 - 新增 subscribe()/unsubscribe() 公开 API，消除 main.py 直接 访问 _client 的脆弱代码 - 移除 MainWindow 中已不再需要的中间信号 _mqtt_status/_mqtt_msg --- DBNMQTTool/dbn_mqtt_tool/mqtt_client.py | 75 +++++++++++++++---------- DBNMQTTool/main.py | 16 ++---- 2 files changed, 50 insertions(+), 41 deletions(-) diff --git a/DBNMQTTool/dbn_mqtt_tool/mqtt_client.py b/DBNMQTTool/dbn_mqtt_tool/mqtt_client.py index c438743..47ed59f 100644 --- a/DBNMQTTool/dbn_mqtt_tool/mqtt_client.py +++ b/DBNMQTTool/dbn_mqtt_tool/mqtt_client.py @@ -1,7 +1,6 @@ """ MQTT 客户端封装 """ - import json import time import threading @@ -9,6 +8,7 @@ from typing import Callable, Optional from dataclasses import dataclass import paho.mqtt.client as mqtt +from PySide6.QtCore import QObject, Signal, Slot, Qt from .protocol import parse_topic_dev_serial @@ -23,67 +23,64 @@ class BrokerConfig: keepalive: int = 60 -class MqttClient: - """DLD960 IoT MQTT 客户端""" +class MqttClient(QObject): + """DLD960 IoT MQTT 客户端 — QObject 基类确保信号线程安全""" + + # 供外部监听 + _status_notify = Signal(bool, str) # connected, message + _msg_notify = Signal(str, str, object) # topic, dev_serial, payload_dict + + # 内部线程安全的 publish 队列 + _publish_sig = Signal(str, str, int) # topic, json_data, qos def __init__(self): + super().__init__() self._client: Optional[mqtt.Client] = None self._config = BrokerConfig() self._connected = False self._lock = threading.Lock() - self._message_callbacks: list[Callable[[str, str, dict], None]] = [] - self._status_callbacks: list[Callable[[bool, str], None]] = [] + + # 信号→槽: publish 通过排队连接确保在主线程的 paho context 执行 + self._publish_sig.connect(self._do_publish, Qt.QueuedConnection) @property def connected(self) -> bool: return self._connected + # ---- 公开回调 (兼容旧接口) ---- + def on_status_change(self, callback: Callable[[bool, str], None]): - self._status_callbacks.append(callback) + self._status_notify.connect(callback) def on_message(self, callback: Callable[[str, str, dict], None]): """回调参数: (topic, dev_serial, payload_dict)""" - self._message_callbacks.append(callback) + self._msg_notify.connect(callback) - def _notify_status(self, connected: bool, msg: str): - for cb in self._status_callbacks: - try: - cb(connected, msg) - except Exception: - pass - - def _notify_message(self, topic: str, payload: dict): - dev_serial = parse_topic_dev_serial(topic) or "" - for cb in self._message_callbacks: - try: - cb(topic, dev_serial, payload) - except Exception: - pass - - # ---- MQTT callbacks ---- + # ---- MQTT callbacks (paho 网络线程) ---- def _on_connect(self, client, userdata, flags, reason_code, properties=None): rc = reason_code if isinstance(reason_code, int) else reason_code.value if rc == 0: self._connected = True - self._notify_status(True, "已连接到 Broker") + self._status_notify.emit(True, "已连接到 Broker") from .protocol import TOPIC_ALL_DEVICE_UP client.subscribe(TOPIC_ALL_DEVICE_UP, qos=1) else: self._connected = False - self._notify_status(False, f"连接失败 (rc={rc})") + self._status_notify.emit(False, f"连接失败 (rc={rc})") def _on_disconnect(self, client, userdata, *args): self._connected = False - self._notify_status(False, "已断开连接") + self._status_notify.emit(False, "已断开连接") def _on_message(self, client, userdata, msg: mqtt.MQTTMessage): try: payload = json.loads(msg.payload.decode("utf-8")) except (json.JSONDecodeError, UnicodeDecodeError): payload = {"_raw": msg.payload.hex()} - self._notify_message(msg.topic, payload) + dev_serial = parse_topic_dev_serial(msg.topic) or "" + self._msg_notify.emit(msg.topic, dev_serial, payload) # ---- Public API ---- @@ -96,7 +93,6 @@ class MqttClient: client_id = self._config.client_id or f"dbn_mqtt_tool_{int(time.time())}" - # MQTT 5.0 用 CallbackAPIVersion.VERSION2 self._client = mqtt.Client( client_id=client_id, protocol=mqtt.MQTTv311, @@ -120,11 +116,30 @@ class MqttClient: self._connected = False def publish(self, topic: str, payload: dict, qos: int = 1): - """发送 JSON 消息到指定 topic""" + """线程安全的发布 — 通过 Qt 信号排队到主线程执行""" if not self._client or not self._connected: raise ConnectionError("MQTT 未连接") data = json.dumps(payload, ensure_ascii=False) - self._client.publish(topic, data, qos=qos) + self._publish_sig.emit(topic, data, qos) + + def subscribe(self, topic: str, qos: int = 1): + """订阅主题""" + if self._client: + self._client.subscribe(topic, qos=qos) + + def unsubscribe(self, topic: str): + """取消订阅""" + if self._client: + self._client.unsubscribe(topic) + + @Slot(str, str, int) + def _do_publish(self, topic: str, data: str, qos: int): + """实际的 paho publish 调用 — 始终在主线程执行""" + if self._client and self._connected: + try: + self._client.publish(topic, data, qos=qos) + except Exception as e: + print(f"[MqttClient] publish error: {e}") def send_command(self, dev_serial: str, cmd: str, data: Optional[dict] = None, qos: int = 1) -> int: """向设备发送命令，返回 msg_id""" diff --git a/DBNMQTTool/main.py b/DBNMQTTool/main.py index b6fd6ed..c2cee2c 100644 --- a/DBNMQTTool/main.py +++ b/DBNMQTTool/main.py @@ -39,8 +39,6 @@ from dbn_mqtt_tool.protocol import ( ) class MainWindow(QMainWindow): - _mqtt_status = Signal(bool, str) - _mqtt_msg = Signal(str, str, object) _devices_changed = Signal() def __init__(self): @@ -52,12 +50,10 @@ class MainWindow(QMainWindow): self._mqtt = MqttClient() self._devmgr = DeviceManager() - self._mqtt.on_status_change(lambda c, m: self._mqtt_status.emit(c, m)) - self._mqtt.on_message(lambda t, s, p: self._mqtt_msg.emit(t, s, p)) + self._mqtt._status_notify.connect(self._on_status) + self._mqtt._msg_notify.connect(self._on_message) self._devmgr.on_change(lambda: self._devices_changed.emit()) - self._mqtt_status.connect(self._on_status) - self._mqtt_msg.connect(self._on_message) self._devices_changed.connect(self._refresh_devices) self._build_ui() @@ -695,7 +691,7 @@ class MainWindow(QMainWindow): self._log(f"[自定义] 发布失败: {e}") def _custom_subscribe(self): - if not self._mqtt.connected or not self._mqtt._client: + if not self._mqtt.connected: QMessageBox.warning(self, "提示", "请先连接 Broker") return topic = self._custom_sub_topic.text() @@ -703,7 +699,7 @@ class MainWindow(QMainWindow): return try: qos = int(self._custom_sub_qos.currentText()) - self._mqtt._client.subscribe(topic, qos=qos) + self._mqtt.subscribe(topic, qos=qos) self._custom_subs[topic] = qos # 更新列表 found = False @@ -724,13 +720,11 @@ class MainWindow(QMainWindow): items = self._custom_sub_list.selectedItems() if not items: return - if not self._mqtt._client: - return for item in items: topic = item.text(0) try: if topic in self._custom_subs: - self._mqtt._client.unsubscribe(topic) + self._mqtt.unsubscribe(topic) del self._custom_subs[topic] idx = self._custom_sub_list.indexOfTopLevelItem(item) self._custom_sub_list.takeTopLevelItem(idx)