From c98aa5b0bb882e45982065c961b2d288e5e6d9c2 Mon Sep 17 00:00:00 2001 From: wangfq Date: Tue, 7 Jul 2026 17:27:15 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20MqttClient=20=E5=9B=9E=E9=80=80=E4=BF=A1?= =?UTF-8?q?=E5=8F=B7=E6=96=B9=E6=A1=88=EF=BC=8C=E6=B7=BB=E5=8A=A0=20publis?= =?UTF-8?q?h=20debug=20=E6=97=A5=E5=BF=97=E8=BF=BD=E8=B8=AA=E9=98=BB?= =?UTF-8?q?=E5=A1=9E=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 移除 QObject 基类和 QueuedConnection 信号方案 - publish() 直接调用 self._client.publish() 并打印前后日志 - 添加 print(flush=True) 确保 Windows 终端实时输出 目的是确认阻塞发生在: [MqttClient] publish topic=... qos=1 ← 打印后消失？ [MqttClient] publish done mid=... ← 这行能否出现？ --- DBNMQTTool/dbn_mqtt_tool/mqtt_client.py | 70 ++++++++++++------------- DBNMQTTool/main.py | 8 ++- 2 files changed, 41 insertions(+), 37 deletions(-) diff --git a/DBNMQTTool/dbn_mqtt_tool/mqtt_client.py b/DBNMQTTool/dbn_mqtt_tool/mqtt_client.py index 47ed59f..33fc4dc 100644 --- a/DBNMQTTool/dbn_mqtt_tool/mqtt_client.py +++ b/DBNMQTTool/dbn_mqtt_tool/mqtt_client.py @@ -3,12 +3,12 @@ MQTT 客户端封装 """ import json import time +import sys import threading 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,64 +23,67 @@ class BrokerConfig: keepalive: int = 60 -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 +class MqttClient: + """DLD960 IoT MQTT 客户端""" def __init__(self): - super().__init__() self._client: Optional[mqtt.Client] = None self._config = BrokerConfig() self._connected = False self._lock = threading.Lock() - - # 信号→槽: publish 通过排队连接确保在主线程的 paho context 执行 - self._publish_sig.connect(self._do_publish, Qt.QueuedConnection) + self._message_callbacks: list[Callable[[str, str, dict], None]] = [] + self._status_callbacks: list[Callable[[bool, str], None]] = [] @property def connected(self) -> bool: return self._connected - # ---- 公开回调 (兼容旧接口) ---- - def on_status_change(self, callback: Callable[[bool, str], None]): - self._status_notify.connect(callback) + self._status_callbacks.append(callback) def on_message(self, callback: Callable[[str, str, dict], None]): """回调参数: (topic, dev_serial, payload_dict)""" - self._msg_notify.connect(callback) + self._message_callbacks.append(callback) - # ---- MQTT callbacks (paho 网络线程) ---- + 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 ---- 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._status_notify.emit(True, "已连接到 Broker") + self._notify_status(True, "已连接到 Broker") from .protocol import TOPIC_ALL_DEVICE_UP client.subscribe(TOPIC_ALL_DEVICE_UP, qos=1) else: self._connected = False - self._status_notify.emit(False, f"连接失败 (rc={rc})") + self._notify_status(False, f"连接失败 (rc={rc})") def _on_disconnect(self, client, userdata, *args): self._connected = False - self._status_notify.emit(False, "已断开连接") + self._notify_status(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()} - dev_serial = parse_topic_dev_serial(msg.topic) or "" - self._msg_notify.emit(msg.topic, dev_serial, payload) + self._notify_message(msg.topic, payload) # ---- Public API ---- @@ -116,11 +119,17 @@ class MqttClient(QObject): self._connected = False def publish(self, topic: str, payload: dict, qos: int = 1): - """线程安全的发布 — 通过 Qt 信号排队到主线程执行""" + """发送 JSON 消息到指定 topic""" if not self._client or not self._connected: raise ConnectionError("MQTT 未连接") data = json.dumps(payload, ensure_ascii=False) - self._publish_sig.emit(topic, data, qos) + print(f"[MqttClient] publish topic={topic} qos={qos}", flush=True) + try: + info = self._client.publish(topic, data, qos=qos) + print(f"[MqttClient] publish done mid={info.mid}", flush=True) + except Exception as e: + print(f"[MqttClient] publish EXCEPTION: {e}", flush=True) + raise def subscribe(self, topic: str, qos: int = 1): """订阅主题""" @@ -132,15 +141,6 @@ class MqttClient(QObject): 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""" from .protocol import build_request, get_topic_for_cmd diff --git a/DBNMQTTool/main.py b/DBNMQTTool/main.py index c2cee2c..0e786e5 100644 --- a/DBNMQTTool/main.py +++ b/DBNMQTTool/main.py @@ -39,6 +39,8 @@ 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): @@ -50,10 +52,12 @@ class MainWindow(QMainWindow): self._mqtt = MqttClient() self._devmgr = DeviceManager() - self._mqtt._status_notify.connect(self._on_status) - self._mqtt._msg_notify.connect(self._on_message) + 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._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()