MQTT：Vue集成MQTT

一、Vue安装MQTT

npm install mqtt@5.14.0 --save 

二、基本使用

在Vue中实现MQTT的基本使用，包括：连接客户端、关闭客户端、订阅主题、取消订阅、发布消息、接收发布消息等。

<template>
  <div class="app-container" id="MqttDiv">
    <el-row :gutter="24" style="margin-bottom: 10px">
      <el-col :span="12">
        <el-card shadow="never">
          <div slot="header" class="clearfix">
            <span>配置信息</span>
          </div>
          <el-form id="mqttSettingForm" ref="mqttSettingForm" :rules="settingRules" :model="mqttSettingParams" label-width="90px" size="small">
            <el-row>
              <el-col :span="12">
                <el-form-item label="MQTT协议" prop="protocol">
                  <el-select v-model="mqttSettingParams.protocol" placeholder="MQTT协议" filterable clearable style="width: 100%">
                    <el-option v-for="item in protocolOperation" :key="item.value" :label="item.label" :value="item.value"></el-option>
                  </el-select>
                </el-form-item>
              </el-col>
              <el-col :span="12">
                <el-form-item label="主机地址" prop="host">
                  <el-input v-model="mqttSettingParams.host" placeholder="主机地址" clearable/>
                </el-form-item>
              </el-col>
            </el-row>
            <el-row>
              <el-col :span="12">
                <el-form-item label="端口号" prop="port">
                  <el-input-number v-model="mqttSettingParams.port" controls-position="right" placeholder="端口号" style="width: 100%"/>
                </el-form-item>
              </el-col>
              <el-col :span="12">
                <el-form-item label="客户端ID" prop="clientId">
                  <el-input v-model="mqttSettingParams.clientId" placeholder="客户端ID" clearable disabled/>
                </el-form-item>
              </el-col>
            </el-row>
            <el-row>
              <el-col :span="12">
                <el-form-item label="用户名" prop="username">
                  <el-input v-model="mqttSettingParams.username" placeholder="用户名" clearable/>
                </el-form-item>
              </el-col>
              <el-col :span="12">
                <el-form-item label="密码" prop="password">
                  <el-input v-model="mqttSettingParams.password" placeholder="密码" show-password clearable/>
                </el-form-item>
              </el-col>
            </el-row>
            <div style="text-align:center">
              <el-button type="primary" size="small" @click="createConnection" :disabled="clientDisable">建立连接</el-button>
              <el-button type="danger" size="small" @click="closeConnection" :disabled="!clientDisable">断开连接</el-button>
            </div>
          </el-form>
        </el-card>
      </el-col>
      <el-col :span="12">
        <el-card shadow="never">
          <div slot="header" class="clearfix">
            <span>订阅主题</span>
          </div>
          <el-form id="subForm" ref="subForm" :rules="subRules" :model="subParams" label-width="80px" size="small">
            <el-form-item label="订阅主题" prop="topic">
              <el-input v-model="subParams.topic" placeholder="订阅主题" :disabled="subDisable"/>
            </el-form-item>
            <el-form-item label="Qos" prop="qos">
              <el-select v-model="subParams.qos" placeholder="Qos" filterable clearable style="width: 100%" :disabled="subDisable">
                <el-option v-for="item in qosOperation" :key="item.value" :label="item.label" :value="item.value">
                  <span style="float: left">{{ item.label }}</span>
                  <span style="float: right; color: #8492a6; font-size: 13px">{{ item.detail }}</span>
                </el-option>
              </el-select>
            </el-form-item>
            <div style="text-align:center">
              <el-button type="primary" size="small" @click="subTopic" :disabled="subDisable">订阅主题</el-button>
              <el-button type="danger" size="small" @click="closeSub" :disabled="!subDisable">取消订阅</el-button>
            </div>
          </el-form>
        </el-card>
      </el-col>
    </el-row>
    <el-row :gutter="24">
      <el-col :span="12">
        <el-card shadow="never">
          <div slot="header" class="clearfix">
            <span>发布消息</span>
          </div>
          <el-form id="pubForm" ref="pubForm" :rules="pubRules" :model="pubParams" label-width="80px" size="small">
            <el-form-item label="订阅主题" prop="topic">
              <el-input v-model="pubParams.topic" placeholder="订阅主题"/>
            </el-form-item>
            <el-form-item label="Payload" prop="payload">
              <el-input type="textarea" :rows="2" v-model="pubParams.payload" placeholder="Payload"/>
            </el-form-item>
            <el-form-item label="Qos" prop="qos">
              <el-select v-model="pubParams.qos" placeholder="Qos" filterable clearable style="width: 100%">
                <el-option v-for="item in qosOperation" :key="item.value" :label="item.label" :value="item.value"></el-option>
              </el-select>
            </el-form-item>
            <div style="text-align: center">
              <el-button type="primary" size="small" @click="pubMessage">发布消息</el-button>
            </div>
          </el-form>
        </el-card>
      </el-col>
      <el-col :span="12">
        <el-card shadow="never">
          <el-input type="textarea" :rows="6" placeholder="接收消息" v-model="receiveMsg">
          </el-input>
        </el-card>
      </el-col>
    </el-row>
  </div>
</template>
<script>
// 在vue中如果想跟MQTT的服务端建立连接，只能使用websocket
import MqttUtils from "@/utils/MqttUtils"

export default {
  name: 'MqttView',
  data() {
    return {
      client: null,  // MQTT客户端
      clientDisable: false,  // MQTT客户端连接
      mqttSettingParams: {
        protocol: "ws",  // 协议
        host: "xxxx",
        port: "8083",
        endpoint: "/mqtt",
        clientId: "emqx_vue_client" + Math.random().toString().substring(2,8),
        username: "admin",
        password: "admin",
        keepalive: 30, // 心跳间隔(秒)
        connectTimeout: 4000, // 连接超时时间
        reconnectPeriod: 4000, // 重连间隔(毫秒)
      },  // 配置信息
      settingRules:{
        protocol: [{ required: true, message: 'MQTT协议不能为空', trigger: 'blur' }],
        host: [{ required: true, message: '主机地址不能为空', trigger: 'blur' }],
        port: [{ required: true, message: '端口号不能为空', trigger: 'blur' }],
      },

      subDisable: false,  // 主题订阅
      subParams: {
        topic: "vue/a",
        qos: 0
      },  // 订阅信息
      subRules: {
        topic: [{ required: true, message: '订阅主题不能为空', trigger: 'blur' }],
        qos: [{ required: true, message: 'Qos不能为空', trigger: 'blur' }],
      },

      pubParams: {
        topic: "vue/a",
        payload: null,
        qos: 0
      },  // 发布信息
      pubRules: {
        topic: [{ required: true, message: '订阅主题不能为空', trigger: 'blur' }],
        qos: [{ required: true, message: 'Qos不能为空', trigger: 'blur' }],
      },

      receiveMsg: null,  // 接收消息
      protocolOperation: [{label: "ws://", value: "ws"}, {label: "wss://", value: "wss"}],  // MQTT协议字典
      qosOperation: [{label: "Qos 0", value: 0, detail: "最多一次"}, {label: "Qos 1", value: 1, detail: "至少一次"}, {label: "Qos 2", value: 2, detail: "仅一次"}]
    }
  },
  mounted() {
  },
  methods: {
    /**
     * 创建连接
     */
    createConnection(){
      this.$refs["mqttSettingForm"].validate((valid) => {
        if (valid) {
          this.client = MqttUtils.createConnection(this.mqttSettingParams)
          if (!this.client) {
            this.clientDisable = false;
            this.$message.error("客户端连接失败");
            return;
          }
          this.clientDisable = true;
          this.$message.success("客户端连接成功");
        }
      });
    },
    /**
     * 断开连接
     * 第一个参数：Boolean类型 false表示不会立马断开连接，给MQTT发送报文之后在关闭；true立马关闭连接，不会发送报文
     * 第二个参数：连接关闭之后的回调函数
     */
    closeConnection(){
      const closeCoonection = MqttUtils.closeConnection(this.client)
      if (closeCoonection) {
        this.clientDisable = false;
        this.$message.success("客户端连接断开");
      } else {
        this.$message.error("断开连接失败");
      }
    },
    /**
     * 订阅主题
     */
    subTopic(){
      this.$refs["subForm"].validate(async (valid) => {
        if (valid) {
          const subTopic = await MqttUtils.subTopic(this.client, this.subParams);
          if (!subTopic) {
            this.$message.error("主题订阅失败");
            return;
          }
          this.subDisable = true
          this.$message.success("主题订阅成功");

          // 给连接对象注册一个接收消息的事件
          this.client.on("message", (topic, message) => {
            this.receiveMsg = topic + "--->" + message;
          });
        }
      });
    },
    /**
     * 取消订阅
     */
    async closeSub(){
      const closeSub = await MqttUtils.closeSub(this.client, this.subParams);
      if (!closeSub) {
        this.$message.error("取消订阅失败");
        return;
      }
      this.subDisable = false;
      this.$message.success("取消订阅成功");
    },
    /**
     * 发布消息
     */
    pubMessage(){
      this.$refs["pubForm"].validate(async (valid) => {
        if (valid) {
          const pubMessage = await MqttUtils.pubMessage(this.client, this.pubParams);
          if (!pubMessage) {
            this.$message.error("消息发送失败");
            return;
          }
          this.$message.success("消息发送成功");
        }
      });

    }
  }
}
</script>
<style scoped>
#MqttDiv {
  width: 100%;
  height: calc(100vh - 84px);
}

</style>

import mqtt from 'mqtt'

/**
 * 连接客户端
 * @param data
 *  protocol: "ws",  // 协议
 *  host: "xxx",
 *  port: "8083",
 *  endpoint: "/mqtt",
 *  clientId: "emqx_vue_client" + Math.random().toString().substring(2,8),
 *  username: "admin",
 *  password: "admin",
 *  keepalive: 30, // 心跳间隔(秒)
 *  connectTimeout: 4000, // 连接超时时间
 *  reconnectPeriod: 4000, // 重连间隔(毫秒)
 * @returns {MqttClient|null}
 */
function createConnection(data) {
  try {
    const {protocol, host, port, endpoint, ...options} = data;
    const connectionUrl = `${protocol}://${host}:${port}${endpoint}`;
    return mqtt.connect(connectionUrl, options);
  } catch (err) {
    console.error(err);
    return null;
  }
}

/**
 * 断开客户端连接
 * @param client
 */
function closeConnection(client) {
  try {
    client.end(false, () => {
      console.log("客户端连接断开")
    });
    return true;
  } catch (err) {
    return false;
  }
}

/**
 * 订阅主题
 * @param client
 * @param data
 * @returns {Promise<unknown>}
 */
function subTopic(client, data) {
  return new Promise((resolve) => {
    try {
      const {topic, qos} = data;
      // 注意：Qos必须是数字
      client.subscribe(topic, {qos}, (error) => {
        if (error) {
          console.error("主题订阅失败", error);
          resolve(false);
        }
        resolve(true);
      })
    } catch (error) {
      console.error("主题订阅失败", error);
      resolve(false);
    }
  })
}

/**
 * 取消订阅
 * @param client
 * @param data
 * @returns {Promise<unknown>}
 */
function closeSub(client, data) {
  return new Promise((resolve) => {
    try {
      const {topic, qos} = data;
      // 注意：Qos必须是数字
      client.unsubscribe(topic, {qos}, (error) => {
        if (error) {
          console.error("取消订阅失败", error);
          resolve(false);
        }
        resolve(true);
      })
    } catch (error) {
      console.error("取消订阅失败", error);
      resolve(false);
    }
  })
}

/**
 * 消息发布
 * @param client
 * @param data
 * @returns {Promise<unknown>}
 */
function pubMessage(client, data) {
  return new Promise((resolve) => {
    try {
      const {topic, qos, payload} = data;
      // 注意：Qos必须是数字
      client.publish(topic, payload, {qos}, (error) => {
        if (error) {
          console.error("消息发布失败", error);
          resolve(false);
        }
        resolve(true);
      })
    } catch (error) {
      console.error("消息发布失败", error);
      resolve(false);
    }
  })
}

const MqttUtils = {createConnection, closeConnection, subTopic, closeSub, pubMessage}
export default MqttUtils;