mqtt架构修改

2026-01-18 19:11:45 +08:00 · 2026-01-18 19:11:45 +08:00 · 001ad6c071
parent 997cd9426b
commit 001ad6c071
6 changed files with 109 additions and 31 deletions
--- a/App.vue
+++ b/App.vue
@ -3,6 +3,8 @@ import config from './config'
 import { getToken } from '@/utils/auth'
 import mqttUtil from '@/utils/mqtt'
 import {startMqttOnlinePing, stopMqttOnlinePing} from "./utils/mqtt";
+import {batchSubscribe} from "./api/system/mqtt";
+import store from "store";

 export default {
  globalData: {
@ -107,8 +109,27 @@ export default {
        console.error('MQTT连接失败')
        return
      }
+      var clientId = mqttUtil.getMqttState().clientId;
+      const subscribeList = [
+        `frontend/${clientId}/dtu/864865085016294/listener`,
+        `frontend/${clientId}/dtu/864536071808560/listener`,
+        `frontend/${clientId}/dtu/864865085008135/listener`,
+        `frontend/${clientId}/dtu/862538065276939/listener`
+      ];
+      if (store.getters && store.getters.name === 'admin') {
+        subscribeList.push(`frontend/${clientId}/dtu/862538065276061/listener`);
+      }
+      this.globalData.mqtt.subscribeList = subscribeList || []
+
+      batchSubscribe({clientId: clientId}).then((result) => {
+        if (result.code === 200) {
+          console.info(`设备列表订阅成功：${subscribeList}`)
+        }
+      })
+
+      // ========== 新增：登录成功时同步到localStorage ==========
+      uni.setStorageSync('mqtt_subscribe_list', subscribeList || [])

-      const subscribeList = this.globalData.mqtt.subscribeList
      if (subscribeList.length > 0) {
        mqttUtil.updateSubscribeList(subscribeList)
        console.log('恢复MQTT订阅列表：', subscribeList)
@ -116,12 +137,10 @@ export default {
        uni.setStorageSync('mqtt_subscribe_list', subscribeList)
      }
    },
-    loginSuccess(token, subscribeList) {
+    loginSuccess(token) {
      this.globalData.mqtt.hasLogin = true
      this.globalData.mqtt.token = token
-      this.globalData.mqtt.subscribeList = subscribeList || []
-      // ========== 新增：登录成功时同步到localStorage ==========
-      uni.setStorageSync('mqtt_subscribe_list', subscribeList || [])
+
      this.reconnectMqtt()
      console.log('登录成功，MQTT已初始化')
    },
--- a/api/system/mqtt.js
+++ b/api/system/mqtt.js
@ -0,0 +1,49 @@
+import request from '@/utils/request'
+
+export function subscribe(query) {
+    return request({
+        url: '/api/mqtt/single',
+        method: 'post',
+        params: query
+    })
+}
+
+export function unsubscribe(query) {
+    return request({
+        url: '/api/mqtt/single',
+        method: 'delete',
+        params: query
+    })
+}
+
+
+export function batchUnsubscribe(query) {
+    return request({
+        url: '/api/mqtt/batchUnsubscribe',
+        method: 'delete',
+        params: query
+    })
+}
+
+export function batchSubscribe(query) {
+    return request({
+        url: '/api/mqtt/batchSubscribe',
+        method: 'post',
+        params: query
+    })
+}
+
+export function manualReconnect(query) {
+    return request({
+        url: '/api/mqtt/reconnect',
+        method: 'get',
+        params: query
+    })
+}
+
+export function status() {
+    return request({
+        url: '/api/mqtt/batch',
+        method: 'get'
+    })
+}
--- a/pages/control/index.vue
+++ b/pages/control/index.vue
@ -117,7 +117,7 @@ const SENSOR_MAP = {
  temp1: "201", temp2: "202", temp3: "203", temp4: "204",
  humi1: "101", humi2: "102", humi3: "103", humi4: "104"
 };
-const MQTT_TOPIC_SUFFIX = { UP: "/up", DOWN: "/down" };
+const MQTT_TOPIC_SUFFIX = { UP: "/listener", DOWN: "/control" };

 import UniDatetimePicker from "../../uni_modules/uni-datetime-picker/components/uni-datetime-picker/uni-datetime-picker.vue";
 import UniPopup from "../../uni_modules/uni-popup/components/uni-popup/uni-popup.vue"; // 引入弹窗组件
@ -141,7 +141,7 @@ export default {
    return {
      temp: "",
      mqttConfig: {
-        subscribeTopic:'/up',
+        subscribeTopic:'/listener',
      },
      value: 1,
      selectedText: '',
@ -151,7 +151,7 @@ export default {
      range: [],
      agriId:'',
      imei:'',
-      publishTopic: '/down',
+      publishTopic: '/control',
      title:'',
      message: {},
      // 优化：声明响应式变量 connected
@ -272,10 +272,11 @@ export default {
    },
    change(e) {
      this.imei = e;
+      var clientId = mqttUtil.getMqttState().clientId;
      if ((e === 'A' || e==='B' || e==='C')
          && store.getters && store.getters.name === 'admin' ) {
          this.getNewSpecialData(e);
-          this.mqttConfig.subscribeTopic = `dtu/862538063921866${MQTT_TOPIC_SUFFIX.UP}`;
+          this.mqttConfig.subscribeTopic = `frontend/${clientId}/dtu/862538063921866${MQTT_TOPIC_SUFFIX.UP}`;
      }
      const selectedItem = this.range.find(item => item.value === e);
      if (selectedItem) {
@ -284,8 +285,8 @@ export default {
        this.title= this.selectedText;
        if (e !== 'A' && e!=='B' && e!=='C') {
          // 优化：使用常量拼接MQTT主题
-          this.publishTopic = `dtu/${this.imei}${MQTT_TOPIC_SUFFIX.DOWN}`;
-          this.mqttConfig.subscribeTopic = `dtu/${this.imei}${MQTT_TOPIC_SUFFIX.UP}`;
+          this.publishTopic = `frontend/${clientId}${MQTT_TOPIC_SUFFIX.DOWN}/${this.imei}`;
+          this.mqttConfig.subscribeTopic = `frontend/${clientId}/dtu/${this.imei}${MQTT_TOPIC_SUFFIX.UP}`;
          var queryParams = {
            imei: this.imei
          }
@ -489,7 +490,7 @@ export default {

    // 消息回调逻辑完全保留（仅依赖全局工具类转发消息）
    ackMessage(topic, payload) {
-      // 1. 先判断是否是目标订阅主题（如dtu/xxx/up）
+      // 1. 先判断是否是目标订阅主题（如frontend/\\w+/control/\\w+"）
      if (topic !== this.mqttConfig.subscribeTopic) return;
      let msgData = {};
      // 优化：捕获JSON解析异常
--- a/pages/login.vue
+++ b/pages/login.vue
@ -52,7 +52,6 @@
  import { getCodeImg } from '@/api/login'
  import { getToken } from '@/utils/auth'
  import UniIcons from "../uni_modules/uni-icons/components/uni-icons/uni-icons.vue";
-  import store from "../store";

  export default {
    components: {UniIcons},
@ -145,13 +144,8 @@
          // 864865085016294 十方北棚
          // 864536071808560 七方北棚
          // 864865085008135 八方北棚
-          const subscribeList = [`dtu/864865085016294/up`, `dtu/864536071808560/up`,`dtu/864865085008135/up`,`dtu/862538065276939/up`]
-
-          if (store.getters && store.getters.name === 'admin') {
-            subscribeList.push(`dtu/862538065276061/up`);
-          }
          // 4. 调用App.vue的loginSuccess方法，初始化MQTT
-          app.loginSuccess(token, subscribeList)
+          app.loginSuccess(token)

          // ========== 原有逻辑保留 ==========
          this.$tab.reLaunch('/pages/control/index')
--- a/store/modules/user.js
+++ b/store/modules/user.js
@ -5,8 +5,10 @@ import { isHttp, isEmpty } from "@/utils/validate"
 import { login, logout, getInfo } from '@/api/login'
 import { getToken, setToken, removeToken } from '@/utils/auth'
 import defAva from '@/static/images/profile.jpg'
-
+import {batchUnsubscribe} from "@/api/system/mqtt";
 const baseUrl = config.baseUrl
+import mqttUtil from '@/utils/mqtt';
+

 const user = {
  state: {
@ -92,16 +94,21 @@ const user = {
    // 退出系统
    LogOut({ commit, state }) {
      return new Promise((resolve, reject) => {
-        logout(state.token).then(() => {
-          commit('SET_TOKEN', '')
-          commit('SET_ROLES', [])
-          commit('SET_PERMISSIONS', [])
-          removeToken()
-          storage.clean()
-          resolve()
-        }).catch(error => {
-          reject(error)
-        })
+          batchUnsubscribe({clientId:mqttUtil.getMqttState().clientId}).then(response => {
+              if (response.code === 200) {
+                  logout(state.token).then(() => {
+                      commit('SET_TOKEN', '')
+                      commit('SET_ROLES', [])
+                      commit('SET_PERMISSIONS', [])
+                      removeToken()
+                      storage.clean()
+                      resolve()
+                  }).catch(error => {
+                      reject(error)
+                  })
+                  console.info("取消订阅成功！")
+              }
+          })
      })
    }
  }
--- a/utils/mqtt.js
+++ b/utils/mqtt.js
@ -10,6 +10,8 @@

 // 微信小程序端：引入适配小程序的 mqtt 版本
 import mqtt from 'mqtt/dist/mqtt'
+import {batchUnsubscribe} from "../api/system/mqtt";
+import {getToken} from "./auth";

 // ===================== MQTT配置（暂时写死，TODO：后续从数据字典获取）=====================
 const MQTT_CONFIG = {
@ -240,7 +242,13 @@ export function disconnectMqtt() {
        resetMqttState()
        return true
    }
-
+    if (getToken()) {
+        batchUnsubscribe({clientId:mqttState.options.clientId}).then(response => {
+            if (response.code === 200) {
+                console.info("取消订阅成功！")
+            }
+        })
+    }
    try {
        mqttState.client.end(true) // 强制断开
        resetMqttState()