RocketMQ 桥接插件配置文件: etc/plugins/emqx_bridge_rocket.conf。

    配置 RocketMQ 桥接规则

    2. ## ${topic}: the RocketMQ topics to which the messages will be published.
    3. ## ${filter}: the mqtt topic (may contain wildcard) on which the action will be performed .
    5. ## Client Connected Record Hook
    6. bridge.rocket.hook.client.connected.1     = {"topic": "ClientConnected"}
    8. ## Client Disconnected Record Hook
    9. bridge.rocket.hook.client.disconnected.1  = {"topic": "ClientDisconnected"}
    11. ## Session Subscribed Record Hook
    12. bridge.rocket.hook.session.subscribed.1   = {"filter": "#",  "topic": "SessionSubscribed"}
    14. ## Session Unsubscribed Record Hook
    15. bridge.rocket.hook.session.unsubscribed.1 = {"filter": "#",  "topic": "SessionUnsubscribed"}
    17. ## Message Publish Record Hook
    18. bridge.rocket.hook.message.publish.1      = {"filter": "#",  "topic": "MessagePublish"}
    19. ## Message Delivered Record Hook
    20. bridge.rocket.hook.message.delivered.1    = {"filter": "#",  "topic": "MessageDeliver"}
    22. ## Message Acked Record Hook
    23. bridge.rocket.hook.message.acked.1        = {"filter": "#",  "topic": "MessageAcked"}

    RocketMQ 桥接规则说明

    1. topic = "ClientConnected",
    2. value = {
    4.          "username": ${username},
    5.          "node": ${node},
    6.          "ts": ${ts}
    7.         }

    设备下线 EMQ X 转发下线事件消息到 RocketMQ:

    1. topic = "ClientDisconnected",
    2. value = {
    3.          "client_id": ${clientid},
    4.          "username": ${username},
    5.          "reason": ${reason},
    6.          "node": ${node},
    7.          "ts": ${ts}
    8.         }

    客户端订阅主题事件转发 RocketMQ

    客户端取消订阅主题事件转发 RocketMQ

    1. topic = "SessionUnsubscribed"
    3. value = {
    4.          "client_id": ${clientid},
    5.          "topic": ${topic},
    6.          "qos": ${qos},
    7.          "node": ${node},
    8.          "ts": ${timestamp}
    9.         }
    2. value = {
    3.          "client_id": ${clientid},
    5.          "topic": ${topic},
    6.          "payload": ${payload},
    7.          "qos": ${qos},
    8.          "node": ${node},
    9.          "ts": ${timestamp}
    10.         }

    MQTT 消息派发 (Deliver) 事件转发 RocketMQ

    1. topic = "MessageDeliver"
    3. value = {
    4.          "client_id": ${clientid},
    5.          "username": ${username},
    6.          "from": ${fromClientId},
    7.          "topic": ${topic},
    8.          "payload": ${payload},
    9.          "qos": ${qos},
    10.          "node": ${node},
    11.          "ts": ${timestamp}
    12.         }

    MQTT 消息确认 (Ack) 事件转发 RocketMQ

    RocketMQ 读取 MQTT 客户端上下线事件消息:

    1. sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer ClientConnected
    3. sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer ClientDisconnected
    1. sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer SessionSubscribed
    3. sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer SessionUnsubscribed

    RocketMQ 读取 MQTT 发布消息:

    1. sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer MessagePublish

    RocketMQ 读取 MQTT 消息发布 (Deliver)、确认 (Ack) 事件:

    默认 payload 被 base64 编码，可通过修改配置 bridge.rocket.encode_payload_type 指定 payload 数据格式。

    启用 RocketMQ 桥接插件