首先

npm install mqtt --save

一,.第一种方法 (相对比较容易看懂) 

使用场景跟MQTTX 类似的测试调试订阅接收信息的工具 (参数都是可配置的)

1.1 新建 mqtt.js

import * as mqtt from "mqtt/dist/mqtt.min";import { ElMessage } from 'element-plus'class MQTT { url = '';// mqtt地址 topic = ''; // clientId = ''; username = ''; password = '';//密码 qos = 0; // 初始化类实例 constructor(params) { this.topic = params.topic; this.url = params.url; // 虽然是mqtt但是在客户端这里必须采用websock的链接方式 this.clientId = params.clientId; this.username = params.username; this.password = params.password; this.qos = params.qos; } //初始化mqtt init() { const options = { // protocol: "ws", // host: this.url, // ws: 8083; wss: 8084 // port: 8083, // endpoint: "/mqtt", clean: true, connectTimeout: 4000, // 超时时间 username: this.username, password: this.password, clientId: this.clientId, clean: true, connectTimeout: 30 * 1000, // ms reconnectPeriod: 4000, // ms };//ws://localhost:8083/mqtt 这里组合起来是这种格式的地址 但是我传递过来的地址直接就是完整地址不用组装 所以我才注释上面 options 中的参数 const connectUrl = `${options.protocol}://${options.host}:${options.port}${options.endpoint}`;//并在这里直接使用地址链接 this.client = mqtt.connect(this.url, options); // 消息处理 this.client.on("message", (topic, message) => { // console.log("收到消息", topic, message); // console.log("收到消息" + topic + '发来的' + JSON.parse(message)); }) // 重连处理 this.client.on('reconnect', (error) => { console.log('正在重连:', error) }); // 链接失败 this.client.on('error', (error) => { console.log(error); }); } //取消订阅 unsubscribes() { this.client.unsubscribe(this.topic, (error) => { if (!error) { console.log('取消订阅成功'); } else { // console.log('取消订阅失败'); } }); } //连接 link() { this.client.on('connect', (con) => { let qosValue = this.qos this.client.subscribe(this.topic, { qosValue }, (error, res) => { if (!error) { console.log('订阅成功'); ElMessage({ message: '订阅成功', type: 'success', }) } else { ElMessage({ message: '订阅失败', type: 'error', }) // console.log('订阅失败'); } }); }); } // 发送信息 SendMessage(topic, sendMsg) { let options = this.qos this.client.publish('rscu/sensor/exterior/up/id', sendMsg, options, (err, a) => { if (!err) { console.log('发送信息成功'); ElMessage({ message: '发送信息成功', type: 'success', }) } else { console.log('发送信息失败'); } }) } //收到的消息 get(callback) { this.client.on('message', callback); } //结束链接 over() { this.client.end(); console.log('结束链接'); }}export default MQTT;

1.2 新建useMqtt.js (当时与对上面 mqtt.js的使用并二次封装)

import MQTT from './mqtt';import { onUnmounted, ref } from 'vue';import { ElMessage } from 'element-plus'export default function useMqtt() { const PublicMqtt = ref(null); const startMqtt = (val, callback) => { //设置订阅地址 PublicMqtt.value = new MQTT(val); //初始化mqtt PublicMqtt.value.init(); //链接mqtt PublicMqtt.value.link(); getMessage(callback); }; // 发送信息 监测有没有链接 没有弹框 const send = (topic, message) => { if (PublicMqtt.value) { let mqttPayload = JSON.parse(message); mqttPayload.dynamicType = ""; message = JSON.stringify(mqttPayload); PublicMqtt.value.SendMessage(topic, message); } else { ElMessage({ message: '尚未连接', type: 'error', }); } } const getMessage = (callback) => { PublicMqtt.value?.get(callback); }; // 断开链接 const endMqtt = () => { if (PublicMqtt.value) { PublicMqtt.value.unsubscribes(); PublicMqtt.value.over(); } } onUnmounted(() => { //页面销毁结束订阅 if (PublicMqtt.value) { PublicMqtt.value.unsubscribes(); PublicMqtt.value.over(); } }); return { startMqtt, send, endMqtt };}

1.3页面使用

import useMqtt from '../../../utils/useMqtt'const { startMqtt, send, endMqtt } = useMqtt();//链接 订阅 方法 在需要的地方调用 (参数可看第一张效果图上的参数)function ConcatMqttFn() {//校验输入信息 protocolForm.value.validate((valid) => { if (valid) { let params = { topic: protocolFormData.topic, //主题 url: protocolFormData.addressPath, //地址 clientId: protocolFormData.clientId, //clientId username: protocolFormData.account, //用户名 password: protocolFormData.password, //密码 qos: protocolFormData.qos, //qos } startMqtt(params, (topic, message) => {//因为我在封装js里面 callback 将他接收的信息返回回来了 所以我在这可以直接接收到 const msg = JSON.parse(message.toString()); requestData.value = msg }); } })}//最后在需要关闭链接 取消订阅的地方使用 endMqtt() 方法

二,第二种方法 (适用于不用配置 全局固定死链接地址和订阅主题) 就一个js文件

2.1 新建allMqtt.js

import * as mqtt from "mqtt/dist/mqtt.min";import { onUnmounted, ref, reactive } from 'vue';import { ElNotification } from 'element-plus'export default function useMqtt() { let client = ref({ connected: false }); const notifyPromise = ref(Promise.resolve()) const qosList = [0, 1, 2]; // 订阅主题 const topic = ref('rscu/sensor/warning/count') // 发送主题 const sendTopic = ref('rscu/sensor/warning') const qos = ref(1) // 链接地址 const hostUrl = ref('')//window.server.fileUploadUrl 这个是我在public文件下 static文件下 //创建的config.js 中定义的一个全局静态地址 并在 index.html中引用了他 他不会被打包 //你们也可以直接固定死 hostUrl.value = window.server.fileUploadUrl ? window.server.fileUploadUrl : '' const connection = reactive({ // 指明协议类型 protocol: "ws", host: hostUrl.value, // ws: 8083; wss: 8084 port: 8083, endpoint: "/mqtt", // for more options, please refer to https://github.com/mqttjs/MQTT.js#mqttclientstreambuilder-options clean: true, connectTimeout: 30 * 1000, // ms reconnectPeriod: 4000, // ms clientId: "emqx_benYing_" + Math.random().toString(16).substring(2, 8), // auth username: "warning", password: "root", }); const messageValue = ref(false) // 订阅的信息 const receiveNews = ref('') const time = ref(null) const startMqtt = (topic, callback) => { try { const { protocol, host, port, endpoint, ...options } = connection; const connectUrl = `${protocol}://${host}:${port}${endpoint}`; client.value = mqtt.connect(connectUrl, options); if (client.value.on) { // 连接 client.value.on("connect", () => { console.log("连接成功 successful"); link() }); // 重连 client.value.on("reconnect", handleOnReConnect); client.value.on("error", (error) => { // console.log("重连失败 error:", error); }); // 收到信息 callback返回收到的信息 client.value.on("message", callback); } } catch (error) { // console.log("mqtt.connect error:", error); } }; // 订阅 const link = () => { client.value.subscribe( topic.value, '1', (error, granted) => { if (error) { // console.log("订阅失败 error:", error); return; } else { sendMessage() // console.log("订阅成功 successfully:", granted); } } ); }; // 取消订阅 const UnSubscribe = () => { let qosValue = qos.value client.value.unsubscribe(topic.value, { qosValue }, (error) => { if (error) { // console.log("取消订阅失败 error:", error); return; } console.log(`取消订阅成功 topic: ${topic}`); }); }; // 取消连接 const destroyConnection = () => { if (client.value.connected) { try { client.value.end(false, () => { console.log("断开连接成功 successfully"); }); } catch (error) { // console.log("断开连接失败 error:", error); } } }; const retryTimes = ref(0); const handleOnReConnect = () => { retryTimes.value += 1; if (retryTimes.value > 5) { try { client.value.end(); initData(); // console.log("connection maxReconnectTimes limit, stop retry"); } catch (error) { // console.log("handleOnReConnect catch error:", error); } } }; const initData = () => { client.value = { connected: false, }; retryTimes.value = 0; };//发送信息 const sendMessage = () => { client.value.publish('rscu/sensor/warning', '1', '1', (err, a) => { if (!err) { } else { } }) }; return { startMqtt, link, UnSubscribe, destroyConnection, sendMessage };}

2.2使用

// 使用MQTTimport useMqtt from '../../utils/allMqtt.js'const { startMqtt, link, UnSubscribe, destroyConnection } = useMqtt();//html: <div class="message-prompt" :class="{ 'change': animateClass == true }"> <el-dropdown trigger="click" @command="messageHandleCommand"> <span style="cursor: pointer;"> <el-badge :value="all" :max="99" class="item"> <el-icon color="#fff" size="20"> <Bell /> </el-icon> </el-badge> </span> <template #dropdown> <el-dropdown-menu> <el-dropdown-item command="warning"> <span>预警</span> <span class="warning-text num">{ { alarm }}</span> </el-dropdown-item> <el-dropdown-item command="alarm"> <span>报警</span> <span class="alarm-text num">{ { warning }}</span> </el-dropdown-item> <el-dropdown-item command="all"> <span>查看全部</span> </el-dropdown-item> </el-dropdown-menu> </template> </el-dropdown> </div>//因为我在链接的时候 顺便调用了他的订阅方法 在js中 所以我在这直接链接function ConcatMqttFn() { startMqtt('', (topic, message) => {//拿到的数据 const msg = JSON.parse(message.toString()); // console.log(msg, 'msg'); alarm.value = msg.data.alarm all.value = msg.data.all warning.value = msg.data.warning });}//生命周期销毁的时候 取消 断开onMounted(() => { destroyConnection() ConcatMqttFn()});

2.3实现效果