Node

本文主要介绍如何在 Node.js 项目中使用 MQTT.js 客户端库 ，实现 SX-IOT设备与 MQTT 服务器的连接、订阅、收发消息等功能。

前提条件

安装依赖包

sudo apt install nodejs npm -y
npm install mqtt

连接使用

连接设置

本文将使用自定义的接入认证方式，服务器接入信息如下：

• Broker: mqtt.geek-smart.cn

• TCP Port: 1883

• WebSocket Port: 8083

导入依赖包

const mqtt = require('mqtt')
/**
  choose which protocol to use for connection here
   * TCP Port: 1883
   * WebSocket port: 8083
*/
// tcp
const { connectOptions } = require('./use_mqtt.js')
// ws
// const { connectOptions } = require('./use_ws.js')

定义连接地址、认证信息以及消息发布主题

// tcp
const connectOptions = {
  protocol: 'mqtt',
  port: 1883,
  host: 'mqtt.geek-smart.cn',
}

// ws
// const connectOptions = {
//   protocol: 'ws',
//   port: 8083,
//   host: 'mqtt.geek-smart.cn',
// }

module.exports = {
  connectOptions,
}
const pub_topic = '/HYUGHV/lVtAcHuorFDJ/4cebd60bfada/publish'
const sub_topic = '/HYUGHV/lVtAcHuorFDJ/4cebd60bfada/subscribe'
// tcp
const clientId = 'nodejs-mqtt-tcp-client'
// ws
// const clientId = 'nodejs-mqtt-ws-client'
const options = {
    clientId,
    clean: true,
    connectTimeout: 4000,
    username: 'OGVkSlBrskPY',
    password: 'XaIkqRXwcNuuQjXGPi',
    reconnectPeriod: 1000,
}

定义消息发布函数

client.publish(pub_topic, payload, { qos }, (error) => {
    if (error) {
        console.error(error)
    }
})

定义 on_message 回调函数，用于打印订阅主题接收的消息内容

client.on('message', (sub_topic, payload) => {
    console.log('Received Message:', sub_topic, payload.toString())
})

连接 MQTT

client.on('connect', () => {
    console.log(`${protocol}: Connected`)
})

完整代码

const mqtt = require('mqtt')
    /**
      choose which protocol to use for connection here
       * TCP Port: 1883
       * WebSocket port: 8083
    */
    // tcp
const { connectOptions } = require('./use_mqtt.js')
    // ws
    // const { connectOptions } = require('./use_ws.js')

const clientId = 'nodejs-mqtt-tcp-client'
    // const clientId = 'nodejs-mqtt-ws-client'
const options = {
    clientId,
    clean: true,
    connectTimeout: 4000,
    username: 'OGVkSlBrskPY',
    password: 'XaIkqRXwcNuuQjXGPi',
    reconnectPeriod: 1000,
}

const { protocol, host, port } = connectOptions
/**
 * if protocol is "mqtt", connectUrl = "mqtt://mqtt.geek-smart.cn:1883"
 * if protocol is "ws", connectUrl = "ws://mqtt.geek-smart.cn:8083/mqtt"
 *
 * for more details about "mqtt.connect" method & options,
 * please refer to https://github.com/mqttjs/MQTT.js#mqttconnecturl-options
 */
let connectUrl = `${protocol}://${host}:${port}`
if (['ws', 'wss'].includes(protocol)) {
    // mqtt: MQTT-WebSocket uniformly uses /path as the connection path,
    // which should be specified when connecting, and the path used on EMQX is /mqtt.
    connectUrl += '/mqtt'
}

const client = mqtt.connect(connectUrl, options)

const pub_topic = '/HYUGHV/lVtAcHuorFDJ/4cebd60bfada/publish'
const sub_topic = '/HYUGHV/lVtAcHuorFDJ/4cebd60bfada/subscribe'
const payload = '{"type":"info"}'
    // https://github.com/mqttjs/MQTT.js#qos
const qos = 0

// https://github.com/mqttjs/MQTT.js#event-connect
client.on('connect', () => {
    console.log(`${protocol}: Connected`)

    // subscribe topic
    // https://github.com/mqttjs/MQTT.js#mqttclientsubscribetopictopic-arraytopic-object-options-callback
    client.subscribe(sub_topic, { qos }, (error) => {
        if (error) {
            console.log('subscribe error:', error)
            return
        }
        console.log(`${protocol}: Subscribe to topic '${sub_topic}'`)
            // publish message
            // https://github.com/mqttjs/MQTT.js#mqttclientpublishtopic-message-options-callback
        client.publish(pub_topic, payload, { qos }, (error) => {
            if (error) {
                console.error(error)
            }
        })
    })
})

// https://github.com/mqttjs/MQTT.js#event-reconnect
client.on('reconnect', (error) => {
    console.log(`Reconnecting(${protocol}):`, error)
})

// https://github.com/mqttjs/MQTT.js#event-error
client.on('error', (error) => {
    console.log(`Cannot connect(${protocol}):`, error)
})

// https://github.com/mqttjs/MQTT.js#event-message
client.on('message', (sub_topic, payload) => {
    console.log('Received Message:', sub_topic, payload.toString())
})

/**
 * If you need to unsubscribe from a topic, you can use the following code.
 */
// // unsubscribe topic
// // https://github.com/mqttjs/MQTT.js#mqttclientunsubscribetopictopic-array-options-callback
// client.unsubscribe(sub_topic, { qos }, (error) => {
//   if (error) {
//     console.log('unsubscribe error:', error)
//     return
//   }
//   console.log(`unsubscribed topic: ${sub_topic}`)
// })

/**
 * If you need to disconnect, you can use the following code.
 */
// if (client.connected) {
//   try {
//     // disconnect
//     // https://github.com/mqttjs/MQTT.js#mqttclientendforce-options-callback
//     client.end(false, () => {
//       console.log('disconnected successfully')
//     })
//   } catch (error) {
//     console.log('disconnect error:', error)
//   }
// }

测试验证

运行

npm start