前端MQTT详细使用

首先

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:                                                                                                                                                                                                                                                                                                            预警                                {{ alarm }}                                                                                        报警                                {{ warning }}                                                                                        查看全部                                                                                                    //因为我在链接的时候 顺便调用了他的订阅方法 在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实现效果