文档中心

云原生消息引擎

最佳实践

设备状态汇报到 Kafka

设备状态汇报到 Kafka

MQTT 场景中，设备状态上报是一个非常常见的场景。当设备信息上报到 MQTT 服务器后，MQTT 服务器需要将全部的 MQTT 消息保存，以便后续的分析。本文将使用 Kafka 实现设备信息的汇报。

场景说明

设备将上报影子信息到其设备 ID 的 MQTT Topic。

MQTT 服务器配置规则，将符合条件的多个 MQTT Topic 转发到一个 Kafka Topic。

准备工作

安装 MQTTX 客户端，参考MQTTX 官方安装文档。

创建云原生消息引擎 MQTT 版实例，计算规格为 mqtt.2k。详细操作请参见创建 MQTT 实例。

创建消息队列 Kafka版实例，版本为 2.8.2。详细操作请参见创建实例。

操作步骤

在消息队列 Kafka版控制台中，创建目标 Topic。

详细操作请参见创建 Topic，创建一个名称为 agg 的 Topic。在实例详情中获取 SASL_PLAINTEXT 接入点的 VPC 接入点信息，如 kafka-xxxxxxxxxxxxxxx.kafka.ivolces.com:9093。

在消息队列 Kafka版控制台中，创建 Kafka 用户，并配置用户名、密码、权限。

详细操作请参见创建 SASL 用户。

在云原生消息引擎 MQTT Dashboard 中，创建 Kafka 生产者连接器。

详细操作请参见创建 Kafka 生产者连接器。

在主机列表中配置 PLAINTEXT 接入点地址

配置用户名、密码

认证方法选择 plain

在云原生消息引擎 MQTT Dashboard 中，配置规则数据输入。

详细操作请参见创建规则。

MQTT 服务器规则匹配 device/# 的全部 Topic

转发到 Kafka 的 agg Topic

在云原生消息引擎 MQTT Dashboard 中，配置规则动作输出。

选择已注册的 Kafka 连接器

输入目标端的 Kafka Topic 名称

结果验证

通过 MQTTX 客户端，模拟 2 个 Device，发送数据到 MQTT 实例的公网接入点。

id00125 发送消息 device/id00125

id00128 发送消息 device/id00128

查看 EMQX 中消息已经命中。

在消息队列 Kafka 实例中，查看到 agg Topic 中已有转发数据。

最近更新时间：2025.08.04 17:46:11

这个页面对您有帮助吗？

有用

有用

无用

无用