Node.js使用MQTT库无法监听Azure IoT Hub事件及设备报错问题

问题分析与修复步骤 #

1. 语法错误引发连接不稳定 #

代码中clientId字段缺少闭合引号，会直接导致语法错误，触发客户端反复重连：

// 错误写法
clientId:"test,
// 修复后
clientId:"test",

2. 订阅主题格式不符合服务端规则 #

Azure IoT Hub中，服务端（后端服务）接收设备上报消息的主题格式需满足：

• 订阅指定设备的上报消息：devices/{deviceId}/messages/events/#（需带通配符#匹配子主题）
• 订阅所有设备的上报消息：messages/events/#

你的代码中订阅主题缺少通配符，且路径不符合服务端订阅逻辑，导致无法接收设备上报数据：

// 错误写法
var subscribetopic = "devices/" + deviceId + "/messages/events/";
// 修复为指定设备的订阅主题
var subscribetopic = "devices/" + deviceId + "/messages/events/#";

3. 身份权限误用（设备密钥无法用于服务端） #

当前代码使用设备的sharedKey生成SAS Token，这是核心错误：设备密钥仅允许设备自身连接IoT Hub，服务端（后端服务）必须使用IoT Hub的服务级共享访问密钥（如iothubowner策略的主/副密钥），并生成对应服务权限的SAS Token。

修改身份验证逻辑：

// 替换为服务级共享访问策略信息
var policyName = "iothubowner";
var serviceSharedKey = "xxxServicePrimaryKeyxxx";

// 生成服务端专用SAS Token
function getServiceSaSToken(hostName, policyName, sharedKey) {
  var sr = encodeUriComponentStrict(hostName);
  var se = Math.round(new Date().getTime() / 1000) + 24 * 3600;
  var StringToSign = sr + "\n" + se;
  var sig = encodeUriComponentStrict(
    CryptoJS.HmacSHA256(
      StringToSign,
      CryptoJS.enc.Base64.parse(sharedKey)
    ).toString(CryptoJS.enc.Base64)
  );
  return `SharedAccessSignature sr=${sr}&sig=${sig}&se=${se}&skn=${policyName}`;
}

// 连接配置更新为服务端身份
var client = mqtt.connect({
  hostname: host,
  port: 8883,
  protocol: "mqtts",
  protocolVersion: 4,
  clientId: "test", // 已修复引号错误
  username: `${host}/?api-version=2021-04-12`, // 服务端username无需拼接deviceId
  password: getServiceSaSToken(host, policyName, serviceSharedKey),
  rejectUnauthorized: false,
});

4. 下发消息主题与格式问题（解决MicroPython设备OsError） #

服务端向设备下发消息的主题应为devices/{deviceId}/messages/devicebound/，需注意：

• 主题末尾不能加#通配符，必须精确匹配
• 消息需使用UTF-8编码的字符串，避免二进制格式导致设备解析失败
• MicroPython设备的OsError可能是消息大小超出设备处理能力、网络中断或设备端未捕获异常，建议设备端添加错误捕获逻辑，同时确保消息大小不超过IoT Hub默认256KB限制

修复后的完整代码示例 #

var mqtt = require("mqtt");
var CryptoJS = require("crypto-js");

var host = "iothub.azure-devices.net";
var deviceId = "test";
// 服务级共享访问策略信息
var policyName = "iothubowner";
var serviceSharedKey = "xxxServicePrimaryKeyxxx";

// 服务端下发消息主题
var publishTopic = `devices/${deviceId}/messages/devicebound/`;
// 服务端订阅指定设备上报消息主题
var subscribeTopic = `devices/${deviceId}/messages/events/#`;

function encodeUriComponentStrict(str) {
  return encodeURIComponent(str).replace(/[!'()*]/g, function (c) {
    return "%" + c.charCodeAt(0).toString(16);
  });
}

// 生成服务端SAS Token
function getServiceSaSToken(hostName, policyName, sharedKey) {
  var sr = encodeUriComponentStrict(hostName);
  var se = Math.round(new Date().getTime() / 1000) + 24 * 3600;
  var StringToSign = sr + "\n" + se;
  var sig = encodeUriComponentStrict(
    CryptoJS.HmacSHA256(
      StringToSign,
      CryptoJS.enc.Base64.parse(sharedKey)
    ).toString(CryptoJS.enc.Base64)
  );
  return `SharedAccessSignature sr=${sr}&sig=${sig}&se=${se}&skn=${policyName}`;
}

var client = mqtt.connect({
  hostname: host,
  port: 8883,
  protocol: "mqtts",
  protocolVersion: 4,
  clientId: "test",
  username: `${host}/?api-version=2021-04-12`,
  password: getServiceSaSToken(host, policyName, serviceSharedKey),
  rejectUnauthorized: false,
});

client.on("connect", function () {
  console.log("服务端MQTT连接成功！");
  client.subscribe(subscribeTopic, (err) => {
    if (err) {
      console.error("订阅失败：", err);
    } else {
      console.log(`已订阅主题：${subscribeTopic}`);
    }
  });
});

client.on("reconnect", function () {
  console.log("MQTT正在重连...");
});

client.on("close", function () {
  console.log("MQTT连接已关闭");
});

client.on("message", function (topic, message) {
  var content = message.toString("utf-8");
  console.log(`收到设备上报消息：主题=${topic}，内容=${content}`);
});

// 示例：向设备下发消息
function sendMessageToDevice(content) {
  client.publish(publishTopic, content, (err) => {
    if (err) {
      console.error("下发消息失败：", err);
    } else {
      console.log(`已下发消息到设备：${content}`);
    }
  });
}

// 测试下发（取消注释即可）
// sendMessageToDevice("Hello from server!");

内容的提问来源于stack exchange，提问作者user28052177