Kafka 桥接

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

image

Kafka 桥接插件配置文件: etc/plugins/emqx_bridge_kafka.conf。

配置 Kafka 集群地址

  1. ## Kafka 服务器地址
  2. ## bridge.kafka.servers = 127.0.0.1:9092,127.0.0.2:9092,127.0.0.3:9092
  3. bridge.kafka.servers = 127.0.0.1:9092
  5. ## Kafka 分区策略。可选值: per_partition | per_broker
  6. bridge.kafka.connection_strategy = per_partition
  8. bridge.kafka.min_metadata_refresh_interval = 5S
  10. ## Produce 写类型。可选值: sync | async
  11. bridge.kafka.produce = sync
  13. bridge.kafka.produce.sync_timeout = 3S
  15. ## 指定 replayq 在磁盘上存储消息的基本目录。
  16. ## 如果该配置项缺失活着设置为 undefined, replayq 将以使用内存的
  17. ## 的方式工作。也就是说，消息不在磁盘上排队 -- 在这种情况下，send
  18. ## 和 send_async API 的调用者负责处理在应用程序、网络或 kafka
  19. ## 干扰时可能丢失的消息。
  20. ## bridge.kafka.replayq_dir = /tmp/emqx_bridge_kafka/
  22. ## default=10MB, replayq 分段大小。
  23. ## bridge.kafka.producer.replayq_seg_bytes = 10MB
  25. ## producer required_acks. 可选值: all_isr | leader_only | none.
  26. bridge.kafka.producer.required_acks = none
  28. ## default=10000. leader 在回复 producer 前等待副本的超时时间。
  29. bridge.kafka.producer.ack_timeout = 10S
  31. ## 收集到一次 produce 请求中的最大字节数
  32. bridge.kafka.producer.max_batch_bytes = 1024KB
  34. ## 收集到一次 produce 请求中的最少字节数
  35. bridge.kafka.producer.min_batch_bytes = 0
  37. ## 在没有接收到上次请求的 ack 的情况下，可以提前发送的 batch 数。
  38. ## 如果消息必须严格按照顺序传递，则必须为0。
  39. bridge.kafka.producer.max_send_ahead = 0
  41. ## 默认为无压缩
  42. ## bridge.kafka.producer.compression = no_compression
  44. ## 默认值为 base64, 可选值: base64 | plain
  45. ## bridge.kafka.encode_payload_type = base64
  47. ## bridge.kafka.sock.buffer = 32KB
  48. ## bridge.kafka.sock.recbuf = 32KB
  49. bridge.kafka.sock.sndbuf = 1MB
  50. ## bridge.kafka.sock.read_packets = 20

配置 Kafka 桥接规则

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

Kafka 桥接规则说明

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

客户端上下线事件转发 Kafka

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

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

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

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

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

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

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

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

MQTT 消息转发到 Kafka

  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) 事件转发 Kafka

  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) 事件转发 Kafka

  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.         }

Kafka 消费示例

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

  1. sh kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic client_connected --from-beginning
  3. sh kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic client_disconnected --from-beginning

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

  1. sh kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic session_subscribed --from-beginning
  3. sh kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic session_unsubscribed --from-beginning

Kafka 读取 MQTT 发布消息:

  1. sh kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic message_publish --from-beginning

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

  1. sh kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic message_delivered --from-beginning
  3. sh kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic message_acked --from-beginning

Tip

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

启用 Kafka 桥接插件

  1. ./bin/emqx_ctl plugins load emqx_bridge_kafka