Node.js + MQTT无法连接Amazon MQ 求问题排查与解决方法

问题描述 #

我正在用Node.js和MQTT测试基于Apache ActiveMQ引擎的Amazon MQ，未达到预期效果。操作流程如下：

1. 在AWS控制台创建了Apache ActiveMQ代理（Active/stand-by模式、mq.t2.micro规格），创建成功后获取了所有连接信息。
2. 本地编写了producer.js和consumer.js，期望启动consumer后，运行producer能让consumer收到消息。

实际操作结果：先执行node consumer.js，终端仅显示“Consumer script has been run”；再执行node producer.js，终端仅显示“Producer script has been run”，无消息接收，连接未正常建立。

现有代码 #

consumer.js

const mqtt = require('mqtt')

const config = {
    username: `myusernmae`,
    password: `mypassword`,
    mqEndpoint: `wss://b-xxxxxx-d21c-xxxx-xxxx-xxxxxxxxxxxxx-1.mq.eu-west-2.amazonaws.com:61619`, // 与producer使用同一端点，来自Broker连接信息的WSS部分
    clientId: 'my_client_id',
}

const client = mqtt.connect(config.mqEndpoint, {
   username: config.username,
   password: config.password,
    clientId: config.clientId,
});

// 连接成功后订阅主题
client.on('connect', function() {
    console.log("connected")
    client.subscribe(topic, function (err) {
        if(err) console.log(err)
    })
})

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

// 打印收到的消息
client.on('message', function (topic, message) {
    console.log(`message received on ${topic}: ${message.toString()}`)
})

console.log("Consumer script has been run");

producer.js

const mqtt = require('mqtt')

const config = {
    username: `myusernmae`,
    password: `mypassword`,
    host: `wss://b-xxxxxx-d21c-xxxx-xxxx-xxxxxxxxxxxxx-1.mq.eu-west-2.amazonaws.com`,// 与consumer使用同一端点，来自Broker连接信息的WSS部分
    clientId: 'my_client_id',
    port: 61619,
    topic: `some/topic`,
}

let client = mqtt.connect(config.host, {
    username: config.username,
    password: config.password,
    clientId: config.clientId,
})

let message = JSON.stringify({
    message: "Hello from Lambda"
})

client.on('connect', function() {
    client.publish(config.topic, message)
    client.end()
})

console.log("Product script has been run");

问题分析与修复方案 #

1. 未定义订阅主题（consumer.js） #

consumer的connect回调中使用了未声明的topic变量，导致订阅失败甚至触发运行时错误。需在config中添加与producer一致的主题配置。

2. ClientId重复 #

producer和consumer使用完全相同的clientId，根据MQTT协议，后连接的客户端会踢掉先连接的客户端，导致consumer在producer启动后被强制下线，无法接收消息。必须为两个客户端设置唯一的clientId。

3. Producer连接配置错误 #

producer中mqtt.connect的参数格式错误：传入的host包含wss://前缀，同时单独指定port，会导致客户端无法正确解析连接地址。应直接传入完整的WSS端点（包含端口）。

4. Producer发布后立即断开连接 #

producer在connect回调中直接调用client.end()，此时publish操作可能未完成（MQTT publish为异步操作），导致消息未成功发送到Broker。需等待publish确认后再断开连接。

修复后的代码 #

修复后的consumer.js

const mqtt = require('mqtt')

const config = {
    username: `myusernmae`,
    password: `mypassword`,
    mqEndpoint: `wss://b-xxxxxx-d21c-xxxx-xxxx-xxxxxxxxxxxxx-1.mq.eu-west-2.amazonaws.com:61619`,
    clientId: 'consumer_client_id', // 唯一客户端ID
    topic: `some/topic` // 添加主题配置
}

const client = mqtt.connect(config.mqEndpoint, {
    username: config.username,
    password: config.password,
    clientId: config.clientId,
});

client.on('connect', function() {
    console.log("connected")
    // 使用配置好的主题订阅
    client.subscribe(config.topic, function (err) {
        if(err) {
            console.error("订阅失败:", err)
        } else {
            console.log(`已订阅主题: ${config.topic}`)
        }
    })
})

client.on('error', function (error) {
    console.error("连接错误:", error)
})

client.on('message', function (topic, message) {
    console.log(`收到消息 [${topic}]: ${message.toString()}`)
})

console.log("Consumer script has been run");

修复后的producer.js

const mqtt = require('mqtt')

const config = {
    username: `myusernmae`,
    password: `mypassword`,
    mqEndpoint: `wss://b-xxxxxx-d21c-xxxx-xxxx-xxxxxxxxxxxxx-1.mq.eu-west-2.amazonaws.com:61619`, // 完整连接端点
    clientId: 'producer_client_id', // 唯一客户端ID
    topic: `some/topic`,
}

let client = mqtt.connect(config.mqEndpoint, {
    username: config.username,
    password: config.password,
    clientId: config.clientId,
})

let message = JSON.stringify({
    message: "Hello from Lambda"
})

client.on('connect', function() {
    console.log("Producer connected")
    // 等待publish确认后再断开连接
    client.publish(config.topic, message, function(err) {
        if(err) {
            console.error("消息发布失败:", err)
        } else {
            console.log("消息已成功发布")
        }
        client.end()
    })
})

client.on('error', function (error) {
    console.error("Producer连接错误:", error)
})

console.log("Producer script has been run");

额外检查项 #

• 确认AWS Broker安全组已开放61619端口（WSS端口），允许本地IP访问。
• 检查用户名、密码是否正确（注意代码中myusernmae为拼写错误，需替换为实际用户名）。
• 确保使用最新版mqtt包，执行npm install mqtt@latest更新。

内容的提问来源于stack exchange，提问作者Wai Yan Hein