快捷導(dǎo)航

前端MQTT詳細(xì)使用的兩種方法

更新時(shí)間：2024年07月24日 08:28:39 作者：CV工程師!!

MQTT(消息隊(duì)列遙測傳輸協(xié)議),是一種基于發(fā)布/訂閱(publish/subscribe)模式的通訊協(xié)議,該協(xié)議構(gòu)建于TCP/IP協(xié)議上,這篇文章主要給大家介紹了關(guān)于前端MQTT詳細(xì)使用的兩種方法,需要的朋友可以參考下

首先

npm install mqtt --save

一,.第一種方法 (相對比較容易看懂)

使用場景跟MQTTX 類似的測試調(diào)試訂閱接收信息的工具 (參數(shù)都是可配置的)

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;
    // 初始化類實(shí)例
    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, // 超時(shí)時(shí)間
            username: this.username,
            password: this.password,
            clientId: this.clientId,
            clean: true,
            connectTimeout: 30 * 1000, // ms
            reconnectPeriod: 4000, // ms
        };
//ws://localhost:8083/mqtt  這里組合起來是這種格式的地址 但是我傳遞過來的地址直接就是完整地址不用組裝 所以我才注釋上面 options 中的參數(shù) 
        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 + '發(fā)來的' + 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('訂閱失敗');
                }
            });
        });
    }
    // 發(fā)送信息
    SendMessage(topic, sendMsg) {
        let options = this.qos
        this.client.publish('rscu/sensor/exterior/up/id', sendMsg, options, (err, a) => {
            if (!err) {
                console.log('發(fā)送信息成功');
                ElMessage({
                    message: '發(fā)送信息成功',
                    type: 'success',
                })
            } else {
                console.log('發(fā)送信息失敗');
            }
        })
    }
    //收到的消息
    get(callback) {
        this.client.on('message', callback);
    }
    //結(jié)束鏈接
    over() {
        this.client.end();
        console.log('結(jié)束鏈接');
    }
}
export default MQTT;

1.2 新建useMqtt.js (當(dāng)時(shí)與對上面 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) => {

        //設(shè)置訂閱地址
        PublicMqtt.value = new MQTT(val);
        //初始化mqtt
        PublicMqtt.value.init();
        //鏈接mqtt
        PublicMqtt.value.link();
        getMessage(callback);
    };
    // 發(fā)送信息 監(jiān)測有沒有鏈接 沒有彈框
    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(() => {
        //頁面銷毀結(jié)束訂閱
        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();

//鏈接 訂閱 方法  在需要的地方調(diào)用  (參數(shù)可看第一張效果圖上的參數(shù))
function ConcatMqttFn() {
//校驗(yàn)輸入信息
    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) => {
//因?yàn)槲以诜庋bjs里面 callback 將他接收的信息返回回來了 所以我在這可以直接接收到
                const msg = JSON.parse(message.toString());
                requestData.value = msg
            });
        }
    })
}

//最后在需要關(guān)閉鏈接 取消訂閱的地方使用   endMqtt() 方法

二,第二種方法 (適用于不用配置全局固定死鏈接地址和訂閱主題) 就一個(gè)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')
    // 發(fā)送主題
    const sendTopic = ref('rscu/sensor/warning')
    const qos = ref(1)
    // 鏈接地址
    const hostUrl = ref('')

//window.server.fileUploadUrl  這個(gè)是我在public文件下 static文件下 
//創(chuàng)建的config.js 中定義的一個(gè)全局靜態(tài)地址 并在 index.html中引用了他 他不會被打包 
//你們也可以直接固定死
    hostUrl.value = window.server.fileUploadUrl ? window.server.fileUploadUrl : ''
    const connection = reactive({
        // 指明協(xié)議類型
        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;
    };
//發(fā)送信息
    const sendMessage = () => {
        client.value.publish('rscu/sensor/warning', '1', '1', (err, a) => {
            if (!err) { } else {

            }
        })
    };

    return {
        startMqtt,
        link,
        UnSubscribe,
        destroyConnection,
        sendMessage
    };
}

2.2使用

// 使用MQTT
import 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>預(yù)警</span>
                                <span class="warning-text num">{{ alarm }}</span>
                            </el-dropdown-item>
                            <el-dropdown-item command="alarm">
                                <span>報(bào)警</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>

//因?yàn)槲以阪溄拥臅r(shí)候 順便調(diào)用了他的訂閱方法 在js中 所以我在這直接鏈接
function ConcatMqttFn() {
    startMqtt('', (topic, message) => {
//拿到的數(shù)據(jù)
        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
    });
}

//生命周期銷毀的時(shí)候 取消 斷開
onMounted(() => {
    destroyConnection()
    ConcatMqttFn()
});