mqtt.js使用

工作中的尝试和积累

安装mqtt.js

npm install --save mqtt@2.18.9  注意版本

封装类 mqtt_service.js

import mqtt from 'mqtt';
class MqttApi {
    client = null;
    constructor(connectUrl, topic) {
        const options = {
            keepalive: 40, // 默认60秒，设置0为禁用
            clean: true, // 设置为false以在脱机时接收QoS 1和2消息
            connectTimeout: 4000,
            clientId: 'mqttjs_' + Math.random().toString(16).substr(2, 8),
            username: "guest",
            password: 'guest',
            // protocolId: 'MQTT',
            // protocolVersion: 4,
            protocolId: 'MQIsdp', // 只支持MQTT 3.1(不符合3.1.1)的代理
            protocolVersion: 3,   // 版本
            reconnectPeriod: 1000, //设置多长时间进行重新连接 单位毫秒 两次重新连接之间的时间间隔。通过将设置为，禁用自动重新连接0
            connectTimeout: 10 * 1000, // 收到CONNACK之前等待的时间
        }
        this.client = mqtt.connect(connectUrl, options);
        this.client.on("connect", (connack) => {
            console.log('链接成功')
            if (topic) {
                this.client.subscribe(topic, (err) => {
                    if (!err) {
                        console.log('订阅成功')
                    }
                })
            }
        });
        this.client.on('reconnect', () => {
            console.log("正在重连.....");
        });
        this.client.on("close", function(error) {
            if(error){
              // console.log(error)
            }else{
              console.log("客户端已断开.....");
            }
            
        });
        this.client.on("offline", function() {
            
        });
        //当客户端无法连接或出现错误时触发回调
        this.client.on("error", (error) => {
            console.log("客户端出现错误....." + error);
        });
        this.client.on("packetsend", (packet) => {
            if (packet && packet.payload) {
                console.log("客户端已发出数据包....." + packet.payload);
            }
        });
    }
    // 断开链接
    disconnect() {
        try {
            if (this.client && this.client.connected) {
                this.client.end();
            }
        } catch (error) {
            console.log(error)
        }

    }
    // 重新链接
    reconnect() {
        this.client.reconnect()
    }
    // 关闭
    close() {
        try {
            if (this.client && this.client.connected) {
                this.client.end();
                this.client = null;
            }
        } catch (error) {
            console.log(error)
        }
    }
    // 发布消息
    publish(topic, message) {
        if (this.client && this.client.connected) {
            this.client.publish(topic, message, { qos: 0, retain: true }, (error) => {
                if (error) {
                    console.log(error)
                } else {
                    console.log('发布消息成功')
                }
            })
        }
    }
    // 订阅主题
    subscribe(topic) {
        if (this.client && this.client.connected) {
            this.client.subscribe(topic, { qos: 0 }, function(error, granted) {
                if (error) {
                    console.log(error)
                } else {
                    console.log(`${granted[0].topic} 订阅成功`)
                }
            })
        }
    }
    // 取消订阅
    unsubscribe(topic) {
        if (this.client && this.client.connected) {
            this.client.unsubscribe(topic, (error) => {
                if (error) {
                    console.log(error)
                } else {
                    console.log('取消订阅成功')
                }
            })
        }
    }
}

在页面中调用

import  MqttApi from 'mqtt_service.js'
const connectUrl = 'ws://192.168.1.22:8209/mqtt'  // 链接地址
const topic = ['DN2', 'DN3']    // 订阅的主题可以是字符串，多个为数组形式
const mqttServer = new MqttApi(connectUrl, topic);  // 初始化链接
// 获取消息的回调
mqttServer.client.on('message', (topic, message, packet) => {
    console.log('收到消息')
    // topic 主题
    // message 消息主体
    console.log(topic, message)
    console.log(message.toString())
})

// 断开的回调
mqttServer.client.on("close", function(error) {
  if(!error){
    console.log("断开操作");
  }
});
/*
其他方法
// 断开链接
mqttServer.disconnect();
// 重新链接
mqttServer.reconnect()
// 关闭链接
mqttServer.close();
// 订阅主题
mqttServer.subscribe('主题')