云原生消息引擎
云原生消息引擎
文档指南
请输入
- 文档首页
云原生消息引擎MQTT 实例用户指南最佳实践通过自动订阅实现 p2p 指令发布
复制全文
下载 pdf通过自动订阅实现 p2p 指令发布
文档反馈
问问助手MQTT 设备在出厂时通常仅指定 MQTT 服务器的接入地址，但没有指定其所订阅的 MQTT Topic。云上的控制服务（通常是 Java/Go/C++）希望通过根据指定的设备 ID 来实现指令的下发。如某个 Java 实现的后端服务希望向 clientid 为 vin001 的 MQTT 设备下发消息，但 vin001 的设备仅建立了 MQTT 连接，并未主动订阅任何 MQTT Topic。该场景可以借助 MQ + MQTT 组合方案搭配 MQTT 自动订阅，实现类似于 “P2P” （点对点）的通信方式。
架构说明
准备工作
- 消息队列 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 语句会自动配置，无需手动修改。
-
-
- 消息验证。
- 通过消息队列 Kafka 控制台的消息发送功能向 Kafka restart Topic 中发送数据。
- 其可以被正确的转发到连接的 vin001 设备，即代表验证成功。
最近更新时间：2025.08.04 17:30:05
这个页面对您有帮助吗？
有用
有用
无用
无用