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

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

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

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

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

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

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

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

1. Kafka連接（Kafka Connect）MQTT

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

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

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

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

2. MQTT代理

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

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

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

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

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

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

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

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

這種方法的主要挑戰(zhàn)是，自定義應(yīng)用程序通常沒有設(shè)計成容錯和彈性。如果物聯(lián)網(wǎng)解決方案要求和端到端保證至少一次或確切一次消息傳遞，這就變得很重要。例如，設(shè)置為服務(wù)質(zhì)量級別1或2的MQTT消息發(fā)送到自定義應(yīng)用程序?qū)⒋_認收到消息。但是，如果自定義應(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代理擴展

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

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

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

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

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

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

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

我們很高興能將這個新產(chǎn)品帶給我們的HiveMQ客戶。這是在物聯(lián)網(wǎng)用例中使用Apache Kafka的最佳方法。
? ? ? ?責(zé)任編輯:pj