使用mqtt协议和kafka作为消息代理

Question

如何通过kafka作为消息代理来使用mqtt协议？

客户端(android/ios/桌面、java应用程序等)将使用mqtt phao客户端库来生成和消费消息，这些库有不同语言的版本，使用kafka作为消息代理。

有什么建议吗？

Answer 1

您可以使用Kafka源连接器，它会将数据从MQTT代理(如Mosquitto )流式传输到Kafka集群。请参阅https://github.com/evokly/kafka-connect-mqtt

运行连接器的最简单方法是在独立模式下，在该模式下，单个实例将在单个节点上的Kafka集群上运行。您也可以在分布式模式下运行它(尽管配置要多得多)，这将在整个集群中分布连接器，以获得更大的吞吐量。在分布式模式下，您可以设计一种支持水平扩展、并行吞吐量和高可用性的拓扑。实现额外的保证需要额外的负载均衡器，多个MQTT代理和最后的遗嘱和遗嘱场景来处理连接器崩溃，但这可能超出了这个问题的范围。

使用连接器方法具有Kafka集群的优势，即确保连接器处于活动状态，并在必要时重新启动它。分布式模式提供了更多的优势。

Answer 2

这不是一个好主意。MQTT客户端通常是非常轻量级的，资源有限。这些设备或IoT的内存/CPU功率很小。Kafka客户端通常是重量级的。例如，Kafka客户端必须跟踪偏移量。它还需要与动物园管理员进行交互。简而言之，Kafka不适合作为MQTT代理。您最好选择流行的MQTT代理，比如Mosquito。

Answer 3

AFAIK，没有针对Kafka的“官方”MQTT连接器。但是你可以使用MQTT-Kafka桥。要获得灵感，请查看https://github.com/km4rcus/mqttKafkaBridge (请注意，此实现中存在一个错误: kafka主题不能包含"/“，因此您可能希望在Bridge.java文件的messageArrived中将它们替换为"_”)

请注意，此代码只是一个非常简单的解决方案，不可伸缩。编写您的自定义实现以更好地满足您的期望可能是一个好主意。但是你应该让它尽可能的简单--这是一个单点故障。只要你把你的数据放到Kafka中，你就会得到一些保证，但你不会从MQTT代理那里得到任何保证。当桥崩溃时，你只是丢失了你的数据...