多端点向同一AWS IoT Thing主题发布数据时设备数据中断问题排查

我正在测试基于AWS IoT Core的IoT项目，已完成MQTT Broker、Thing、连接策略及认证证书的配置。使用策略指定的ClientID，Raspberry Pi Pico(RP2040)可以正常向MQTT Broker发布温湿度、土壤pH、环境光等传感器数据，Broker会将数据写入DynamoDB，再通过Node.js API监听并在Web应用实时展示，这部分功能完全正常。

但当我通过平台API向同一主题发布阈值数据后，设备后续发布的数据立即不再被Broker接收。我原本以为是并发连接的问题，已经调整为平台仅在设置新阈值时连接Broker，发布后立即关闭连接，但问题依然存在。

这是否意味着AWS IoT Thing不支持多设备向同一主题发布？是否需要为不同设备创建独立的主题？

相关代码 #

平台发布阈值的API代码

// -----------Publish to broker---------------
// DHT sensor component threshold
router.get('/dht-threshold-mqtt-publish', async (req, res) => {
  console.log("\nData Publishing ---------------------------");
  try {
    const { project_id, user_id } = req.query;

    // Fetch data from MongoDB based on project_id and user_id
    const dhtSensorData = await DHTSensor.findOne({ project_id, user_id });

    if (!dhtSensorData) {// validate if data isa available
      return res.status(404).json({ success: false, message: 'DHTSensor data not found' });
    }

    // Extract the temp_threshold from the retrieved data
    const tempThreshold = dhtSensorData.temp_threshold.toString();
    const humidThreshold = dhtSensorData.humid_threshold.toString();

    console.log("Component: DHT\nPublished thresholds;\n -Temperature: "+tempThreshold+"\n -Humidity: "+humidThreshold);

    // Construct the JSON payload
    const jsonPayload = JSON.stringify({
      user_id: user_id.toString(),
      project_id: project_id.toString(),
      dht_temperature_threshold: tempThreshold,
      dht_humidity_threshold: humidThreshold,
    });

    // call MQTT setup
    await mqttDataPublishHandler(jsonPayload, 'DHTSensor', res);

  } catch (error) {
    console.error('Error:', error);
    res.status(500).json({ success: false, message: 'Internal server error' });
  }
});

MQTT发布处理函数

// MQTT broker data publish handler
async function mqttDataPublishHandler(dataPayload, targetComponent, res) {
  const mqttBrokerUrl = process.env.MQTT_BROKER_URL; // MQTT broker URL
  const topic = process.env.MQTT_PUB_TOPIC;
  const clientId = process.env.MQTT_CLIENT_ID; // Set your unique client_id
  console.log("MQTT broker url: ", mqttBrokerUrl);

  const mqttOptions = {
    clientId: clientId,
    protocol: 'mqtts', // Use 'mqtts' for MQTT over TLS/SSL
    rejectUnauthorized: true, // Set to "false" to ignore certificate validation (for testing only)
    key: fs.readFileSync(process.env.MQTT_CLIENT_KEY_PATH), // Set path to your client key
    cert: fs.readFileSync(process.env.MQTT_CLIENT_CERT_PATH), // Set path to your client certificate
    ca: fs.readFileSync(process.env.MQTT_CA_CERT_PATH), // Set path to your CA certificate
  };

  const mqttClient = mqtt.connect(mqttBrokerUrl, mqttOptions);

  let responseSent = false;

  mqttClient.on('error', (err) => {
    if (!responseSent) {
      console.error('MQTT Connection Error:', err);
      res.status(500).json({ success: false, message: 'MQTT Connection Error' });
      responseSent = true;
    }
  });

  mqttClient.on('connect', () => {
    // Publish payload to the specified MQTT topic
    mqttClient.publish(topic, dataPayload.toString(), (err) => {
      if (!responseSent) {
        if (err) {
          console.error('MQTT Publish Error:', err);
          res.status(500).json({ success: false, message: 'MQTT Publish Error' });
        } else {
          console.log(targetComponent+" threshold data published to MQTT broker.");
          // Close the MQTT connection
          mqttClient.end();
          // Respond to the client
          res.json({ success: true, data: dataPayload });
        }
        responseSent = true;
      }
    });
  });
}

问题根源与解决方案 #

AWS IoT Core完全支持多客户端向同一主题发布数据，问题出在ClientID重复上：

• ClientID唯一性要求：MQTT协议规定同一Broker下的ClientID必须唯一。你的平台API和设备固件使用了同一个MQTT_CLIENT_ID，当平台API连接Broker时，会触发AWS IoT Core的会话踢除机制——新连接会强制断开旧的设备连接，导致设备无法再向Broker发送数据。

• 修复步骤：

  1. 为平台API生成独立的ClientID，比如在原有ClientID基础上添加后缀（如${process.env.MQTT_CLIENT_ID}-backend），确保和设备的ClientID不重复。
  2. 检查AWS IoT Thing的连接策略，确保平台API使用的ClientID（或对应的Thing）拥有该主题的发布权限。
  3. 设备端添加重连逻辑，避免被踢除后无法自动恢复连接。

• 主题设计建议：虽然同一主题支持多客户端发布，但从架构清晰性和权限管控角度，建议按设备/功能划分主题，比如：

  • 设备发布数据：devices/{device-id}/sensor-data
  • 平台下发阈值：devices/{device-id}/thresholds
    这样后续扩展或排查问题会更方便，也能更精细地控制不同客户端的主题权限。

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