文档中心

云原生消息引擎

最佳实践

通过自动订阅实现 p2p 指令发布

MQTT 设备在出厂时通常仅指定 MQTT 服务器的接入地址，但没有指定其所订阅的 MQTT Topic。云上的控制服务（通常是 Java/Go/C++）希望通过根据指定的设备 ID 来实现指令的下发。如某个 Java 实现的后端服务希望向 clientid 为 vin001 的 MQTT 设备下发消息，但 vin001 的设备仅建立了 MQTT 连接，并未主动订阅任何 MQTT Topic。该场景可以借助 MQ + MQTT 组合方案搭配 MQTT 自动订阅，实现类似于 “P2P” （点对点）的通信方式。

架构说明

准备工作

创建消息队列 Kafka 版实例，并创建一个名称为 restart 的 Topic。详细操作请参见创建实例和创建 Topic。

消息队列 Kafka 版中配置 Kafka 用户，其对 restart Topic 有读写权限。详细操作请参见创建 SASL 用户和创建 ACL。

测试用户：名称 root，密码 ******。

MQTT Dashboard 中完成用户认证和授权相关配置，参考快速构建第一个 MQTT 连接。

测试用户：名称 user，密码 123456。

确保消息队列 Kafka 版和云原生消息引擎 MQTT 版在同一个 VPC 中。

操作步骤

创建 Kafka 消费者连接器，为 EMQX 配置 Kafka 消费者订阅。

详细操作请参见创建连接器。

选择连接器类型为 Kafka消费者。

填写 Kafka VPC 接入点、认证、认证方法、用户名、密码。

配置转发规则。

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

在界面右侧单击数据输入，然后单击编辑。

将数据输入配置为 Kafka消费者。

输入类型选择 Kafka 消费者。

配置所指定的 Kafka 消费者连接器。

声明该规则所订阅的 Kafka 主题名称为 restart。

选择 Key 和 Value 编码模式

示例中为 Base64 和 None，可根据实际情况修改。如修改编码模式，后续规则 SQL 语句相应字段的处理函数也需要同步修改。

单击动作输出，然后单击添加动作，配置动作输出。

动作类型选择消息重发布。

声明该规则所订阅的主题名称为 /restart/${key}。

指定订阅 Kafka Topic 数据后，经过 EMQX 处理转发的 MQTT Topic 为动态 Topic，其来自 Kafka 消息 Key。Java 后端服务发送消息到 Kafka 时 clientid 被有意的设置为消息 key，其被 EMQX 规则处理为 /restart/key。设备自动订阅其 /restart/vin 的 MQTT Topic，消息 key 与 clientid 相同，如均为 vin001，即完成消息与设备的订阅绑定。

设置 MQTT payload 为 {aim} 字段。

编辑规则 SQL。

将 json 格式消息 value 中的 aim 字段作为新的 aim 字段（根据上文的规则，其被传递到 payload 中）。

From 字段在配置数据输入为 kafka 消费者时，其 SQL 语句会自动配置，无需手动修改。

SELECT
value.aim as aim, -- payload 配置为 {aim}
base64_decode(key) as key -- payload中未包含此字段，故不会在本例子中实际传递
FROM
"$bridges/kafka_consumer:kafka" -- 该字段为系统自动更新

消息验证。

通过消息队列 Kafka 控制台的消息发送功能向 Kafka restart Topic 中发送数据。

其可以被正确的转发到连接的 vin001 设备，即代表验证成功。

最近更新时间：2025.08.04 17:30:05

这个页面对您有帮助吗？

有用

无用