如何将物联网数据从设备连接到Kafka集群？

Apache Kafka是一个实时流平台，在大大小小的组织中得到了广泛的采用。Kafka的分布式微服务架构和发布/订阅协议使得它非常适合在企业系统和应用程序之间实时移动数据。据统计，超过三分之一的财富500强公司正在使用Kafka。在Github上，Kafka是最受欢迎的Apache项目之一，有超过11K之星和超过500名贡献者。毫无疑问，Kafka是一个开源项目，它改变了企业在云和数据中心内移动数据的方式。

Kafka的架构已经被优化为在系统和应用程序之间以可伸缩的方式尽可能快地流数据。Kafka客户端/生产者与Kafka集群紧密耦合，要求每个客户端知道Kafka集群的IP地址，并直接访问所有单独的节点。在可信网络内部，这允许对代理拓扑进行更改，这意味着可以通过直接使用来自Kafka客户端的多个节点来扩展主题和分区。在大多数情况下，Kafka主题空间也保持相当扁平，因为通常使用多个分区来扩展单个Kafka主题。在Kafka系统中拥有数百甚至数千个主题通常是不可取的，首选的方法是对大多数数据流使用很少的主题。Kafka非常适合在具有稳定IP地址和连接的相同可信网络内的系统之间进行通信。

对于设备连接到公共互联网上的数据中心或云的物联网用例，Kafka架构不适合开箱即用。如果您试图在Internet上使用Kafka从数千甚至数百万设备流数据，那么Apache Kafka架构是不合适的。有许多原因Kafka本身不是很适合物联网用例：

Kafka代理需要由客户端直接处理，这意味着每个客户端需要能够直接连接到Kafka代理。专业的IoT部署通常利用负载均衡器作为云中的第一道防线，因此设备只使用负载均衡器的IP地址连接到基础设施，负载均衡器有效地充当代理。如果您希望您的设备直接连接到Kafka，您的Kafka代理必须暴露给公共互联网。Kafka不支持大量的主题。当通过公共Internet连接数以百万计的物联网设备时，通常使用单个和唯一的主题（通常在主题名称中包含一些唯一的物联网设备标识符），因此可以根据单个客户机的权限限制读写操作。你不希望你的智能恒温器被黑客攻击，那些证书可能被用来窃听系统中的所有数据流。与物联网协议的客户端库相比，Kafka客户端是相当复杂和资源密集型的。大多数编程语言的Kafka api都非常直接和简单，但是在其内部存在很多复杂性。例如，客户端将使用并维护到Kafka代理的多个TCP连接。物联网的部署通常会限制设备的使用，这些设备需要最小的内存占用，并且在设备端不需要很高的吞吐量。默认情况下，Kafka客户端针对吞吐量进行了优化。Kafka客户端需要一个稳定的TCP连接来获得最好的结果。许多物联网使用案例涉及不可靠的网络，例如联网的汽车或智能农业，因此典型的物联网设备需要不断地重新建立与Kafka的连接。将数万甚至数百万客户机连接到一个Kafka集群是不寻常的（通常甚至根本不可能）。在物联网使用案例中，通常有大量设备同时连接到后端，并不断产生数据。Kafka缺少一些关键的物联网功能。Kafka协议缺少了诸如《活着》和《遗嘱》等功能。这些特性对于构建一个有弹性的物联网解决方案非常重要，该解决方案可以处理遇到意外连接丢失和网络不可靠的设备。

Kafka仍然为物联网用例带来了很多价值。物联网解决方案创建了大量的实时数据，很适合由Kafka处理。挑战在于：如何将物联网数据从设备连接到Kafka集群？

许多实现物联网用例的公司正在研究集成MQTT和Kafka来处理物联网数据的可能性。MQTT是另一个发布/订阅协议，它已经成为连接IoT设备数据的标准。MQTT标准设计用于通过不可靠的网络连接大量的物联网设备，解决了Kafka的许多限制。具体来说，MQTT是一种轻量级协议，需要在每个设备上占用较小的客户机空间。它可以在不可靠的网络上安全地支持数百万个连接，并在高延迟和低吞吐量环境下无缝工作。它包括物联网功能，如保持活力，最后的遗嘱和遗嘱功能，不同质量的服务水平的可靠消息，以及客户端负载平衡（共享订阅）在其他功能设计的公共互联网通信。主题是动态的，这意味着系统中可以存在任意数量的MQTT主题，在MQTT服务器集群中，每次部署通常多达数千万个主题。

尽管Kafka和MQTT有不同的设计目标，但两者在一起工作得非常好。问题不是Kafka vs MQTT，而是如何将这两个世界集成在一起，形成一个物联网端到端数据管道。为了将MQTT消息集成到Kafka集群中，需要某种类型的桥接器将MQTT消息转发到Kafka。

有四种不同的架构方法来实现这种类型的桥梁：

1. Kafka连接（Kafka Connect）MQTT

Kafka有一个扩展框架，叫做Kafka Connect，它允许Kafka从其他系统摄取数据。Kafka Connect for MQTT充当一个MQTT客户端，订阅来自MQTT代理的所有消息。

如果您没有对MQTT代理的控制权，那么Kafka Connect for MQTT是一个值得追求的方法。这种方法允许Kafka摄取MQTT消息流。

在MQTT中使用Kafka Connect存在性能和可伸缩性限制。如前所述，Kafka Connect for MQTT是一个MQTT客户机，它订阅通过代理传递的所有MQTT消息。MQTT客户机库并不打算处理大量的MQTT消息，因此使用这种方法的物联网系统将存在性能和可伸缩性问题。

这种方法集中了业务和消息转换逻辑，并创建了紧密耦合，这在分布式（微服务）体系结构中应该避免。业界领先的咨询公司Thoughtworks称这是一种反模式，甚至在他们之前的技术雷达出版物中将Kafka归入“持有”类别。

2. MQTT代理

另一种方法是使用代理应用程序，它接受来自物联网设备的MQTT消息，但不实现发布/订阅或任何MQTT会话特性，因此不是MQTT代理。物联网设备连接到MQTT代理，然后该代理将MQTT消息推送到Kafka代理。

MQTT代理方法允许在Kafka部署中完成MQTT消息处理，因此可以从单个控制台完成管理和操作。MQTT代理通常是无状态的，因此通过添加代理的多个实例，它（理论上）可以独立于Kafka集群进行伸缩。

MQTT代理的限制是它不是真正的MQTT实现。MQTT代理不是基于发布/订阅的。相反，它在设备和Kafka之间创建了一个紧密耦合的流。MQTT发布/订阅的好处是，它创建了一个松散耦合的端点系统（设备或后端应用程序），可以在每个端点之间通信和移动数据。例如，MQTT允许两个设备之间的通信，例如两个连接的汽车可以彼此通信，但是MQTT代理应用程序只允许从一辆汽车到Kafka集群的数据传输，而不允许与另一辆汽车的数据传输。

一些Kafka MQTT代理应用程序支持QoS级别等特性。值得注意的是，只有在MQTT客户端重新连接到相同的MQTT代理实例时，才可能在连接丢失后恢复QoS消息流，而这是不可能的，前提是使用负载均衡器，该均衡器使用最少连接或循环策略来实现可伸缩性。因此，在MQTT中使用QoS级别的主要原因（即没有消息丢失）仅适用于稳定连接，这在大多数物联网场景中是一个不现实的假设。

使用这种方法的主要风险是代理不是功能完整的MQTT代理，因此它不是MQTT规范定义的MQTT实现，只是实现了一个很小的子集，因此它不是一个标准化的解决方案。为了在MQTT客户机中正确地使用MQTT，需要一个功能齐全的MQTT代理。

如果消息丢失不是一个重要因素，并且没有使用为可靠的物联网通信而设计的MQTT特性，如果您只想通过Internet单向地向Kafka发送数据，那么代理方法可能是一个轻量级的替代方法。

3.构建您自己的自定义桥接

一些公司建立了他们自己的MQTT到Kafka桥。典型的方法是使用开源MQTT客户端库和开源Kafka客户端库创建应用程序。自定义应用程序负责在MQTT代理和Kafka实例之间调换和路由数据。

这种方法的主要挑战是，自定义应用程序通常没有设计成容错和弹性。如果物联网解决方案要求和端到端保证至少一次或确切一次消息传递，这就变得很重要。例如，设置为服务质量级别1或2的MQTT消息发送到自定义应用程序将确认收到消息。但是，如果自定义应用程序在将消息转发给Kafka之前崩溃，则消息将丢失。类似地，如果Kafka集群不可用，自定义应用程序将需要缓冲MQTT消息。如果定制应用程序在Kafka集群恢复可用之前崩溃，所有缓冲的消息将丢失。要解决这些问题，定制应用程序将需要大量的开发工作，构建与Kafka和MQTT代理中已经发现的技术类似的功能。

4. MQTT代理扩展

最后一种方法是扩展MQTT代理，以创建包含本机Kafka协议的扩展。这允许MQTT代理充当一流的Kafka客户机，并将物联网设备数据流传递给多个Kafka集群。

要实现这种方法，您需要访问MQTT代理，代理需要能够安装扩展。

这种方法允许物联网解决方案使用本地MQTT实现和本地Kafka实现。物联网设备使用MQTT客户机将数据发送到功能齐全的MQTT代理。MQTT代理被扩展为包括一个本地Kafka客户机，并将MQTT消息置换到Kafka协议。这使得物联网数据可以同时路由到多个Kafka集群和非Kafka应用程序。使用MQTT代理还将提供对物联网设备所需的所有MQTT特性的访问，例如遗嘱和遗嘱。MQTT代理（如HiveMQ）是为高可用性、持久性、性能和弹性而设计的，因此消息可以在Kafka不可写时在代理上缓冲，因此重要消息不会从物联网设备中丢失。因此，这种方法提供了真正的端到端消息传递保证，即使是在不可靠的网络、公共Internet通信和不断变化的网络拓扑（在容器化部署中经常看到，例如Kubernetes）。

用于Kafka的HiveMQ企业扩展

在与HiveMQ客户的对话中，一些操作集群具有数百万台设备和非常高的消息吞吐量，我们看到需要为Kafka创建MQTT代理扩展。我们的客户希望从MQTT和Kafka协议的本地实现中受益，因为这两个协议都有所有的交付保证。因此，我们很高兴地宣布Kafka的HiveMQ企业扩展。

我们的客户看到了MQTT和Kafka联合解决方案的巨大价值。他们将Kafka视为在数据中心或云环境中处理和分发实时数据的优秀平台。他们希望使用MQTT和HiveMQ将数据从设备移动到不同的后端系统。后端系统包括Kafka和非Kafka系统。他们还知道，如果他们试图连接数以百万计的设备，比如连接的汽车，他们需要使用MQTT的本机和经过实战测试的实现，比如HiveMQ。

用于Kafka的HiveMQ企业扩展在HiveMQ代理中实现了本地Kafka协议。这允许MQTT消息与单个Kafka集群或多个Kafka集群同时无缝集成。它100%支持整个MQTT 3和MQTT 5规范。我们甚至可以将潜在的数百万个MQTT主题映射到数量有限的Kafka主题。最后，我们扩展了HiveMQ控制中心，使其能够监视写入Kafka的MQTT消息。
       责任编辑:pj