准备
EMQX - 导读
EMQX - 源码阅读(六)
emqx_plugin_kafka
emqx_plugin_kafka
Kafka plugin for EMQX >= V5.4.0
Usage
Release
> git clone https://github.com/jostar-y/emqx_plugin_kafka.git
> cd emqx_plugin_kafka
> make rel
_build/default/emqx_plugrel/emqx_plugin_kafka-<vsn>.tar.gz
Config
Explain
> cat priv/emqx_plugin_kafka.hocon
plugin_kafka {
// required
connection {
// Kafka client id: "emqx_plugin:kafka_client:${client_id}"
// optional default:"emqx_plugin:kafka_client:emqx_plugin_kafka_connection"
client_id = "kafka_client"
// Kafka address.
// required
bootstrap_hosts = ["10.3.64.223:9192", "10.3.64.223:9292", "10.3.64.223:9392"]
// Reference type: kpro_connection:config().
// https://github.com/kafka4beam/kafka_protocol/blob/master/src/kpro_connection.erl
// optional default:5s
connect_timeout = 5s
// enum: per_partition | per_broker
// optional default:per_partition
connection_strategy = per_partition
// optional default:5s
min_metadata_refresh_interval = 5s
// optional default:true
query_api_versions = true
// optional default:3s
request_timeout = 3s
sasl {
// enum: plain | scram_sha_256 | scram_sha_512
mechanism = plain
username = "username"
password = "password"
}
ssl {
enable = false
}
//Emqx resource opts.
// optional default:32s
health_check_interval = 32s
}
// optional
producer {
// Most number of bytes to collect into a produce request.
// optional default:896KB
max_batch_bytes = 896KB
// enum: no_compression | snappy | gzip
// optional default:no_compression
compression = no_compression
// enum: random | roundrobin | first_key_dispatch
// optional default:random
partition_strategy = random
// Encode kafka value.
// enum: plain | base64
// optional default:plain
encode_payload_type = plain
}
// required
hooks = [
{
// Hook point.
// required
endpoint = message.publish
// Emqx topic pattern.
// 1. Cannot match the system message;
// 2. Cannot use filters that start with '+' or '#'.
// message required
filter = "test/#"
// Kafka topic, must be created in advance in Kafka.
// required
kafka_topic = emqx_test
// Matching template, value = ${.} indicates that all keys match
// optional default:{timestamp = "${.timestamp}", value = "${.}",key = "${.clientid}"}
kafka_message = {
timestamp = "${.timestamp}"
value = "${.}"
key = "${.clientid}"
}
}
]
}
Some examples in the directory priv/example/
.
Hook Point
endpoint | filter |
---|---|
client.connect | / |
client.connack | / |
client.connected | / |
client.disconnected | / |
client.authenticate | / |
client.authorize | / |
client.authenticate | / |
client.check_authz_complete | / |
session.created | / |
session.subscribed | / |
session.unsubscribed | / |
session.resumed | / |
session.discarded | / |
session.takenover | / |
session.terminated | / |
message.publish | required |
message.delivered | required |
message.acked | required |
message.dropped | required |
Path
- Default path:
emqx/etc/emqx_plugin_kafka.hocon
- Attach to path: set system environment variables
export EMQX_PLUGIN_KAFKA_CONF="absolute_path"