修复mqtt客户端问题

pages/6170/deviceControl/index.vue

@@ -918,10 +918,6 @@
 					...this.Progress,
 					show: false, // 隐藏进度条
 				};
-				uni.showToast({
-					title: '网络异常',
-					icon: 'none'
-				});
 			}
 
 		},
@@ -945,44 +941,48 @@
 					// 订阅来自设备的状态更新
 					const statusTopic = `A/${this.itemInfo.deviceImei}`;
 					this.mqttClient.subscribe(statusTopic, (payload) => {
-						console.log(`收到来自 ${statusTopic} 的消息:`, payload);
-						//收到电量上报。延迟20s请求接口数据
-						const parsedMessage = typeof payload === 'string' ? JSON.parse(payload) :
-							payload;
-						const deviceState = parsedMessage.state; // 直接取 state 数组
-						// ✅ 发送全局事件通知主页面更新
-						uni.$emit('deviceStatusUpdate', {
-							message: parsedMessage, // 消息内容
-							timestamp: new Date().getTime() // 时间戳
-						});
-						if (deviceState[0] === 12) {
-							setTimeout(() => {
-								this.fetchDeviceDetail(data.data.id);
-							}, 20000);
-							// 这里判断电量低于20%,弹框提示
-							if (this.deviceInfo.batteryPercentage < 20) {
-								this.popupType = 'bettery'
-								this.popupMessage = '请及时充电';
-								this.showPopupFlag = true;
+						try {
+							console.log(`收到来自 ${statusTopic} 的消息:`, payload);
+							//收到电量上报。延迟20s请求接口数据
+							const parsedMessage = typeof payload === 'string' ? JSON.parse(payload) :
+								payload;
+							const deviceState = parsedMessage.state; // 直接取 state 数组
+							// ✅ 发送全局事件通知主页面更新
+							uni.$emit('deviceStatusUpdate', {
+								message: parsedMessage, // 消息内容
+								timestamp: new Date().getTime() // 时间戳
+							});
+							if (deviceState[0] === 12) {
+								setTimeout(() => {
+									this.fetchDeviceDetail(data.data.id);
+								}, 20000);
+								// 这里判断电量低于20%,弹框提示
+								if (this.deviceInfo.batteryPercentage < 20) {
+									this.popupType = 'bettery'
+									this.popupMessage = '请及时充电';
+									this.showPopupFlag = true;
+								}
 							}
-						}
-						// 处理上传照片进度消息
-						if (deviceState[0] === 3) {
-							const progress = deviceState[1];
-							// 更新进度条
-							this.Progress = {
-								...this.Progress,
-								curr: progress * 2,
-								show: progress < 50 // 进度达到100时自动隐藏
-							};
-							// 当进度为100时显示成功弹框
-							if (progress === 50) {
-								this.popupType = 'logo';
-								this.popupMessage = '上传成功';
-								this.showPopupFlag = true;
-								this.lightModeB = false; // 关闭上传弹窗
-								this.selectedImage = ''; // 清空已选图片
+							// 处理上传照片进度消息
+							if (deviceState[0] === 3) {
+								const progress = deviceState[1];
+								// 更新进度条
+								this.Progress = {
+									...this.Progress,
+									curr: progress * 2,
+									show: progress < 50 // 进度达到100时自动隐藏
+								};
+								// 当进度为100时显示成功弹框
+								if (progress === 50) {
+									this.popupType = 'logo';
+									this.popupMessage = '上传成功';
+									this.showPopupFlag = true;
+									this.lightModeB = false; // 关闭上传弹窗
+									this.selectedImage = ''; // 清空已选图片
+								}
 							}
+						} catch (error) {
+							console.error('解析MQTT消息失败:', error, '原始消息:', payload);
 						}
 					});
 				})

utils/mqtt.js

@@ -133,94 +133,146 @@ class MqttClient {
 			password: config.MQTT_PASSWORD,
 		};
 		this.onConnectCallback = null;
-		this.messageCallbacks = new Map(); // 使用 Map 存储不同主题的回调
+		this.messageCallbacks = new Map();
+		this.isReconnecting = false;
+		this.reconnectAttempts = 0;
+		this.maxReconnectAttempts = 10;
+		this.reconnectTimeoutId = null;
+		this.manualDisconnect = false;
+
+		this.client = new Paho.Client(
+			this.options.host,
+			Number(this.options.port),
+			"/mqtt",
+			this.options.clientId
+		);
+		this.setCallbacks();
+	}
+
+	setCallbacks() {
+		this.client.onConnectionLost = (responseObject) => {
+			if (responseObject.errorCode !== 0) {
+				console.log("MQTT连接丢失: " + responseObject.errorMessage);
+				uni.$emit('mqttConnectionLost', {
+					error: responseObject.errorMessage,
+					timestamp: Date.now()
+				});
+				if (!this.manualDisconnect) {
+					this.startReconnecting();
+				}
+			}
+		};
+
+		this.client.onMessageArrived = (message) => {
+			const topic = message.destinationName;
+			const payload = message.payloadString;
+			console.log(`收到消息, 主题: ${topic}, 内容: ${payload}`);
+			const potentialJsons = payload.replace(/}\s*{/g, '}|{').split('|');
+			potentialJsons.forEach(jsonString => {
+				if (jsonString.trim() === '') return;
+				if (this.messageCallbacks.has(topic)) {
+					this.messageCallbacks.get(topic)(jsonString);
+				}
+			});
+		};
 	}
 
-	/**
-	 * 连接到 MQTT Broker
-	 * @param {Function} onConnectCallback - 连接成功后的回调函数
-	 */
 	connect(onConnectCallback) {
-		const brokerUrl = this.options.host;
-		const brokerPort = this.options.port;
-		console.log(`正在连接MQTT: ${brokerUrl}:${brokerPort}/mqtt`);
+		if (this.client && this.client.isConnected()) {
+			console.log('MQTT客户端已连接。');
+			if (onConnectCallback) onConnectCallback();
+			return;
+		}
+
+		this.manualDisconnect = false;
+		this.onConnectCallback = onConnectCallback;
+
+		console.log(`正在连接MQTT: ${this.options.host}:${this.options.port}/mqtt`);
 
 		try {
-			// Paho-MQTT 的客户端需要 host, port, path, clientId
-			this.client = new Paho.Client(brokerUrl, Number(brokerPort), "/mqtt", this.options.clientId);
-
-			this.onConnectCallback = onConnectCallback;
-
-			// 设置回调
-			this.client.onConnectionLost = (responseObject) => {
-				if (responseObject.errorCode !== 0) {
-					console.log("MQTT连接丢失: " + responseObject.errorMessage);
-				  // 发送全局事件
-				    uni.$emit('mqttConnectionLost', {
-				      error: responseObject.errorMessage,
-				      timestamp: Date.now()
-				    });
-				}
-			};
-
-			this.client.onMessageArrived = (message) => {
-				const topic = message.destinationName;
-				const payload = message.payloadString;
-				console.log(`收到消息, 主题: ${topic}, 内容: ${payload}`);
-				if (this.messageCallbacks.has(topic)) {
-					// 调用该主题对应的回调
-					this.messageCallbacks.get(topic)(payload);
-				}
-			};
-
 			const connectOptions = {
 				timeout: 4,
 				userName: this.options.username,
 				password: this.options.password,
-				useSSL: false, // 如果是 wss, 需要设置为 true
+				useSSL: false,
 				cleanSession: true,
 				onSuccess: () => {
 					console.log('MQTT连接成功');
+					this.reconnectAttempts = 0;
+					this.isReconnecting = false;
+					clearTimeout(this.reconnectTimeoutId);
+					this.resubscribe();
 					if (this.onConnectCallback) {
 						this.onConnectCallback();
 					}
 				},
 				onFailure: (message) => {
 					console.error('MQTT连接失败:', message.errorMessage);
-					this.disconnect();
+					if (!this.manualDisconnect) {
+						this.startReconnecting();
+					}
 				}
 			};
 
 			this.client.connect(connectOptions);
-
 		} catch (error) {
-			console.error('MQTT客户端初始化失败:', error);
+			console.error('MQTT客户端连接时发生异常:', error);
+			if (!this.manualDisconnect) {
+				this.startReconnecting();
+			}
+		}
+	}
+
+	startReconnecting() {
+		if (this.isReconnecting || this.manualDisconnect) {
+			return;
+		}
+
+		if (this.reconnectAttempts >= this.maxReconnectAttempts) {
+			console.error('MQTT: 达到最大重连次数，已放弃。');
+			this.isReconnecting = false;
+			this.reconnectAttempts = 0;
+			return;
+		}
+
+		this.isReconnecting = true;
+		this.reconnectAttempts++;
+
+		const delay = 1000 * Math.pow(2, this.reconnectAttempts - 1);
+		console.log(`MQTT: 第 ${this.reconnectAttempts} 次重连将在 ${delay / 1000}s 后开始...`);
+
+		this.reconnectTimeoutId = setTimeout(() => {
+			if (!this.manualDisconnect) {
+				this.connect(this.onConnectCallback);
+			}
+		}, delay);
+	}
+
+	resubscribe() {
+		if (!this.client || !this.client.isConnected()) {
+			return;
+		}
+		console.log('MQTT: 重新订阅所有主题...');
+		for (const [topic] of this.messageCallbacks) {
+			console.log(`重新订阅: ${topic}`);
+			this.client.subscribe(topic, {
+				qos: 1
+			});
 		}
 	}
 
-	/**
-	 * 订阅主题
-	 * @param {String} topic - 需要订阅的主题
-	 * @param {Function} onMessageCallback - 收到该主题消息后的回调函数
-	 */
 	subscribe(topic, onMessageCallback) {
+		this.messageCallbacks.set(topic, onMessageCallback);
 		if (this.client && this.client.isConnected()) {
 			console.log(`尝试订阅主题: ${topic}`);
 			this.client.subscribe(topic, {
 				qos: 1
 			});
-			// 存储该主题的回调函数
-			this.messageCallbacks.set(topic, onMessageCallback);
 		} else {
-			console.error('MQTT未连接，无法订阅');
+			console.error('MQTT未连接，订阅请求已缓存，连接后将自动订阅。');
 		}
 	}
 
-	/**
-	 * 发布消息
-	 * @param {String} topic - 需要发布的主题
-	 * @param {String} message - 需要发布的消息
-	 */
 	publish(topic, message) {
 		if (this.client && this.client.isConnected()) {
 			const mqttMessage = new Paho.Message(message);
@@ -233,31 +285,26 @@ class MqttClient {
 		}
 	}
 
-	/**
-	 * 取消订阅
-	 * @param {String} topic - 需要取消订阅的主题
-	 */
 	unsubscribe(topic) {
+		if (this.messageCallbacks.has(topic)) {
+			this.messageCallbacks.delete(topic);
+		}
 		if (this.client && this.client.isConnected()) {
 			this.client.unsubscribe(topic);
-			// 从回调Map中移除
-			if (this.messageCallbacks.has(topic)) {
-				this.messageCallbacks.delete(topic);
-			}
 			console.log(`成功取消订阅: ${topic}`);
 		} else {
-			console.error('MQTT未连接，无法取消订阅');
+			console.error('MQTT未连接或已断开，无法取消订阅');
 		}
 	}
 
-	/**
-	 * 断开连接
-	 */
 	disconnect() {
+		this.manualDisconnect = true;
+		clearTimeout(this.reconnectTimeoutId);
 		if (this.client && this.client.isConnected()) {
 			this.client.disconnect();
 		}
-		this.client = null;
+		this.isReconnecting = false;
+		this.reconnectAttempts = 0;
 		console.log('MQTT连接已断开');
 	}
 }