1

api/middleware/mqtt/mqttClient.js

@@ -2,13 +2,16 @@ const mqtt = require('mqtt')
 const { v4: uuidv4 } = require('uuid'); // 顶部添加
 const Framework = require('../../../framework/node-core-framework');
 const logs = require('../logProxy');
+
 // 获取logsService
 class MqttSyncClient {
     constructor(brokerUrl, options = {}) {
         this.client = mqtt.connect(brokerUrl, options)
         this.isConnected = false
 
-        this.messageListeners = []
+        // 使用 Map 结构优化消息监听器，按 topic 分组
+        this.messageListeners = new Map(); // Map<topic, Set<listener>>
+        this.globalListeners = new Set(); // 全局监听器（监听所有 topic）
 
         this.client.on('connect', () => {
             this.isConnected = true
@@ -18,11 +21,42 @@ class MqttSyncClient {
 
         this.client.on('message', (topic, message) => {
 
-            message = JSON.parse(message.toString())
+            let messageObj;
+            try {
+                messageObj = JSON.parse(message.toString());
+            } catch (error) {
+                console.warn('[MQTT] 消息解析失败:', error.message);
+                return;
+            }
 
-            console.log('MQTT 收到消息', topic, message)
+            // 记录日志但不包含敏感信息
+            const { maskSensitiveData } = require('../../utils/crypto_utils');
+            const safeMessage = maskSensitiveData(messageObj, ['password', 'pwd', 'token', 'secret', 'key', 'cookie']);
+            console.log('[MQTT] 收到消息', topic, '类型:', messageObj.action || messageObj.type || 'unknown');
 
-            this.messageListeners.forEach(listener => listener(topic, message))
+            // 优化：只通知相关 topic 的监听器，而不是所有监听器
+            // 1. 触发该 topic 的专用监听器
+            const topicListeners = this.messageListeners.get(topic);
+            if (topicListeners && topicListeners.size > 0) {
+                topicListeners.forEach(listener => {
+                    try {
+                        listener(topic, messageObj);
+                    } catch (error) {
+                        console.error('[MQTT] Topic监听器执行失败:', error.message);
+                    }
+                });
+            }
+
+            // 2. 触发全局监听器
+            if (this.globalListeners.size > 0) {
+                this.globalListeners.forEach(listener => {
+                    try {
+                        listener(topic, messageObj);
+                    } catch (error) {
+                        console.error('[MQTT] 全局监听器执行失败:', error.message);
+                    }
+                });
+            }
 
         })
 
@@ -139,12 +173,56 @@ class MqttSyncClient {
         });
     }
 
-    addMessageListener(fn) {
-        this.messageListeners.push(fn)
+    /**
+     * 添加消息监听器
+     * @param {Function} fn - 监听器函数
+     * @param {string} topic - 可选，指定监听的 topic，不指定则监听所有
+     */
+    addMessageListener(fn, topic = null) {
+        if (typeof fn !== 'function') {
+            throw new Error('监听器必须是函数');
+        }
+
+        if (topic) {
+            // 添加到特定 topic 的监听器
+            if (!this.messageListeners.has(topic)) {
+                this.messageListeners.set(topic, new Set());
+            }
+            this.messageListeners.get(topic).add(fn);
+        } else {
+            // 添加到全局监听器
+            this.globalListeners.add(fn);
+        }
     }
 
-    removeMessageListener(fn) {
-        this.messageListeners = this.messageListeners.filter(f => f !== fn)
+    /**
+     * 移除消息监听器
+     * @param {Function} fn - 监听器函数
+     * @param {string} topic - 可选，指定从哪个 topic 移除
+     */
+    removeMessageListener(fn, topic = null) {
+        if (topic) {
+            // 从特定 topic 移除
+            const topicListeners = this.messageListeners.get(topic);
+            if (topicListeners) {
+                topicListeners.delete(fn);
+                // 如果该 topic 没有监听器了，删除整个 Set
+                if (topicListeners.size === 0) {
+                    this.messageListeners.delete(topic);
+                }
+            }
+        } else {
+            // 从全局监听器移除
+            this.globalListeners.delete(fn);
+
+            // 也尝试从所有 topic 中移除（兼容旧代码）
+            for (const [topicKey, listeners] of this.messageListeners.entries()) {
+                listeners.delete(fn);
+                if (listeners.size === 0) {
+                    this.messageListeners.delete(topicKey);
+                }
+            }
+        }
     }
 
     /**