本文主要是介绍EMQX之Kafka插件编译安装，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
Kafka插件github上有好几个：
https://github.com/bob403/emqx_kafka_bridge （我们选择这个，支持emqx-v3.0版本）
https://github.com/iotblue/emqx-kafka-bridge/tree/emqx30
https://github.com/shsjdcbg/emq-kafka
修改 emqx-rel目录下的Makefile：
DEPS += $(foreachdep,$(MAIN_APPS),$(call app_name,$(dep))) DEPS += emqx_kafka_bridge 【增加这行】 ... $(foreach dep,$(MAIN_APPS),$(evaldep_$(call app_name,$(dep)) = $(CLONE_METHOD) https://github.com/emqx/$(dep)$(call app_vsn,$(dep)))) dep_emqx_kafka_bridge = git https://github.com/bob403/emqx_kafka_bridge.git master 【增加这行】
在relx.config里增加：
{emqx_kafka_bridge, load}
重新编译emqx：
cd emqx-relx rm -rf _rel //删除之前编译的结果 make clean make
修改kafka插件配置：
/emq/emqx-rel/_rel/emqx/etc/plugins/emqx_kafka_bridge.conf
【注意】这个插件的配置文件在好几个地方都存在，我们需要找到编译后的【_rel】目录下的配置，否则不生效。
根据实际情况调整配置：
##-------------------------------------------------------------------- ## kafka Bridge ##--------------------------------------------------------------------## The Kafka loadbalancer node hostthat bridge is listening on. ## ## Value: 127.0.0.1, localhost kafka.host = localhost## The kafka loadbalancer node portthat bridge is listening on. ## ## Value: Port kafka.port = 9092## The kafka loadbalancer nodepartition strategy. ## ## Value: random, sticky_round_robin,strict_round_robin, custom kafka.partitionstrategy =strict_round_robin## Each worker represents aconnection to a broker + topic + partition combination. ## You can decide how many workers tostart for each partition. ## ## Value: kafka.partitionworkers = 8## payload topic. ## kafka上要提前建立一个名为Processing的Topic ## Value: string kafka.payloadtopic = Processing
启用插件
列出所有插件：
./emqx_ctl plugins list
看到：
Plugin(emqx_kafka_bridge,version=3.0, description=EMQ X Kafka Bridge, active=false)
启动kafka插件：
./emqx_ctl plugins load emqx_kafka_bridge
返回：
Start apps: [emqx_kafka_bridge] Plugin emqx_kafka_bridge loadedsuccessfully.
测试步骤MQTT消息是否转发到kafka：
1，kafka上要提前建立一个名为Processing的Topic（分区为3，方便多个客户端消费）
./bin/kafka-topics.sh --create--zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic Processing
2，打开MQTT客户端，我使用的是chrome浏览器应用MQTTBox：
随便发布消息到一个MQTT主题，比如test，这里的MQTT 的topic和kafka的topic毫无关系，不要搞混了。
3，Kafka消费终端，消费主题Processing：
./bin/kafka-console-consumer.sh--bootstrap-server localhost:9092 --topic Processing
收到反馈如下：
{"action":"connected","device_id":"3fbf56e4-3396-47b0-a99f-5ecf0964e23b1561539149561","username":"undefined","ts":1561539149}
{"action":"message_publish","device_id":"3fbf56e4-3396-47b0-a99f-5ecf0964e23b1561539149561","username":"undefined","topic":"test","payload":"a1","ts":1561539155}
{"action":"message_publish","device_id":"3fbf56e4-3396-47b0-a99f-5ecf0964e23b1561539149561","username":"undefined","topic":"test","payload":"a2","ts":1561539157}
{"action":"disconnected","device_id":"3fbf56e4-3396-47b0-a99f-5ecf0964e23b1561537211825","username":"undefined","ts":1561539143}
可以收到MQTT转发的事件，比如connected、disconnected、message_publish等。
剩下的事我们就可以利用PHP、java等客户端去消费Processing主题，将需要的信息存入MYSQL等数据库了。
这篇关于EMQX之Kafka插件编译安装的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
