如何將物聯(lián)網(wǎng)數(shù)據(jù)從設(shè)備連接到Kafka集群？

Apache Kafka是一個(gè)實(shí)時(shí)流平臺(tái)，在大大小小的組織中得到了廣泛的采用。Kafka的分布式微服務(wù)架構(gòu)和發(fā)布/訂閱協(xié)議使得它非常適合在企業(yè)系統(tǒng)和應(yīng)用程序之間實(shí)時(shí)移動(dòng)數(shù)據(jù)。據(jù)統(tǒng)計(jì)，超過三分之一的財(cái)富500強(qiáng)公司正在使用Kafka。在Github上，Kafka是最受歡迎的Apache項(xiàng)目之一，有超過11K之星和超過500名貢獻(xiàn)者。毫無疑問，Kafka是一個(gè)開源項(xiàng)目，它改變了企業(yè)在云和數(shù)據(jù)中心內(nèi)移動(dòng)數(shù)據(jù)的方式。

Kafka的架構(gòu)已經(jīng)被優(yōu)化為在系統(tǒng)和應(yīng)用程序之間以可伸縮的方式盡可能快地流數(shù)據(jù)。Kafka客戶端/生產(chǎn)者與Kafka集群緊密耦合，要求每個(gè)客戶端知道Kafka集群的IP地址，并直接訪問所有單獨(dú)的節(jié)點(diǎn)。在可信網(wǎng)絡(luò)內(nèi)部，這允許對(duì)代理拓?fù)溥M(jìn)行更改，這意味著可以通過直接使用來自Kafka客戶端的多個(gè)節(jié)點(diǎn)來擴(kuò)展主題和分區(qū)。在大多數(shù)情況下，Kafka主題空間也保持相當(dāng)扁平，因?yàn)橥ǔＪ褂枚鄠€(gè)分區(qū)來擴(kuò)展單個(gè)Kafka主題。在Kafka系統(tǒng)中擁有數(shù)百甚至數(shù)千個(gè)主題通常是不可取的，首選的方法是對(duì)大多數(shù)數(shù)據(jù)流使用很少的主題。Kafka非常適合在具有穩(wěn)定IP地址和連接的相同可信網(wǎng)絡(luò)內(nèi)的系統(tǒng)之間進(jìn)行通信。

對(duì)于設(shè)備連接到公共互聯(lián)網(wǎng)上的數(shù)據(jù)中心或云的物聯(lián)網(wǎng)用例，Kafka架構(gòu)不適合開箱即用。如果您試圖在Internet上使用Kafka從數(shù)千甚至數(shù)百萬設(shè)備流數(shù)據(jù)，那么Apache Kafka架構(gòu)是不合適的。有許多原因Kafka本身不是很適合物聯(lián)網(wǎng)用例：

Kafka代理需要由客戶端直接處理，這意味著每個(gè)客戶端需要能夠直接連接到Kafka代理。專業(yè)的IoT部署通常利用負(fù)載均衡器作為云中的第一道防線，因此設(shè)備只使用負(fù)載均衡器的IP地址連接到基礎(chǔ)設(shè)施，負(fù)載均衡器有效地充當(dāng)代理。如果您希望您的設(shè)備直接連接到Kafka，您的Kafka代理必須暴露給公共互聯(lián)網(wǎng)。Kafka不支持大量的主題。當(dāng)通過公共Internet連接數(shù)以百萬計(jì)的物聯(lián)網(wǎng)設(shè)備時(shí)，通常使用單個(gè)和唯一的主題（通常在主題名稱中包含一些唯一的物聯(lián)網(wǎng)設(shè)備標(biāo)識(shí)符），因此可以根據(jù)單個(gè)客戶機(jī)的權(quán)限限制讀寫操作。你不希望你的智能恒溫器被黑客攻擊，那些證書可能被用來竊聽系統(tǒng)中的所有數(shù)據(jù)流。與物聯(lián)網(wǎng)協(xié)議的客戶端庫(kù)相比，Kafka客戶端是相當(dāng)復(fù)雜和資源密集型的。大多數(shù)編程語言的Kafka api都非常直接和簡(jiǎn)單，但是在其內(nèi)部存在很多復(fù)雜性。例如，客戶端將使用并維護(hù)到Kafka代理的多個(gè)TCP連接。物聯(lián)網(wǎng)的部署通常會(huì)限制設(shè)備的使用，這些設(shè)備需要最小的內(nèi)存占用，并且在設(shè)備端不需要很高的吞吐量。默認(rèn)情況下，Kafka客戶端針對(duì)吞吐量進(jìn)行了優(yōu)化。Kafka客戶端需要一個(gè)穩(wěn)定的TCP連接來獲得最好的結(jié)果。許多物聯(lián)網(wǎng)使用案例涉及不可靠的網(wǎng)絡(luò)，例如聯(lián)網(wǎng)的汽車或智能農(nóng)業(yè)，因此典型的物聯(lián)網(wǎng)設(shè)備需要不斷地重新建立與Kafka的連接。將數(shù)萬甚至數(shù)百萬客戶機(jī)連接到一個(gè)Kafka集群是不尋常的（通常甚至根本不可能）。在物聯(lián)網(wǎng)使用案例中，通常有大量設(shè)備同時(shí)連接到后端，并不斷產(chǎn)生數(shù)據(jù)。Kafka缺少一些關(guān)鍵的物聯(lián)網(wǎng)功能。Kafka協(xié)議缺少了諸如《活著》和《遺囑》等功能。這些特性對(duì)于構(gòu)建一個(gè)有彈性的物聯(lián)網(wǎng)解決方案非常重要，該解決方案可以處理遇到意外連接丟失和網(wǎng)絡(luò)不可靠的設(shè)備。

Kafka仍然為物聯(lián)網(wǎng)用例帶來了很多價(jià)值。物聯(lián)網(wǎng)解決方案創(chuàng)建了大量的實(shí)時(shí)數(shù)據(jù)，很適合由Kafka處理。挑戰(zhàn)在于：如何將物聯(lián)網(wǎng)數(shù)據(jù)從設(shè)備連接到Kafka集群？

許多實(shí)現(xiàn)物聯(lián)網(wǎng)用例的公司正在研究集成MQTT和Kafka來處理物聯(lián)網(wǎng)數(shù)據(jù)的可能性。MQTT是另一個(gè)發(fā)布/訂閱協(xié)議，它已經(jīng)成為連接IoT設(shè)備數(shù)據(jù)的標(biāo)準(zhǔn)。MQTT標(biāo)準(zhǔn)設(shè)計(jì)用于通過不可靠的網(wǎng)絡(luò)連接大量的物聯(lián)網(wǎng)設(shè)備，解決了Kafka的許多限制。具體來說，MQTT是一種輕量級(jí)協(xié)議，需要在每個(gè)設(shè)備上占用較小的客戶機(jī)空間。它可以在不可靠的網(wǎng)絡(luò)上安全地支持?jǐn)?shù)百萬個(gè)連接，并在高延遲和低吞吐量環(huán)境下無縫工作。它包括物聯(lián)網(wǎng)功能，如保持活力，最后的遺囑和遺囑功能，不同質(zhì)量的服務(wù)水平的可靠消息，以及客戶端負(fù)載平衡（共享訂閱）在其他功能設(shè)計(jì)的公共互聯(lián)網(wǎng)通信。主題是動(dòng)態(tài)的，這意味著系統(tǒng)中可以存在任意數(shù)量的MQTT主題，在MQTT服務(wù)器集群中，每次部署通常多達(dá)數(shù)千萬個(gè)主題。

盡管Kafka和MQTT有不同的設(shè)計(jì)目標(biāo)，但兩者在一起工作得非常好。問題不是Kafka vs MQTT，而是如何將這兩個(gè)世界集成在一起，形成一個(gè)物聯(lián)網(wǎng)端到端數(shù)據(jù)管道。為了將MQTT消息集成到Kafka集群中，需要某種類型的橋接器將MQTT消息轉(zhuǎn)發(fā)到Kafka。

有四種不同的架構(gòu)方法來實(shí)現(xiàn)這種類型的橋梁：

1. Kafka連接（Kafka Connect）MQTT

Kafka有一個(gè)擴(kuò)展框架，叫做Kafka Connect，它允許Kafka從其他系統(tǒng)攝取數(shù)據(jù)。Kafka Connect for MQTT充當(dāng)一個(gè)MQTT客戶端，訂閱來自MQTT代理的所有消息。

如果您沒有對(duì)MQTT代理的控制權(quán)，那么Kafka Connect for MQTT是一個(gè)值得追求的方法。這種方法允許Kafka攝取MQTT消息流。

在MQTT中使用Kafka Connect存在性能和可伸縮性限制。如前所述，Kafka Connect for MQTT是一個(gè)MQTT客戶機(jī)，它訂閱通過代理傳遞的所有MQTT消息。MQTT客戶機(jī)庫(kù)并不打算處理大量的MQTT消息，因此使用這種方法的物聯(lián)網(wǎng)系統(tǒng)將存在性能和可伸縮性問題。

這種方法集中了業(yè)務(wù)和消息轉(zhuǎn)換邏輯，并創(chuàng)建了緊密耦合，這在分布式（微服務(wù)）體系結(jié)構(gòu)中應(yīng)該避免。業(yè)界領(lǐng)先的咨詢公司Thoughtworks稱這是一種反模式，甚至在他們之前的技術(shù)雷達(dá)出版物中將Kafka歸入“持有”類別。

2. MQTT代理

另一種方法是使用代理應(yīng)用程序，它接受來自物聯(lián)網(wǎng)設(shè)備的MQTT消息，但不實(shí)現(xiàn)發(fā)布/訂閱或任何MQTT會(huì)話特性，因此不是MQTT代理。物聯(lián)網(wǎng)設(shè)備連接到MQTT代理，然后該代理將MQTT消息推送到Kafka代理。

MQTT代理方法允許在Kafka部署中完成MQTT消息處理，因此可以從單個(gè)控制臺(tái)完成管理和操作。MQTT代理通常是無狀態(tài)的，因此通過添加代理的多個(gè)實(shí)例，它（理論上）可以獨(dú)立于Kafka集群進(jìn)行伸縮。

MQTT代理的限制是它不是真正的MQTT實(shí)現(xiàn)。MQTT代理不是基于發(fā)布/訂閱的。相反，它在設(shè)備和Kafka之間創(chuàng)建了一個(gè)緊密耦合的流。MQTT發(fā)布/訂閱的好處是，它創(chuàng)建了一個(gè)松散耦合的端點(diǎn)系統(tǒng)（設(shè)備或后端應(yīng)用程序），可以在每個(gè)端點(diǎn)之間通信和移動(dòng)數(shù)據(jù)。例如，MQTT允許兩個(gè)設(shè)備之間的通信，例如兩個(gè)連接的汽車可以彼此通信，但是MQTT代理應(yīng)用程序只允許從一輛汽車到Kafka集群的數(shù)據(jù)傳輸，而不允許與另一輛汽車的數(shù)據(jù)傳輸。

一些Kafka MQTT代理應(yīng)用程序支持QoS級(jí)別等特性。值得注意的是，只有在MQTT客戶端重新連接到相同的MQTT代理實(shí)例時(shí)，才可能在連接丟失后恢復(fù)QoS消息流，而這是不可能的，前提是使用負(fù)載均衡器，該均衡器使用最少連接或循環(huán)策略來實(shí)現(xiàn)可伸縮性。因此，在MQTT中使用QoS級(jí)別的主要原因（即沒有消息丟失）僅適用于穩(wěn)定連接，這在大多數(shù)物聯(lián)網(wǎng)場(chǎng)景中是一個(gè)不現(xiàn)實(shí)的假設(shè)。

使用這種方法的主要風(fēng)險(xiǎn)是代理不是功能完整的MQTT代理，因此它不是MQTT規(guī)范定義的MQTT實(shí)現(xiàn)，只是實(shí)現(xiàn)了一個(gè)很小的子集，因此它不是一個(gè)標(biāo)準(zhǔn)化的解決方案。為了在MQTT客戶機(jī)中正確地使用MQTT，需要一個(gè)功能齊全的MQTT代理。

如果消息丟失不是一個(gè)重要因素，并且沒有使用為可靠的物聯(lián)網(wǎng)通信而設(shè)計(jì)的MQTT特性，如果您只想通過Internet單向地向Kafka發(fā)送數(shù)據(jù)，那么代理方法可能是一個(gè)輕量級(jí)的替代方法。

3.構(gòu)建您自己的自定義橋接

一些公司建立了他們自己的MQTT到Kafka橋。典型的方法是使用開源MQTT客戶端庫(kù)和開源Kafka客戶端庫(kù)創(chuàng)建應(yīng)用程序。自定義應(yīng)用程序負(fù)責(zé)在MQTT代理和Kafka實(shí)例之間調(diào)換和路由數(shù)據(jù)。

這種方法的主要挑戰(zhàn)是，自定義應(yīng)用程序通常沒有設(shè)計(jì)成容錯(cuò)和彈性。如果物聯(lián)網(wǎng)解決方案要求和端到端保證至少一次或確切一次消息傳遞，這就變得很重要。例如，設(shè)置為服務(wù)質(zhì)量級(jí)別1或2的MQTT消息發(fā)送到自定義應(yīng)用程序?qū)⒋_認(rèn)收到消息。但是，如果自定義應(yīng)用程序在將消息轉(zhuǎn)發(fā)給Kafka之前崩潰，則消息將丟失。類似地，如果Kafka集群不可用，自定義應(yīng)用程序?qū)⑿枰彌_MQTT消息。如果定制應(yīng)用程序在Kafka集群恢復(fù)可用之前崩潰，所有緩沖的消息將丟失。要解決這些問題，定制應(yīng)用程序?qū)⑿枰罅康拈_發(fā)工作，構(gòu)建與Kafka和MQTT代理中已經(jīng)發(fā)現(xiàn)的技術(shù)類似的功能。

4. MQTT代理擴(kuò)展

最后一種方法是擴(kuò)展MQTT代理，以創(chuàng)建包含本機(jī)Kafka協(xié)議的擴(kuò)展。這允許MQTT代理充當(dāng)一流的Kafka客戶機(jī)，并將物聯(lián)網(wǎng)設(shè)備數(shù)據(jù)流傳遞給多個(gè)Kafka集群。

要實(shí)現(xiàn)這種方法，您需要訪問MQTT代理，代理需要能夠安裝擴(kuò)展。

這種方法允許物聯(lián)網(wǎng)解決方案使用本地MQTT實(shí)現(xiàn)和本地Kafka實(shí)現(xiàn)。物聯(lián)網(wǎng)設(shè)備使用MQTT客戶機(jī)將數(shù)據(jù)發(fā)送到功能齊全的MQTT代理。MQTT代理被擴(kuò)展為包括一個(gè)本地Kafka客戶機(jī)，并將MQTT消息置換到Kafka協(xié)議。這使得物聯(lián)網(wǎng)數(shù)據(jù)可以同時(shí)路由到多個(gè)Kafka集群和非Kafka應(yīng)用程序。使用MQTT代理還將提供對(duì)物聯(lián)網(wǎng)設(shè)備所需的所有MQTT特性的訪問，例如遺囑和遺囑。MQTT代理（如HiveMQ）是為高可用性、持久性、性能和彈性而設(shè)計(jì)的，因此消息可以在Kafka不可寫時(shí)在代理上緩沖，因此重要消息不會(huì)從物聯(lián)網(wǎng)設(shè)備中丟失。因此，這種方法提供了真正的端到端消息傳遞保證，即使是在不可靠的網(wǎng)絡(luò)、公共Internet通信和不斷變化的網(wǎng)絡(luò)拓?fù)洌ㄔ谌萜骰渴鹬薪?jīng)常看到，例如Kubernetes）。

用于Kafka的HiveMQ企業(yè)擴(kuò)展

在與HiveMQ客戶的對(duì)話中，一些操作集群具有數(shù)百萬臺(tái)設(shè)備和非常高的消息吞吐量，我們看到需要為Kafka創(chuàng)建MQTT代理擴(kuò)展。我們的客戶希望從MQTT和Kafka協(xié)議的本地實(shí)現(xiàn)中受益，因?yàn)檫@兩個(gè)協(xié)議都有所有的交付保證。因此，我們很高興地宣布Kafka的HiveMQ企業(yè)擴(kuò)展。

我們的客戶看到了MQTT和Kafka聯(lián)合解決方案的巨大價(jià)值。他們將Kafka視為在數(shù)據(jù)中心或云環(huán)境中處理和分發(fā)實(shí)時(shí)數(shù)據(jù)的優(yōu)秀平臺(tái)。他們希望使用MQTT和HiveMQ將數(shù)據(jù)從設(shè)備移動(dòng)到不同的后端系統(tǒng)。后端系統(tǒng)包括Kafka和非Kafka系統(tǒng)。他們還知道，如果他們?cè)噲D連接數(shù)以百萬計(jì)的設(shè)備，比如連接的汽車，他們需要使用MQTT的本機(jī)和經(jīng)過實(shí)戰(zhàn)測(cè)試的實(shí)現(xiàn)，比如HiveMQ。

用于Kafka的HiveMQ企業(yè)擴(kuò)展在HiveMQ代理中實(shí)現(xiàn)了本地Kafka協(xié)議。這允許MQTT消息與單個(gè)Kafka集群或多個(gè)Kafka集群同時(shí)無縫集成。它100%支持整個(gè)MQTT 3和MQTT 5規(guī)范。我們甚至可以將潛在的數(shù)百萬個(gè)MQTT主題映射到數(shù)量有限的Kafka主題。最后，我們擴(kuò)展了HiveMQ控制中心，使其能夠監(jiān)視寫入Kafka的MQTT消息。
? ? ? ?責(zé)任編輯:pj