代码收藏家技术教程 2025-06-21

MQTT与Node.js设备对接实践：构建高效物联网通信系统实战指南

MQTT + Node.js 设备对接实战：构建高效的物联网通信系统

前言

在物联网（IoT）应用中，设备之间的通信至关重要，MQTT（Message Queuing Telemetry Transport） 作为轻量级的 消息协议，被广泛用于 设备数据采集、远程控制、消息推送 等场景。

本篇文章将介绍如何使用 Node.js + MQTT.js 实现设备对接，涵盖：

MQTT 协议解析
MQTT 服务器（EMQX/Mosquitto）搭建
Node.js 设备端 MQTT 连接
设备数据上报（Publish）
远程设备控制（Subscribe）
QoS 级别优化
最佳实践与安全策略

通过实战代码，你将掌握如何使用 Node.js 轻松对接 MQTT 设备，实现高效、稳定、可扩展的物联网通信方案。🚀

一、MQTT 协议概述

1.1 MQTT 核心特点

轻量级：消息头仅 2 字节

发布-订阅模式（Pub/Sub）：设备可同时订阅多个主题

QoS 级别：

QoS 0：最多一次（At most once）

QoS 1：至少一次（At least once）

QoS 2：仅一次（Exactly once）

支持遗嘱消息（Last Will），检测设备离线

二、MQTT 服务器部署（EMQX / Mosquitto）

2.1 使用 Docker 安装 EMQX（推荐）

docker run -d --name emqx -p 1883:1883 -p 8083:8083 emqx/emqx

默认端口

1883（MQTT 连接）

8083（WebSocket 连接）

2.2 使用 Docker 安装 Mosquitto

docker run -d --name mosquitto -p 1883:1883 eclipse-mosquitto

2.3 测试 MQTT 服务器

# 订阅主题
mosquitto_sub -h localhost -t "iot/device/data"

# 发布消息
mosquitto_pub -h localhost -t "iot/device/data" -m "Hello MQTT"

三、Node.js 设备对接 MQTT

3.1 安装 MQTT.js

npm install mqtt

3.2 设备端 MQTT 连接

const mqtt = require('mqtt');

// 连接 MQTT 服务器
const client = mqtt.connect("mqtt://localhost:1883");

client.on("connect", () => {
    console.log("设备已连接到 MQTT 服务器");
    client.subscribe("iot/device/command"); // 订阅远程控制主题
});

// 监听消息
client.on("message", (topic, message) => {
    console.log(`收到消息 [${topic}]: ${message.toString()}`);
});

// 处理设备断线重连
client.on("error", (err) => {
    console.error("MQTT 连接错误:", err);
});

效果：

设备连接 MQTT 服务器

订阅 iot/device/command 主题，等待远程指令

四、设备数据上报（Publish）

4.1 定期上报传感器数据

setInterval(() => {
    const payload = JSON.stringify({
        deviceId: "sensor-001",
        temperature: (20 + Math.random() * 10).toFixed(2),
        humidity: (40 + Math.random() * 20).toFixed(2),
        timestamp: Date.now()
    });

    client.publish("iot/device/data", payload, { qos: 1 });
    console.log("发送数据:", payload);
}, 5000);

效果：

设备每 5 秒 发送一次温湿度数据

服务器收到数据后，可存入数据库或做实时分析

五、远程设备控制（Subscribe）

5.1 设备监听远程指令

client.on("message", (topic, message) => {
    console.log(`收到控制指令: ${message.toString()}`);

    if (message.toString() === "TURN_ON") {
        console.log("设备已开启");
    } else if (message.toString() === "TURN_OFF") {
        console.log("设备已关闭");
    }
});

5.2 远程控制设备

const controlClient = mqtt.connect("mqtt://localhost:1883");

controlClient.on("connect", () => {
    console.log("控制端已连接 MQTT");
    controlClient.publish("iot/device/command", "TURN_ON");
});

效果：

控制端发送 "TURN_ON" 指令

设备端收到 "TURN_ON" 后执行相应操作

六、QoS 级别优化

QoS	说明	适用场景
`0`	最多一次（At most once）	低优先级数据，如日志
`1`	至少一次（At least once）	重要数据，如设备状态
`2`	仅一次（Exactly once）	关键数据，如交易记录

6.1 设置 QoS

设备端

client.publish("iot/device/data", payload, { qos: 1 });

远程控制端

controlClient.publish("iot/device/command", "TURN_OFF", { qos: 2 });

测试：

QoS 1 保障消息至少送达一次

QoS 2 确保消息仅送达一次

七、最佳实践与安全策略

7.1 设备定期发送心跳

setInterval(() => {
    client.publish("iot/device/heartbeat", JSON.stringify({ deviceId: "sensor-001" }), { qos: 1 });
}, 30 * 1000);

作用：

监测设备在线状态

设备掉线时，MQTT 服务器可触发告警

7.2 遗嘱消息（LWT）

const client = mqtt.connect("mqtt://localhost:1883", {
    will: {
        topic: "iot/device/status",
        payload: JSON.stringify({ deviceId: "sensor-001", status: "offline" }),
        qos: 1,
        retain: false
    }
});

作用：

设备异常掉线后，MQTT 服务器通知其他客户端

7.3 启用 TLS 连接（提高安全性）

const client = mqtt.connect("mqtts://broker.emqx.io:8883", {
    username: "myuser",
    password: "mypassword"
});

作用：

防止 MQTT 数据被中间人攻击（MITM）

适用于生产环境，防止未经授权的访问

八、总结

功能	实现方式
设备连接 MQTT	`mqtt.connect()`
数据上报（Publish）	`client.publish()`
远程控制（Subscribe）	`client.on("message")`
QoS 级别	`publish({ qos: 1 })`
心跳检测	`iot/device/heartbeat`
遗嘱消息（LWT）	`will: { topic: "status" }`
TLS 加密	`mqtts://broker.emqx.io`