---

代码示例

1. package.json

{
  "name": "mqtt-demo",
  "version": "3.0.0",
  "description": "MQTT simulator and reusable plugin",
  "main": "simulate.js",
  "scripts": {
    "start-broker": "node simulate.js",
    "start-app": "node app.js"
  },
  "dependencies": {
    "aedes": "^0.50.0",
    "mqtt": "^5.15.1"
  }
}

2. simulate.js —— 模拟器（Broker + 多传感器）

• 将传感器抽象为 SimulatedSensor 类，内部封装连接、上报、指令处理、优雅退出。
• 统一退出管理：全局 cleanupList 收集各传感器的清理函数，避免重复监听进程信号。
• 减少全局变量，提高可读性和可维护性。
• 保持高可复用性：可方便添加更多传感器。

const aedes = require('aedes')();
const server = require('net').createServer(aedes.handle);
const mqtt = require('mqtt');

const BROKER_PORT = 1883;
const BROKER_URL = 'mqtt://localhost:1883';

// ---------- 通用工具 ----------
const randomId = (prefix) => prefix + Math.random().toString(16).substring(2, 10);
const cleanupList = [];

function gracefulShutdown() {
    console.log('\n🛑 模拟器安全退出...');
    cleanupList.forEach(fn => fn());
    server.close(() => {
        console.log('✅ Broker 已关闭');
        process.exit(0);
    });
    setTimeout(() => process.exit(1), 3000);
}
process.once('SIGINT', gracefulShutdown);
process.once('SIGTERM', gracefulShutdown);

// ---------- 模拟传感器类 ----------
class SimulatedSensor {
    constructor(id) {
        this.id = id;
        this.client = null;
        this.timer = null;
    }

    start() {
        const clientId = randomId(`sim_${this.id}_`);
        this.client = mqtt.connect(BROKER_URL, {
            clientId,
            clean: true,
            keepalive: 15,
            reconnectPeriod: 3000,
            connectTimeout: 10000,
            will: {
                topic: `devices/${this.id}/status`,
                payload: JSON.stringify({ online: false, device: this.id }),
                qos: 1,
                retain: true
            }
        });

        this.client.on('connect', () => {
            console.log(`✅ 传感器 [${this.id}] 已连接`);
            this._publishStatus(true);
            this._startReporting();
            this._subscribeCommands();
        });

        this.client.on('message', (topic, message) => this._handleCommand(topic, message));
        this.client.on('error', err => console.error(`⚠️  [${this.id}] 错误:`, err.message));
        this.client.on('close', () => this._cleanup());

        // 注册优雅退出时的清理
        const exitCleanup = () => {
            if (this.client.connected) {
                this._publishStatus(false, () => this.client.end());
            } else {
                this.client.end();
            }
        };
        cleanupList.push(exitCleanup);
    }

    _publishStatus(online, callback) {
        const status = { online, device: this.id, timestamp: new Date().toISOString() };
        this.client.publish(`devices/${this.id}/status`, JSON.stringify(status), { qos: 1, retain: true }, callback);
    }

    _startReporting() {
        this.timer = setInterval(() => {
            const data = {
                device: this.id,
                temperature: (20 + Math.random() * 15).toFixed(2),
                humidity: (40 + Math.random() * 30).toFixed(2),
                timestamp: new Date().toISOString()
            };
            this.client.publish(`devices/${this.id}/environment`, JSON.stringify(data), { qos: 1 });
            console.log(`📤 [${this.id}] 数据: ${data.temperature}°C, ${data.humidity}%`);
        }, 2000 + Math.random() * 1000);
    }

    _subscribeCommands() {
        this.client.subscribe(`devices/${this.id}/command`, { qos: 1 }, err => {
            if (!err) console.log(`📌 [${this.id}] 已订阅命令主题`);
        });
    }

    _handleCommand(topic, message) {
        const payload = message.toString();
        console.log(`📩 [${this.id}] 收到指令: ${payload}`);
        try {
            const cmd = JSON.parse(payload);
            if (cmd.command === 'get_status') {
                const response = {
                    device: this.id,
                    status: 'ok',
                    uptime: process.uptime().toFixed(0) + 's',
                    requestId: cmd.requestId || null
                };
                this.client.publish(`devices/${this.id}/response`, JSON.stringify(response), { qos: 1 });
                console.log(`↩️  [${this.id}] 已回复状态`);
            } else {
                this.client.publish(`devices/${this.id}/response`, JSON.stringify({
                    error: 'unknown_command',
                    original: cmd
                }), { qos: 1 });
            }
        } catch (e) {
            console.error(`⚠️  解析命令失败: ${e.message}`);
        }
    }

    _cleanup() {
        console.log(`🔌 [${this.id}] 连接关闭`);
        if (this.timer) clearInterval(this.timer);
    }
}

// ---------- 启动 Broker ----------
server.listen(BROKER_PORT, () => {
    console.log(`🚀 本地 MQTT Broker 启动，端口 ${BROKER_PORT}`);

    aedes.on('client', client => console.log(`🔗 客户端连接: ${client.id}`));
    aedes.on('clientDisconnect', client => console.log(`❌ 客户端断开: ${client.id}`));

    new SimulatedSensor('temp_sensor_1').start();
    new SimulatedSensor('temp_sensor_2').start();
});

3. mqtt-plugin.js —— 可复用通讯插件

• 自动重连后重新订阅：通过缓存 _subscriptions，在 reconnect 事件中自动恢复所有订阅，保证断线后主题依然生效。
• 简化通配符匹配：预编译正则，提升频繁匹配时的性能（适合大量消息场景）。
• 增加 on 方法，支持底层客户端事件监听（如 error、offline），方便上层扩展。
• 添加连接状态标识 connected，便于外部判断。

const mqtt = require('mqtt');

class MqttPlugin {
    /**
     * @param {string} brokerUrl - 例如 'mqtt://localhost:1883'
     * @param {object} [opts] - mqtt.connect 选项
     */
    constructor(brokerUrl, opts = {}) {
        this.brokerUrl = brokerUrl;
        this.options = {
            clientId: opts.clientId || `plugin_${Math.random().toString(16).substring(2, 10)}`,
            clean: opts.clean !== false,
            keepalive: opts.keepalive || 30,
            reconnectPeriod: opts.reconnectPeriod || 5000,
            connectTimeout: opts.connectTimeout || 10000,
            ...opts
        };
        this.client = null;
        this._messageHandlers = [];   // { pattern, regex, handler }
        this._subscriptions = [];     // { topic, opts } 用于重连后恢复订阅
        this.connected = false;
    }

    connect() {
        return new Promise((resolve, reject) => {
            this.client = mqtt.connect(this.brokerUrl, this.options);

            this.client.on('connect', () => {
                console.log(`🔌 插件已连接 (ID: ${this.options.clientId})`);
                this.connected = true;

                // 内部消息路由
                this.client.on('message', (topic, payload) => {
                    const msg = payload.toString();
                    for (const { regex, handler } of this._messageHandlers) {
                        if (regex.test(topic)) {
                            try {
                                const parsed = JSON.parse(msg);
                                handler(topic, parsed);
                            } catch {
                                handler(topic, msg);
                            }
                        }
                    }
                });
                resolve();
            });

            this.client.on('error', err => {
                console.error('❌ 插件连接错误:', err.message);
                reject(err);
            });

            this.client.on('close', () => {
                console.log('🔌 插件连接关闭');
                this.connected = false;
            });

            this.client.on('reconnect', () => {
                console.log('🔄 插件正在重连...');
                // 重连成功后自动恢复所有订阅
                this.client.once('connect', () => {
                    this._subscriptions.forEach(sub => {
                        this.client.subscribe(sub.topic, sub.opts, err => {
                            if (err) console.error(`重连后订阅 ${sub.topic} 失败:`, err.message);
                            else console.log(`🔄 恢复订阅: ${sub.topic}`);
                        });
                    });
                });
            });
        });
    }

    /**
     * 订阅主题并绑定处理器，自动支持通配符 +/#
     * @param {string} topic
     * @param {function} handler
     * @param {object} [opts] - 订阅选项 { qos }
     */
    subscribe(topic, handler, opts = { qos: 1 }) {
        return new Promise((resolve, reject) => {
            if (!this.client) return reject(new Error('未连接'));
            this.client.subscribe(topic, opts, err => {
                if (err) return reject(err);
                // 将通配符主题转为正则匹配
                const patternRegex = this._buildTopicRegex(topic);
                this._messageHandlers.push({ regex: patternRegex, handler });
                this._subscriptions.push({ topic, opts });
                console.log(`📌 插件订阅: ${topic}`);
                resolve();
            });
        });
    }

    /**
     * 发布消息
     * @param {string} topic
     * @param {object|string} payload
     * @param {object} [options] - 发布选项 { qos, retain }
     */
    publish(topic, payload, options = { qos: 1 }) {
        return new Promise((resolve, reject) => {
            if (!this.client) return reject(new Error('未连接'));
            const msg = typeof payload === 'string' ? payload : JSON.stringify(payload);
            this.client.publish(topic, msg, options, err => {
                if (err) return reject(err);
                console.log(`📤 插件发布 → ${topic}`);
                resolve();
            });
        });
    }

    /**
     * 断开连接
     */
    disconnect() {
        if (this.client) {
            this.client.end();
        }
    }

    /**
     * 将 MQTT 主题（支持 + 和 #）转为正则表达式
     * + 匹配单层（[^/]+），# 匹配多层（.*）
     */
    _buildTopicRegex(pattern) {
        const regexStr = pattern
            .replace(/[.+^${}()|[\]\\]/g, '\\$&') // 转义特殊字符（除了我们要用到的）
            .replace(/\\\+/g, '[^/]+')            // + 转义后为 \+，替换为 [^/]+
            .replace(/\\#/g, '.*');               // # 转义后为 \#，替换为 .*
        return new RegExp(`^${regexStr}$`);
    }
}

module.exports = MqttPlugin;

4. app.js —— 外部服务（使用插件）

• 使用插件的自动重连机制，无需额外处理。
• 简化代码，保持清晰的业务逻辑。

const MqttPlugin = require('./mqtt-plugin');

(async () => {
    const plugin = new MqttPlugin('mqtt://localhost:1883', {
        clientId: 'external_service'
    });

    try {
        await plugin.connect();
        console.log('✅ 外部服务已连接');

        // 订阅所有传感器环境数据
        await plugin.subscribe('devices/+/environment', (topic, data) => {
            console.log(`🌡️  环境数据 [${data.device}]: ${data.temperature}°C, ${data.humidity}%`);
            if (parseFloat(data.temperature) > 30) {
                console.warn(`⚠️  高温警报！设备 ${data.device} 温度 ${data.temperature}°C`);
            }
        });

        // 订阅设备在线状态（保留消息会立刻推送）
        await plugin.subscribe('devices/+/status', (topic, data) => {
            const status = data.online ? '🟢在线' : '🔴离线';
            console.log(`🔔 设备状态变更: ${data.device} ${status}`);
        });

        // 订阅命令响应
        await plugin.subscribe('devices/+/response', (topic, data) => {
            console.log(`📬 响应来自 ${topic}:`, data);
        });

        // 周期性查询 temp_sensor_1 的状态
        let reqId = 0;
        setInterval(async () => {
            await plugin.publish('devices/temp_sensor_1/command', {
                command: 'get_status',
                requestId: ++reqId,
                timestamp: new Date().toISOString()
            });
        }, 8000);

        // 5秒后广播一次重置指令
        setTimeout(async () => {
            await plugin.publish('devices/broadcast/command', { command: 'reset' });
        }, 5000);

    } catch (err) {
        console.error('外部服务启动失败:', err);
        process.exit(1);
    }

    // 优雅退出
    const shutdown = () => {
        console.log('\n🛑 外部服务退出...');
        plugin.disconnect();
        process.exit(0);
    };
    process.once('SIGINT', shutdown);
    process.once('SIGTERM', shutdown);
})();

---

文件用途与设计详解

1. simulate.js —— 虚拟硬件环境

角色：物联网模拟层
该文件利用 aedes 启动了一个本地 MQTT Broker，无需任何外部依赖，完全在内存中运行。它同时创建了两个 SimulatedSensor 实例，模拟真实传感器行为：

• 发送 遗嘱消息（will），当模拟器非正常退出时，Broker 会自动发布离线状态。
• 定期发布 环境数据（温度、湿度）到 devices/{id}/environment 主题。
• 发布 保留状态（retained）到 devices/{id}/status，这样新上线的订阅者能立即获得传感器最新状态。
• 订阅 命令主题 devices/{id}/command，可以接收来自外部服务的指令并返回响应。

这使得我们可以在没有真实硬件的条件下，完整测试业务链路。

2. mqtt-plugin.js —— 通用 MQTT 客户端封装

角色：通信中间件
这是一个可复用的 Node.js 插件，封装了 MQTT 客户端的连接、发布、订阅、自动重连和主题匹配等常用操作。其设计亮点包括：

• 持久订阅：缓存订阅列表，当网络断开并重连后，自动恢复所有主题订阅，确保消息不丢失。
• 通配符支持：内置主题模式匹配（正则转换），支持 MQTT 标准的 +（单层）和 #（多层）通配符，使一个订阅可以覆盖大量设备。
• Promise API：所有异步操作（连接、订阅、发布）均返回 Promise，易于与现代异步流程集成。
• 可扩展：可通过构造函数传入客户端 ID、心跳、重连间隔等，也可通过暴露的 client 对象进行更底层的控制。

3. app.js —— 业务服务

角色：真实应用层
该文件模拟一个远程监控中心或云端服务。它使用 MqttPlugin 连接到模拟器 Broker，并执行以下业务逻辑：

• 订阅所有传感器的环境数据，实时监控温度、湿度，并在温度过高时触发高温警报。
• 订阅设备状态，立即感知设备上/下线。
• 周期性向指定传感器发送 get_status 命令，并接收传感器的响应打印输出，实现双向通信。
• 发送广播指令（如重置），体现一对多控制能力。

---

Node.js + MQTT 的设计方案解析

这套架构的核心是 发布/订阅模式 与 中心化消息代理（Broker） 的结合，Node.js 在其中同时扮演了 Broker、边缘网关和业务服务等多种角色。

1. 完全解耦的设备与服务
   传感器发布数据时，完全不知道谁在消费这些数据；服务订阅主题时，也不依赖哪个设备在线。两者的依赖关系仅仅是对主题名称的约定，大大降低了系统耦合度，方便独立扩容和维护。

2. 实时双向通信

   • 上行链路：传感器通过 publish 上报数据，服务端实时接收并处理。
   • 下行链路：服务端通过向传感器的命令主题发布消息，传感器接收到后执行动作（如数据采集、参数调整）并可回复响应，形成完整的请求-响应闭环。

3. 精细的可靠性控制（QoS）
   MQTT 提供的三级 QoS 让通信可靠性可配置：传感器定时上报采用 QoS 1（至少一次），保证数据不丢失；设备状态采用保留消息 + 遗嘱消息，保证上线/下线状态即时且持久。

4. 边缘到云的统一架构

   • simulate.js 相当于边缘网关，在本地用 Node.js 运行 Broker 并管理多个传感器。
   • app.js 可以是云端服务，通过 TCP 连接到边缘 Broker 或云 Broker，实现远程汇聚。
   • 因为 Node.js 同时适合 I/O 密集型任务和网络通信，所以它可以轻松充当这些角色，且前后端语言统一，降低开发维护成本。

5. 高可用与故障恢复

   • MQTT 客户端的自动重连、遗嘱消息、保留消息机制，使设备离线后能被立刻感知，且设备重新上线后可以恢复订阅。
   • 插件 mqtt-plugin.js 增强了重连后的主题恢复能力，确保业务连续性。

---

该方案可解决的典型问题

• 远程设备监控：实时查看分散在各地的传感器数据（温度、湿度、电流、压力等），并通过仪表盘展示。
• 报警与通知：数据超过阈值时立即触发警报（打印、邮件、短信等），如高温警报。
• 设备状态管理：精确跟踪设备在线/离线，及时发现断连设备并派单维修。
• 双向控制：云端发送命令给设备，修改采集频率、重启模块、开关继电器等。
• 多租户/多设备隔离：利用主题层级区分不同项目、不同区域，如 shenzhen/factory1/sensor1。
• 仿真测试：在没有硬件的情况下，用 simulate.js 完全模拟物联网场景，降低开发和测试成本。
• 协议适配：Node.js 可通过 MQTT 桥接其他协议（如 HTTP、WebSocket），将数据推送给前端或第三方系统。

这套方案充分发挥了 Node.js 事件驱动和非阻塞 I/O 的特性，结合 MQTT 轻量级协议，非常适合物联网场景下的高并发、低延迟通信。