Pulsar 桥接

EMQ X 桥接转发 MQTT 消息到 Pulsar 集群:

image

Pulsar 桥接插件配置文件: etc/plugins/emqx_bridge_pulsar.conf。

配置 Pulsar 集群地址

  1. ## Pulsar 服务器集群配置
  2. ## bridge.pulsar.servers = 127.0.0.1:6650,127.0.0.2:6650,127.0.0.3:6650
  3. bridge.pulsar.servers = 127.0.0.1:6650
  5. ## 分区生产者是同步/异步模式选择
  6. bridge.pulsar.produce = sync
  8. ## 生产者同步模式下的超时时间
  9. ## bridge.pulsar.produce.sync_timeout = 3s
  11. ## 生产者 batch 的消息数量
  12. ## bridge.pulsar.producer.batch_size = 1000
  14. ## 默认情况下不为生产者启用压缩选项
  15. ## bridge.pulsar.producer.compression = no_compression
  17. ## 采用 base64 编码或不编码
  18. ## bridge.pulsar.encode_payload_type = base64
  20. ## bridge.pulsar.sock.buffer = 32KB
  21. ## bridge.pulsar.sock.recbuf = 32KB
  22. bridge.pulsar.sock.sndbuf = 1MB
  23. ## bridge.pulsar.sock.read_packets = 20

配置 Pulsar 桥接规则

  1. ## Bridge Pulsar Hooks
  2. ## ${topic}: the pulsar 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.pulsar.hook.client.connected.1     = {"topic": "client_connected"}
  8. ## Client Disconnected Record Hook
  9. bridge.pulsar.hook.client.disconnected.1  = {"topic": "client_disconnected"}
  11. ## Session Subscribed Record Hook
  12. bridge.pulsar.hook.session.subscribed.1   = {"filter": "#",  "topic": "session_subscribed"}
  14. ## Session Unsubscribed Record Hook
  15. bridge.pulsar.hook.session.unsubscribed.1 = {"filter": "#",  "topic": "session_unsubscribed"}
  17. ## Message Publish Record Hook
  18. bridge.pulsar.hook.message.publish.1      = {"filter": "#",  "topic": "message_publish"}
  20. ## Message Delivered Record Hook
  21. bridge.pulsar.hook.message.delivered.1    = {"filter": "#",  "topic": "message_delivered"}
  23. ## Message Acked Record Hook
  24. bridge.pulsar.hook.message.acked.1        = {"filter": "#",  "topic": "message_acked"}
  26. ## More Configures
  27. ## partitioner strategy:
  28. ## Option:  random | roundrobin | first_key_dispatch
  29. ## Example: bridge.pulsar.hook.message.publish.1 = {"filter":"#", "topic":"message_publish", "strategy":"random"}
  31. ## key:
  32. ## Option: ${clientid} | ${username}
  33. ## Example: bridge.pulsar.hook.message.publish.1 = {"filter":"#", "topic":"message_publish", "key":"${clientid}"}
  35. ## format:
  36. ## Option: json | json
  37. ## Example: bridge.pulsar.hook.message.publish.1 = {"filter":"#", "topic":"message_publish", "format":"json"}

Pulsar 桥接规则说明

事件说明
bridge.pulsar.hook.client.connected.1客户端登录
bridge.pulsar.hook.client.disconnected.1客户端退出
bridge.pulsar.hook.session.subscribed.1订阅主题
bridge.pulsar.hook.session.unsubscribed.1取消订阅主题
bridge.pulsar.hook.message.publish.1发布消息
bridge.pulsar.hook.message.delivered.1delivered 消息
bridge.pulsar.hook.message.acked.1ACK 消息

客户端上下线事件转发 Pulsar

设备上线 EMQ X 转发上线事件消息到 Pulsar:

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

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

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

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

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

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

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

MQTT 消息转发到 Pulsar

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

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

  1. topic = message_delivered
  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) 事件转发 Pulsar

  1. topic = message_acked
  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.         }

Pulsar 消费示例

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

  1. sh pulsar-client consume client_connected  -s "client_connected" -n 1000
  3. sh pulsar-client consume client_disconnected  -s "client_disconnected" -n 1000

Pulsar 读取 MQTT 主题订阅事件消息:

  1. sh pulsar-client consume session_subscribed  -s "session_subscribed" -n 1000
  3. sh pulsar-client consume session_unsubscribed  -s "session_unsubscribed" -n 1000

Pulsar 读取 MQTT 发布消息:

  1. sh pulsar-client consume message_publish  -s "message_publish" -n 1000

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

  1. sh pulsar-client consume message_delivered  -s "message_delivered" -n 1000
  3. sh pulsar-client consume message_acked  -s "message_acked" -n 1000

Tip

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

启用 Pulsar 桥接插件

  1. ./bin/emqx_ctl plugins load emqx_bridge_pulsar