1 <template>
  2   <view>
  3     <button :disabled="client.connected" type="primary" @click="onCreateConnect">连接</button>
  4     <button :disabled="!client.connected" @click="onCloseConnect">断开连接</button>
  5     <view style="margin: 10rpx 0">发送信息：</view>
  6     <view>
  7       <up-input placeholder="请输入内容" border="surround" v-model="pubMsg"></up-input>   <!-- up-input和up-button 是引入 uview-plus 组件库 -->
  8       <up-button type="primary" text="发送" @click="publish"></up-button>
  9     </view>
 10     <view style="margin: 10rpx 0;">
 11       接受的信息：
 12     </view>
 13     <view v-for="(item, index) in receivedMessages" :key="index">
 14       {{ item }}
 15     </view>
 16   </view>
 17 </template>
 18 
 19 <script setup>
 20   import { ref, reactive } from 'vue';
 21   import mqtt from 'mqtt/dist/mqtt.min' 
 22 
 23   const connection = reactive({
 24     //#region 链接协议说明
 25       // 连接字符串, 通过协议指定使用的连接方式
 26       // ws 未加密 WebSocket 连接
 27       // wss 加密 WebSocket 连接
 28       // mqtt 未加密 TCP 连接
 29       // mqtts 加密 TCP 连接
 30       // wxs 微信小程序连接
 31       // alis 支付宝小程序连接 
 32     //#endregion
 33     protocol: "wxs",    // 这个协议是针对微信小程序
 34     host: "broker.emqx.io",  // 
 35     port: 8084,  
 36     clientId: new Date().getTime(), 
 37     // username: "",  // 这里是账户名 网上免费的mqtt服务器没有账户密码，所以这里注释了
 38     // password: "",  // 这里是密码
 39     clean: true,
 40     connectTimeout: 30 * 1000, // ms
 41     reconnectPeriod: 4000, // ms 
 42   });
 43 
 44   const subTopic = ref("testtopic/miniprogram");    // 订阅主题
 45   const pubTopic = ref("testtopic/miniprogram");    // 发布主题
 46   const pubMsg = ref('');
 47   const client = ref({
 48     connected: false,
 49   });
 50   const receivedMessages = ref([])
 51 
 52   const initData = () => {
 53     client.value = {
 54       connected: false,
 55     }; 
 56   };
 57 
 58   /**
 59    * 创建连接
 60    */
 61   const onCreateConnect = () => {
 62     try {  
 63       const { protocol, host, port, ...options } = connection;
 64       const connectUrl = `${protocol}://${host}:${port}/mqtt`;  
 65       client.value = mqtt.connect(connectUrl, options);
 66       
 67       // 监听连接
 68       client.value.on("connect", () => { 
 69         console.log("connection successful");
 70         console.log("开启订阅消息:");
 71         subscribe();
 72       });
 73 
 74       // 监听错误
 75       client.value.on("error", (error) => {
 76         console.log("connection error:", error);
 77       });
 78 
 79       // 监听消息
 80       client.value.on("message", (topic, message) => {
 81         receivedMessages.value.push(message.toString())
 82         console.log(`received message: ${message} from topic: ${topic}`);
 83       }); 
 84     } catch (error) { 
 85       console.log("mqtt.connect error:", error);
 86     }
 87   }
 88 
 89   const subscribe = () => {
 90       if(client.value) {
 91           client.value.subscribe(subTopic.value, {qos: 2}, (err) => {
 92               if(!err) {
 93                 console.log(`成功订阅主题：${subTopic.value}`);
 94               }
 95           })
 96           return
 97       } 
 98   } 
 99 
100   /**
101    * 发送消息
102    */
103   const publish = () => {
104     if(client.value) {
105       client.value.publish(pubTopic.value, pubMsg.value, {qos: 2}, (err) => {
106         if(!err) {
107             console.log("发布信息成功",pubTopic.value, pubMsg.value)
108         }
109       })
110       return
111     } 
112 }
113 
114 
115   /**
116    * 断开连接
117    */
118   const onCloseConnect = () => {
119     if (client.value.connected) { 
120       try {
121         client.value.end(false, () => {
122           initData();
123           console.log("disconnected successfully");
124         });
125       } catch (error) { 
126         console.log("disconnect error:", error);
127       }
128     }
129   }
130 
131 </script>
132 

/**
 * author: wang.p 2025-07-09
 * description: 创建mqtt连接对象
 * 
 * 环境：node 20.18.3 + vue3
 * 使用的mqtt.js版本是4.1.0，
 * 
 * 
 * 参考文档： https://docs-im.easemob.com/mqtt/qsalisdk
 * 
 */
import PubSub from 'pubsub-js';
import mqtt from 'mqtt/dist/mqtt.min'

class MQTTClient {
   
  constructor() {
    if (!MQTTClient.client) {
      MQTTClient.client = this;
    }
    return MQTTClient.client;
  }
  
  qosList = [0, 1, 2]  // 消息级别：QoS 0 最多交付一次；QoS 1 至少交付一次；QoS 2 只交付一次；
  // 链接的对象
  client = {
    connected: false
  } 
  connecting = false;   // 是否链接中
  retryTimes = 0;  // 重连次数
  acceptMessages = [];   // 缓存接收的信息
  timeoutMessage = null;  // 轮询处理消息的定时器

  /**
   * 创建连接
   */
  createConnect () {
    try {
      this.connecting = true;

      const host = "192.168.1.231";  // IP
      const port = 8084;  // 端口 ws: 8083; wss: 8084
      const endpoint = "/mqtt";  // 
      const MQTT_CLIENTID = new Date().getTime();  // APPID 
      const MQTT_USERNAME = '你自己的mqtt账号';  // 账号
      const MQTT_PASSWORD = 'mqtt账号密码';  // 密码
      //#region 协议说明
      // 连接字符串, 通过协议指定使用的连接方式
      // ws 未加密 WebSocket 连接
      // wss 加密 WebSocket 连接
      // mqtt 未加密 TCP 连接
      // mqtts 加密 TCP 连接
      // wxs 微信小程序连接
      // alis 支付宝小程序连接 
      //#endregion
      let protocol = "wxs";  // 链接协议 
      //#ifdef MP-ALIPAY
      protocol = 'alis';
      //#endif 

      const options = {
        keepalive: 60, //默认为 60 秒 60s 定义了客户端发送心跳包（PING）的时间间隔（秒） 如果客户端在 keepalive 时间内没有发送消息，它会主动发送一个 PING 请求给服务器，服务器则返回 PING 响应以确认连接正常
        clientId: MQTT_CLIENTID,  // APPID
        username: MQTT_USERNAME,//账号
        password: MQTT_PASSWORD,//密码
        protocolVersion: 4, //MQTT连接协议版本
        clean: true,  //是否清除会话。为true时，断开连接后将清除会话，订阅过的Topics也将失效。为false时，离线状态下也能收到QoS为1和2的消息
        reconnectPeriod: 1000, //1000毫秒，两次重新连接之间的间隔, 设置为0禁用自动重新连接
        connectTimeout: 10 * 1000, //1000毫秒，两次重新连接之间的间隔,  这里的两次重新连接是指调用reconnect时重新链接需要的时间间隔
      };
 
      const connectUrl = `${protocol}://${host}:${port}${endpoint}`;
      this.client = mqtt.connect(connectUrl, options);
      if (this.client.on) {
        // 链接成功
        this.client.on("connect", () => {
          this.connecting = false;
          console.log("Connection succeeded!");
        });
        
        // 重连
        this.client.on("reconnect", this.onReConnect);

        // 错误
        this.client.on("error", (error) => {
          console.log("Connection failed", error);
        });

        // 接收消息
        this.client.on("message", (topic, payload) => { 
          if (payload) {
            let message = null;
            try {
              message = this.byteToString(payload);
            } catch (err) { 
              return;
            }
            if (!message) {
              return;
            }
            
            // 缓存接收的消息
            this.acceptMessages.push(message); 
          }

        });
      }

      // 开启轮询消息处理队列 
      this.startTimeout();
      
    } catch (error) {
      this.connecting = false;
      console.log("mqtt.connect error", error);
    }
  }

  /**
   * 字节序列转ASCII码
   * [0x24, 0x26, 0x28, 0x2A] ==> "$&C*"
   **/
  byteToString = (arr) => {
    if (typeof arr === 'string') {
      return arr;
    }
    let str = '',
      _arr = arr;
    for (let i = 0; i < _arr.length; i++) {
      let one = _arr[i].toString(2),
        v = one.match(/^1+?(?=0)/);
      if (v && one.length == 8) {
        let bytesLength = v[0].length;
        let store = _arr[i].toString(2).slice(7 - bytesLength);
        for (let st = 1; st < bytesLength; st++) {
          store += _arr[st + i].toString(2).slice(2);
        }
        str += String.fromCharCode(parseInt(store, 2));
        i += bytesLength - 1;
      } else {
        str += String.fromCharCode(_arr[i]);
      }
    }
    return str;
  }

  initData() {
    this.client = {
      connected: false,
    };
    this.retryTimes = 0;
    this.connecting = false; 
  }

  /**
   * 重连
   */
  onReConnect() {
    this.retryTimes += 1;
    if (this.retryTimes > 5) {
      try {
        this.client.end();
        this.initData(); 
      } catch (error) {
        console.log('reconnect error: ', error.toString());
      }
    }
  }

  /**
   * 订阅主题
   * @param {*} topic 字符串或者字符串数组
   */
  onSubscribe(topic, qos = 0) { 
    this.client.subscribe(topic, { qos }, (error, res) => {
      if (error) {
        console.log('Subscribe to topics error', error)
        return
      } 
      console.log('Subscribe to topics res', res)
    })
  }

  /**
   * 取消订阅
   */
  onUnSubscribe(topic) { 
    this.client.unsubscribe(topic, error => {
      if (error) {
        console.log('Unsubscribe error', error)
      }
    })
  }

  /**
   * 消息发布
   */
  onPublish(topic, payload, qos=0) { 
    this.client.publish(topic, payload, { qos }, error => {
      if (error) {
        console.log('Publish error', error)
      }
    })
  }

  /**
   * 断开连接
   */
  destroyConnection() {
    if (this.client.connected) {
      try {
        this.client.end(false, () => {
          this.initData()
          console.log('Successfully disconnected!')
        })
      } catch (error) {
        console.log('Disconnect failed', error.toString())
      }
    }
  }

    /**
   * 开启轮询处理消息队列
   * */
  startTimeout = () => {
    if (this.timeoutMessage === null) {
      this.timeoutMessage = setTimeout(this.pollMessage, 100)
    }
  }

  pollMessage = () => {
    if (this.acceptMessages.length > 0) {
       let messageData = this.acceptMessages.shift();
       if (messageData) { 
         this.disposeMessage(messageData);
         this.timeoutMessage = setTimeout(this.pollMessage, 50)
       }
    } else {
      this.timeoutMessage = setTimeout(this.pollMessage, 200)
    }
  }

    /**
   * 处理接受到的信息
   * */
  disposeMessage = (data) => {
    if (data) {   
      console.log('pubsub: ', data)

      PubSub.publish('testorder', data) //发布消息 
    }
  }

}

const MqttUtils = new MQTTClient();

export default MqttUtils;

import MqttUtils from "./mqttUtils";

  // 创建mqtt连接

  MqttUtils.createConnect();