2. 发布订阅基础消息

任务目标

本案例将使用MQTT.js在Node.js环境中实现一个最小可运行的发布/订阅闭环：创建一个客户端订阅指定主题，另一个客户端向该主题发布文本消息，通过接收回调验证消息传递的实时性。

环境依赖

需要Node.js 14+环境，并安装MQTT.js库：

npm init -y
npm install mqtt

完整实现代码

将以下代码保存为 pubsub.js：

const mqtt = require('mqtt');

// 使用公共测试Broker（无需本地部署）
const brokerUrl = 'mqtt://broker.emqx.io:1883';
const topic = 'demo/100case/basic';

// 创建订阅端客户端
const subClient = mqtt.connect(brokerUrl, { clientId: 'sub_' + Date.now() });

subClient.on('connect', () => {
    console.log('[订阅端] 已连接Broker');
    subClient.subscribe(topic, { qos: 0 }, (err) => {
        if (err) {
            console.error('[订阅端] 订阅失败:', err.message);
            process.exit(1);
        }
        console.log(`[订阅端] 已订阅主题: ${topic}`);
    });
});

subClient.on('message', (receivedTopic, payload) => {
    const msg = payload.toString();
    console.log(`[订阅端] 收到消息 | 主题: ${receivedTopic} | 内容: ${msg}`);
    console.log('[验证] 消息实时接收成功，测试通过');
    subClient.end();
    pubClient.end();
    process.exit(0);
});

// 创建发布端客户端
const pubClient = mqtt.connect(brokerUrl, { clientId: 'pub_' + Date.now() });

pubClient.on('connect', () => {
    console.log('[发布端] 已连接Broker');
    // 延迟1秒确保订阅已建立
    setTimeout(() => {
        const message = `Hello MQTT - ${new Date().toISOString()}`;
        pubClient.publish(topic, message, { qos: 0 }, (err) => {
            if (err) {
                console.error('[发布端] 发布失败:', err.message);
                process.exit(1);
            }
            console.log(`[发布端] 已发布消息: ${message}`);
        });
    }, 1000);
});

// 超时保护（10秒未收到消息则退出）
setTimeout(() => {
    console.error('[超时] 10秒内未收到消息，请检查网络或Broker状态');
    process.exit(1);
}, 10000);

运行与验证

在终端执行以下命令启动测试：

node pubsub.js

通过条件：控制台依次输出以下信息即表示验证通过：

1. 订阅端和发布端均显示“已连接Broker”；
2. 订阅端显示“已订阅主题: demo/100case/basic”；
3. 发布端显示“已发布消息: Hello MQTT - ...”；
4. 订阅端显示“收到消息”且内容与发布端一致；
5. 最终输出“消息实时接收成功，测试通过”并自动退出。

关键说明

• 使用公共Broker broker.emqx.io，无需本地搭建服务端；若网络受限可替换为本地Mosquitto地址（如 mqtt://localhost:1883）。
• 客户端ID添加时间戳后缀避免冲突；QoS设为0，验证默认消息级别下的实时转发。
• 发布端延迟1秒发送是为了确保订阅关系已在Broker注册，避免消息丢失导致误判。
• 10秒超时保护用于处理网络异常或Broker不可达的情况，避免进程挂起。