fix: MqttClient 回退信号方案,添加 publish debug 日志追踪阻塞点
- 移除 QObject 基类和 QueuedConnection 信号方案 - publish() 直接调用 self._client.publish() 并打印前后日志 - 添加 print(flush=True) 确保 Windows 终端实时输出 目的是确认阻塞发生在: [MqttClient] publish topic=... qos=1 ← 打印后消失? [MqttClient] publish done mid=... ← 这行能否出现?
This commit is contained in:
@@ -3,12 +3,12 @@ MQTT 客户端封装
|
|||||||
"""
|
"""
|
||||||
import json
|
import json
|
||||||
import time
|
import time
|
||||||
|
import sys
|
||||||
import threading
|
import threading
|
||||||
from typing import Callable, Optional
|
from typing import Callable, Optional
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
|
||||||
import paho.mqtt.client as mqtt
|
import paho.mqtt.client as mqtt
|
||||||
from PySide6.QtCore import QObject, Signal, Slot, Qt
|
|
||||||
|
|
||||||
from .protocol import parse_topic_dev_serial
|
from .protocol import parse_topic_dev_serial
|
||||||
|
|
||||||
@@ -23,64 +23,67 @@ class BrokerConfig:
|
|||||||
keepalive: int = 60
|
keepalive: int = 60
|
||||||
|
|
||||||
|
|
||||||
class MqttClient(QObject):
|
class MqttClient:
|
||||||
"""DLD960 IoT MQTT 客户端 — QObject 基类确保信号线程安全"""
|
"""DLD960 IoT MQTT 客户端"""
|
||||||
|
|
||||||
# 供外部监听
|
|
||||||
_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):
|
def __init__(self):
|
||||||
super().__init__()
|
|
||||||
self._client: Optional[mqtt.Client] = None
|
self._client: Optional[mqtt.Client] = None
|
||||||
self._config = BrokerConfig()
|
self._config = BrokerConfig()
|
||||||
self._connected = False
|
self._connected = False
|
||||||
self._lock = threading.Lock()
|
self._lock = threading.Lock()
|
||||||
|
self._message_callbacks: list[Callable[[str, str, dict], None]] = []
|
||||||
# 信号→槽: publish 通过排队连接确保在主线程的 paho context 执行
|
self._status_callbacks: list[Callable[[bool, str], None]] = []
|
||||||
self._publish_sig.connect(self._do_publish, Qt.QueuedConnection)
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def connected(self) -> bool:
|
def connected(self) -> bool:
|
||||||
return self._connected
|
return self._connected
|
||||||
|
|
||||||
# ---- 公开回调 (兼容旧接口) ----
|
|
||||||
|
|
||||||
def on_status_change(self, callback: Callable[[bool, str], None]):
|
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]):
|
def on_message(self, callback: Callable[[str, str, dict], None]):
|
||||||
"""回调参数: (topic, dev_serial, payload_dict)"""
|
"""回调参数: (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):
|
def _on_connect(self, client, userdata, flags, reason_code, properties=None):
|
||||||
rc = reason_code if isinstance(reason_code, int) else reason_code.value
|
rc = reason_code if isinstance(reason_code, int) else reason_code.value
|
||||||
if rc == 0:
|
if rc == 0:
|
||||||
self._connected = True
|
self._connected = True
|
||||||
self._status_notify.emit(True, "已连接到 Broker")
|
self._notify_status(True, "已连接到 Broker")
|
||||||
|
|
||||||
from .protocol import TOPIC_ALL_DEVICE_UP
|
from .protocol import TOPIC_ALL_DEVICE_UP
|
||||||
client.subscribe(TOPIC_ALL_DEVICE_UP, qos=1)
|
client.subscribe(TOPIC_ALL_DEVICE_UP, qos=1)
|
||||||
else:
|
else:
|
||||||
self._connected = False
|
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):
|
def _on_disconnect(self, client, userdata, *args):
|
||||||
self._connected = False
|
self._connected = False
|
||||||
self._status_notify.emit(False, "已断开连接")
|
self._notify_status(False, "已断开连接")
|
||||||
|
|
||||||
def _on_message(self, client, userdata, msg: mqtt.MQTTMessage):
|
def _on_message(self, client, userdata, msg: mqtt.MQTTMessage):
|
||||||
try:
|
try:
|
||||||
payload = json.loads(msg.payload.decode("utf-8"))
|
payload = json.loads(msg.payload.decode("utf-8"))
|
||||||
except (json.JSONDecodeError, UnicodeDecodeError):
|
except (json.JSONDecodeError, UnicodeDecodeError):
|
||||||
payload = {"_raw": msg.payload.hex()}
|
payload = {"_raw": msg.payload.hex()}
|
||||||
dev_serial = parse_topic_dev_serial(msg.topic) or ""
|
self._notify_message(msg.topic, payload)
|
||||||
self._msg_notify.emit(msg.topic, dev_serial, payload)
|
|
||||||
|
|
||||||
# ---- Public API ----
|
# ---- Public API ----
|
||||||
|
|
||||||
@@ -116,11 +119,17 @@ class MqttClient(QObject):
|
|||||||
self._connected = False
|
self._connected = False
|
||||||
|
|
||||||
def publish(self, topic: str, payload: dict, qos: int = 1):
|
def publish(self, topic: str, payload: dict, qos: int = 1):
|
||||||
"""线程安全的发布 — 通过 Qt 信号排队到主线程执行"""
|
"""发送 JSON 消息到指定 topic"""
|
||||||
if not self._client or not self._connected:
|
if not self._client or not self._connected:
|
||||||
raise ConnectionError("MQTT 未连接")
|
raise ConnectionError("MQTT 未连接")
|
||||||
data = json.dumps(payload, ensure_ascii=False)
|
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):
|
def subscribe(self, topic: str, qos: int = 1):
|
||||||
"""订阅主题"""
|
"""订阅主题"""
|
||||||
@@ -132,15 +141,6 @@ class MqttClient(QObject):
|
|||||||
if self._client:
|
if self._client:
|
||||||
self._client.unsubscribe(topic)
|
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:
|
def send_command(self, dev_serial: str, cmd: str, data: Optional[dict] = None, qos: int = 1) -> int:
|
||||||
"""向设备发送命令,返回 msg_id"""
|
"""向设备发送命令,返回 msg_id"""
|
||||||
from .protocol import build_request, get_topic_for_cmd
|
from .protocol import build_request, get_topic_for_cmd
|
||||||
|
|||||||
+6
-2
@@ -39,6 +39,8 @@ from dbn_mqtt_tool.protocol import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
class MainWindow(QMainWindow):
|
class MainWindow(QMainWindow):
|
||||||
|
_mqtt_status = Signal(bool, str)
|
||||||
|
_mqtt_msg = Signal(str, str, object)
|
||||||
_devices_changed = Signal()
|
_devices_changed = Signal()
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
@@ -50,10 +52,12 @@ class MainWindow(QMainWindow):
|
|||||||
self._mqtt = MqttClient()
|
self._mqtt = MqttClient()
|
||||||
self._devmgr = DeviceManager()
|
self._devmgr = DeviceManager()
|
||||||
|
|
||||||
self._mqtt._status_notify.connect(self._on_status)
|
self._mqtt.on_status_change(lambda c, m: self._mqtt_status.emit(c, m))
|
||||||
self._mqtt._msg_notify.connect(self._on_message)
|
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._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._devices_changed.connect(self._refresh_devices)
|
||||||
|
|
||||||
self._build_ui()
|
self._build_ui()
|
||||||
|
|||||||
Reference in New Issue
Block a user