MQTT.js 入门使用教程

简介

MQTT.js 是一款基于 JavaScript 开发的开源 MQTT 协议客户端库，主要应用于 Node.js 和浏览器环境，目前已成为 JavaScript 生态中使用最广泛的 MQTT 客户端解决方案。

MQTT 是一种基于发布/订阅模式的轻量级物联网消息传输协议，旨在以极少的代码和带宽消耗为联网设备提供实时可靠的消息服务，广泛应用于物联网、移动互联网、智能硬件、车联网及电力能源等行业。

鉴于 JavaScript 的单线程特性，MQTT.js 设计为全异步 MQTT 客户端，支持 MQTT/TCP、MQTT/TLS 以及 MQTT/WebSocket 协议。不同运行环境的支持情况如下：

• 浏览器环境：支持 MQTT over WebSocket（涵盖微信小程序、支付宝小程序等定制浏览器环境）
• Node.js 环境：支持 MQTT 和 MQTT over WebSocket

注意：除部分连接参数因环境不同而有所差异外，其他 API 保持一致。自 MQTT.js v3.0.0 版本起，已完整支持 MQTT 5.0 协议。

安装

使用 npm 或 yarn 安装

npm install mqtt --save

# 或使用 yarn

yarn add mqtt

注意：若 Node 环境版本为 v12 或 v14 及以上，请确保使用 MQTT.js 4.0.0 或更高版本。

使用 CDN 安装

在浏览器环境中，可通过 CDN 方式引入 MQTT.js。MQTT.js 的 bundle 包由 http://unpkg.com 管理，直接添加 unpkg.com/mqtt/dist/mqtt.min.js 即可使用。

<script src="https://unpkg.com/mqtt/dist/mqtt.min.js"></script>
<script>
  // 将在全局初始化一个 mqtt 变量
  console.log(mqtt)
</script>

全局安装

除上述安装方式外，MQTT.js 还提供全局安装方式，利用命令行工具完成 MQTT 的连接、发布和订阅等操作。

npm install mqtt -g

下文的使用教程中将详细描述如何使用 MQTT.js 的命令行工具。

使用

本文测试将使用 EMQX Cloud 提供的免费公共 MQTT 服务器作为 MQTT 服务器地址，服务器接入信息如下：

• Broker: broker.emqx.io
• TCP Port: 1883
• SSL/TLS Port: 8883

更多详情请访问 EMQX Cloud 官网，或查看 EMQX Cloud 文档。

简单例子

我们编写一段代码，实现连接到 EMQX Cloud 并完成订阅主题、收发消息的简单示例。由于浏览器环境仅支持 WebSocket 连接，我们将分别展示浏览器环境和 Node.js 环境下的不同连接参数。除连接地址外，其他参数相同，读者可根据实际情况选择。

const mqtt = require('mqtt')

/***
 * 浏览器环境
 * 使用协议为 ws 和 wss 的 MQTT over WebSocket 连接
 * EMQX 的 ws 连接默认端口为 8083，wss 为 8084
 * 注意需要在连接地址后加上一个 path, 例如 /mqtt
 */
const url = 'ws://broker.emqx.io:8083/mqtt'
/***
 * Node.js 环境
 * 使用协议为 mqtt 和 mqtts 的 MQTT over TCP 连接
 * EMQX 的 mqtt 连接默认端口为 1883，mqtts 为 8084
 */
// const url = 'mqtt://broker.emqx.io:1883'

// 创建客户端实例
const options = {
  // Clean session
  clean: true,
  connectTimeout: 4000,
  // 认证信息
  clientId: 'emqx_test',
  username: 'emqx_test',
  password: 'emqx_test',
}
const client = mqtt.connect(url, options)
client.on('connect', function () {
  console.log('Connected')
  // 订阅主题
  client.subscribe('test', function (err) {
    if (!err) {
      // 发布消息
      client.publish('test', 'Hello mqtt')
    }
  })
})

// 接收消息
client.on('message', function (topic, message) {
  // message is Buffer
  console.log(message.toString())
  client.end()
})

命令行

全局安装 MQTT.js 后，同样可以使用命令行工具完成主题订阅、消息发布和接收操作。

示例连接到 broker.emqx.io 并订阅 testtopic/# 主题：

mqtt sub -t 'testtopic/#' -h 'broker.emqx.io' -v

示例连接到 broker.emqx.io 并向 testtopic/hello 主题发送消息

mqtt pub -t 'testtopic/hello' -h 'broker.emqx.io' -m 'from MQTT.js'

如需使用功能更全面的 MQTT 命令行工具，可参考使用 MQTT X CLI。

API 介绍

mqtt.connect([url], options)

该函数用于连接到指定的 MQTT Broker，始终返回一个 Client 对象。第一个参数为 URL，可以是以下协议：mqtt, mqtts, tcp, tls, ws, wss。URL 也可以是 URL.parse() 返回的对象。随后传入 Options 对象以配置 MQTT 连接选项。使用 WebSocket 连接时，需注意地址后是否需要添加 path，例如 /mqtt。

以下列举常用的 Options 对象属性值：

• Options

  • keepalive: 单位为秒，数值类型，默认为 60 秒，设置为 0 时禁止
  • clientId: 默认为 'mqttjs_' + Math.random().toString(16).substr(2, 8)，可自定义修改的字符串
  • protocolVersion: MQTT 协议版本号，默认为 4（v3.1.1），可修改为 3（v3.1）和 5（v5.0）
  • clean: 默认为 true，是否清除会话。设置为 true 时，断开连接后将清除会话，订阅过的 Topics 失效。设置为 false 时，离线状态下也能收到 QoS 为 1 和 2 的消息
  • reconnectPeriod: 重连间隔时间，单位为毫秒，默认为 1000 毫秒。注意：设置为 0 后取消自动重连
  • connectTimeout: 连接超时时长，收到 CONNACK 前的等待时间，单位为毫秒，默认 30000 毫秒
  • username: 认证用户名，若 Broker 要求用户名认证，请设置该值
  • password: 认证密码，若 Broker 要求密码认证，请设置该值
  • will: 遗嘱消息，一个可配置的对象值，当客户端非正常断开连接时，Broker 就会向遗嘱 Topic 里面发布一条消息，格式为：
    • topic: 遗嘱发送的 Topic
    • payload: 遗嘱发布的消息
    • QoS: 遗嘱发送的 QoS 值
    • retain: 遗嘱发布的消息的 retain 标志
  • properties: MQTT 5.0 新增，可配置对象的属性值，详情请参考：https://github.com/mqttjs/MQTT.js#mqttclientstreambuilder-options

• 如需配置 SSL/TLS 连接，Option 对象将传递给 tls.connect()，因此可在 option 中配置以下属性

  • rejectUnauthorized: 是否验证服务端证书链和地址名称，设置为 false 时将跳过验证，存在中间人攻击风险，不建议在生产环境使用。设置为 true 时开启强认证模式，自签名证书需在证书配置时设置 Alt name。
  • ca: 仅在服务器使用自签名证书时必要，即自签名证书中生成的 CA 文件
  • cert: 仅在服务器需要客户证书认证（双向认证）时必要，即客户端证书
  • key: 仅在服务器需要客户证书认证（双向认证）时必要，即客户端密钥

Client 事件

连接成功后，返回的 Client 对象可通过 on 方法监听多个事件，业务逻辑可在监听的回调函数中完成。以下列举一些常用事件：

connect

连接成功时触发，参数为 connack

client.on('connect', function (connack) {
  console.log('Connected')
})

reconnect

断开连接后，经过重连间隔时间重新自动连接到 Broker 时触发

client.on('reconnect', function () {
  console.log('Reconnecting...')
})

close

断开连接以后触发

client.on('close', function () {
  console.log('Disconnected')
})

disconnect

收到 Broker 发送的断开连接报文时触发，参数 packet 为断开连接时接收到的报文，MQTT 5.0 中的功能

client.on('disconnect', function (packet) {
  console.log(packet)
})

offline

客户端下线时触发

client.on('offline', function () {
  console.log('offline')
})

error

客户端无法成功连接或发生解析错误时触发，参数 error 为错误信息

client.on('error', function (error) {
  console.log(error)
})

message

客户端收到一个发布过来的 Payload 时触发，包含三个参数：topic、payload 和 packet。其中 topic 为接收到的消息的 topic，payload 为接收到的消息内容，packet 为 MQTT 报文信息（包含 QoS、retain 等）。

client.on('message', function (topic, payload, packet) {
  // Payload is Buffer
  console.log(`Topic: ${topic}, Message: ${payload.toString()}, QoS: ${packet.qos}`)
})

Client 方法

Client 除监听事件外，还内置一些方法，用于进行发布订阅等操作。以下列举一些常用方法。

Client.publish(topic, message, [options], [callback])

向某一 topic 发布消息的函数方法，包含四个参数：

// 向 testtopic 主题发送一条 QoS 为 0 的测试消息
client.publish('testtopic', 'Hello, MQTT!', { qos: 0, retain: false }, function (error) {
  if (error) {
    console.log(error)
  } else {
    console.log('Published')
  }
})

• topic: 要发送的主题，为字符串
• message: 要发送的消息内容，可以是字符串或者是 Buffer
• options: 可选值，发布消息时的配置信息，主要设置 QoS、Retain 值等。
• callback: 发布消息后的回调函数，参数为 error，仅当发布失败时存在。

Client.subscribe(topic/topic array/topic object, [options], [callback])

订阅一个或者多个 topic 的方法。连接成功后需订阅主题以获取消息，该方法包含三个参数：

// 订阅一个名为 testtopic QoS 为 0 的 Topic
client.subscribe('testtopic', { qos: 0 }, function (error, granted) {
  if (error) {
    console.log(error)
  } else {
    console.log(`${granted[0].topic} was subscribed`)
  }
})

• topic: 可传入一个字符串，或者一个字符串数组，也可以是一个 topic 对象，{'test1': {qos: 0}, 'test2': {qos: 1}}
• options: 可选值，订阅 Topic 时的配置信息，主要填写订阅的 Topic 的 QoS 等级。
• callback: 订阅 Topic 后的回调函数，参数为 error 和 granted。订阅失败时 error 参数存在，granted 是一个 {topic, qos} 的数组，其中 topic 是被订阅的主题，qos 是 Topic 被授予的 QoS 等级。

Client.unsubscribe(topic/topic array, [options], [callback])

取消订阅单个主题或多个主题的方法，包含三个参数：

// 取消订阅名为 testtopic 的 Topic
client.unsubscribe('testtopic', function (error) {
  if (error) {
    console.log(error)
  } else {
    console.log('Unsubscribed')
  }
})

• topic: 可传入一个字符串或一个字符串数组
• options: 可选值，取消订阅时的配置信息
• callback: 取消订阅时的回调函数，参数为 error，仅当取消订阅失败时存在。

Client.end([force], [options], [callback])

关闭客户端的方法，包含三个参数：

client.end()

• force: 设置为 true 时将立即关闭客户端，无需等待断开连接的消息被接受。该参数可选，默认为 false。注意：使用该值为 true 时，Broker 无法接收到 disconnect 的报文。
• options: 可选值，关闭客户端时的配置信息，主要可配置 reasonCode（断开连接时的 Reason Code）。
• callback: 当客户端关闭时的回调函数

总结

至此简要介绍了 MQTT.js 一些常用 API 的使用方法。具体在实际项目中的使用请参考以下链接：

如何在 Vue 项目中使用 MQTT

如何在 React 项目中使用 MQTT

如何在 Electron 项目中使用 MQTT

如何在 Node.js 项目中使用 MQTT

使用 WebSocket 连接 MQTT 服务器

到此这篇关于MQTT.js 入门教程的文章就介绍到这了，更多相关MQTT.js 入门内容请搜索本站以前的文章或继续浏览下面的相关文章，希望大家以后多多支持本站！