MQTT 协议详解与 Node.js 实践

阿赛工作室

236人浏览 · 2026-05-06 09:54:05

阿赛工作室 · 2026-05-06 09:54:05 发布

MQTT（Message Queuing Telemetry Transport）是一种基于发布/订阅模式的轻量级物联网通信协议，它工作在 TCP/IP 协议族之上。其设计目标是低带宽、低功耗、不稳定的网络环境，因此非常适用于传感器、移动设备和嵌入式系统。

一、MQTT 协议核心概念

1. 核心角色

发布者（Publisher）：负责生产消息，并将消息发送到特定的主题（Topic）。
订阅者（Subscriber）：向代理订阅自己感兴趣的主题，一旦该主题有消息到达，代理就会将消息推送给该订阅者。
代理（Broker）：消息中介，负责接收所有消息、管理主题订阅关系，并将消息正确转发给匹配的订阅者。所有客户端只与 Broker 通信，完全解耦。

2. 主题（Topic）

主题是 UTF-8 字符串，采用分层结构，用 / 分隔，如 home/bedroom/temperature。

主题支持两种通配符，用于订阅时匹配多个层级：

单层通配符 +：匹配单层，如 home/+/temperature 可匹配 home/bedroom/temperature 和 home/livingroom/temperature，但不能匹配 home/bedroom/humidity。
多层通配符 #：匹配当前层及所有子层，必须放在主题末尾，如 home/bedroom/# 可匹配 home/bedroom/temperature、home/bedroom/light、home/bedroom/light/status 等所有子主题。

特别注意：

主题区分大小写。
不建议以 / 开头，如 /home/temp，虽然合法，但会产生一个无意义的空层级。
避免在主题中使用特殊字符和空格。

3. 服务质量（QoS）

MQTT 定义了三种消息传递的可靠性等级，确保在不可靠网络下的数据完整性：

QoS	名称	描述	适用场景
0	最多一次	消息只发送一次，不保证送达，接收方不发送确认。	环境传感器持续上报，偶尔丢包影响不大。
1	至少一次	保证消息至少送达一次，可能重复，发送方会不断重试直到收到 PUBACK。	重要控制指令，可容忍重复，但必须送达。
2	只有一次	使用四次握手确保消息精确到达且只到达一次。开销最大，延迟最高。	计费、支付等严格不允许重复的场合。

4. 保留消息（Retained Message）

当发布者发送消息时，可设置 Retain 标志为 true，那么 Broker 会为该主题存储最后一条保留消息。后续任何新的订阅者订阅该主题时，会立即收到这条保留消息，即获取最新状态。

常见用途：设备上线就能马上知道某传感器的当前值，而无需等待下一次上报。
若要清除保留消息，可向相同主题发送一条空消息体且 Retain=true。

5. 遗嘱消息（Last Will and Testament, LWT）

客户端可以在连接时设置遗嘱消息。如果该客户端非正常断开（如心跳超时），Broker 会自动将遗嘱消息发布到指定主题，通知其他设备该客户端已离线。

6. 保活心跳（Keep Alive）

客户端连接时指定以秒为单位的心跳间隔。如果在 1.5 倍心跳时间内没有收到任何 MQTT 报文，Broker 就判定连接断开。

二、Node.js + MQTT.js 原生方案示例

我们将使用 mqtt 这个 npm 包（即 MQTT.js）来实现一个完整的示例：包含一个 Broker、一个发布者模拟传感器、一个订阅者接收数据。为了便于快速运行，我们使用公共测试 Broker，也可以选择在本地启动 Mosquitto。

环境准备

确保已安装 Node.js (版本 > 12)。
新建项目目录并初始化：

mkdir mqtt-demo && cd mqtt-demo
npm init -y
npm install mqtt

选择一个 Broker

你可根据情况选择：

公共 Broker：mqtt://test.mosquitto.org（无需认证，端口 1883），适合测试。
本地 Mosquitto：若想完全控制，可在本地安装 Mosquitto 并启动（mosquitto -v）。
自建 Broker：若不依赖外部，也可使用 aedes 等 Broker 库，但为了简洁，本例采用公共 Broker。

⚠️ 注意：公共 Broker 任何人都可以收到发布的消息，不要传输敏感数据。

完整代码

我们将创建两个文件：publisher.js 和 subscriber.js，分别代表传感器数据发布者和控制中心订阅者。

📡 publisher.js —— 模拟传感器节点

const mqtt = require('mqtt');

// 1. 连接到 MQTT Broker（无认证）
const client = mqtt.connect('mqtt://test.mosquitto.org');

// 2. 连接成功时执行
client.on('connect', () => {
    console.log('✅ 传感器节点已连接到 Broker');

    // 每 2 秒发布一个随机温度值
    setInterval(() => {
        const temperature = (20 + Math.random() * 15).toFixed(2);
        const timestamp = new Date().toISOString();
        const payload = JSON.stringify({
            sensor: 'temp_01',
            value: parseFloat(temperature),
            unit: '°C',
            timestamp
        });

        // 发布到主题 'home/sensors/temperature'
        // QoS 1：至少一次，确保上报成功
        // retain: false —— 普通消息，不保留
        client.publish('home/sensors/temperature', payload, {
            qos: 1,
            retain: false
        }, (err) => {
            if (err) {
                console.error('发布失败:', err);
            } else {
                console.log(`📤 发布温度: ${temperature} °C`);
            }
        });
    }, 2000);
});

// 3. 设置遗嘱消息：当传感器意外断开时，Broker 会发送此消息通知订阅者
client.on('connect', () => {
    // 应当在连接成功回调中设置，也可在 connect 包的 options 中设置
    // 这里演示在连接后更新遗嘱（不常用，推荐在 connect() 参数中设置）
});

// 正确的做法：在连接参数中设置遗嘱
// 为演示清楚，重新创建连接，但保持文件逻辑简洁，这里用另一个变量展示：
// 更好的方式是在 connect 时传入 will 选项，我们可以在下面 adjust

说明：遗嘱最好在 mqtt.connect 时通过 options 传入。以下为更新后的连接方式：

// 重写 connect 部分，加入遗嘱
const client = mqtt.connect('mqtt://test.mosquitto.org', {
    will: {
        topic: 'home/sensors/temperature/status',
        payload: JSON.stringify({ online: false, sensor: 'temp_01' }),
        qos: 1,
        retain: true   // 保留，让订阅者上线即知状态
    }
});

完整 publisher.js 修正后：

const mqtt = require('mqtt');

// 连接配置：包含遗嘱消息
const client = mqtt.connect('mqtt://test.mosquitto.org', {
    clientId: 'sensor_temp_01_' + Math.random().toString(16).substr(2, 8),
    clean: true,
    keepalive: 10, // 10秒心跳
    will: {
        topic: 'home/sensors/temperature/status',
        payload: JSON.stringify({ online: false, sensor: 'temp_01' }),
        qos: 1,
        retain: true
    }
});

client.on('connect', () => {
    console.log('✅ 传感器节点已连接');
    // 同时发布上线状态
    client.publish('home/sensors/temperature/status', JSON.stringify({
        online: true,
        sensor: 'temp_01',
        timestamp: new Date().toISOString()
    }), { qos: 1, retain: true });

    // 定时发布温度数据
    setInterval(() => {
        const temp = (20 + Math.random() * 15).toFixed(2);
        client.publish('home/sensors/temperature', JSON.stringify({
            sensor: 'temp_01',
            value: parseFloat(temp),
            unit: '°C',
            timestamp: new Date().toISOString()
        }), { qos: 1 }, (err) => {
            if (!err) console.log(`📤 温度发布: ${temp} °C`);
        });
    }, 2000);
});

client.on('error', (err) => console.error('连接错误:', err));

📥 subscriber.js —— 控制中心订阅者

const mqtt = require('mqtt');

const client = mqtt.connect('mqtt://test.mosquitto.org');

client.on('connect', () => {
    console.log('✅ 控制中心已连接到 Broker');

    // 订阅传感器温度主题
    client.subscribe('home/sensors/temperature', { qos: 1 }, (err) => {
        if (!err) console.log('📌 已订阅温度主题');
    });

    // 订阅通配符：监听所有传感器状态
    client.subscribe('home/sensors/+/status', { qos: 1 }, (err) => {
        if (!err) console.log('📌 已订阅传感器状态主题');
    });
});

// 接收消息
client.on('message', (topic, message) => {
    console.log(`📩 收到消息 - 主题: ${topic}`);
    try {
        const data = JSON.parse(message.toString());
        if (topic === 'home/sensors/temperature') {
            console.log(`🌡️  当前温度: ${data.value}${data.unit} (来自 ${data.sensor})`);
            // 这里可以添加阈值判断、报警等业务逻辑
            if (data.value > 30) {
                console.warn('⚠️  高温警报！');
            }
        } else if (topic.endsWith('/status')) {
            console.log(`🔔 传感器状态更新: ${data.online ? '在线' : '离线'}`);
        }
    } catch (e) {
        console.log('原始消息:', message.toString());
    }
});

client.on('error', (err) => console.error('订阅者错误:', err));

运行与测试

打开两个终端。
在先运行订阅者：node subscriber.js，你会看到订阅成功，并且如果之前发布过保留的状态消息，会立即收到。
再运行发布者：node publisher.js，传感器开始每2秒发送温度数据，订阅者终端会实时打印。

测试遗嘱消息：直接按 Ctrl+C 强行结束 publisher 进程（非正常断开），大约在心跳时间后，订阅者会收到 online: false 的状态消息，验证遗嘱机制。

三、方案扩展与生产建议

认证与加密：生产环境务必启用 TLS/SSL 和用户名密码认证。mqtt.connect('mqtts://broker:8883', { username, password, ... })。
本地 Broker：可使用 aedes 在 Node.js 中快速自建 Broker，或使用成熟的 Mosquitto、EMQX 等。
持久化与会话：clean: false 可让 Broker 为客户端保留未处理的消息和订阅关系，重新连接后自动恢复。
消息处理：在订阅者中可接入流处理框架（如 event-stream），或直接对接时序数据库（如 InfluxDB）。
资源优化：对于低功耗传感器，MQTT-SN 协议更适合 UDP 传输，但 Node.js 生态支持有限，可考虑 lwm2m 等替代方案。