一、什么是消息系統(tǒng)

消息系統(tǒng)負(fù)責(zé)將數(shù)據(jù)從一個應(yīng)用程序傳輸?shù)搅硪粋€應(yīng)用程序，應(yīng)用程序可以專注于數(shù)據(jù)，不擔(dān)心如何共享它。 分布式消息傳遞基于可靠消息隊列的概念。 消息在客戶端應(yīng)用程序和消息傳遞系統(tǒng)之間異步排隊。 有兩種類型的消息模式可用：一種是點對點，另一種是發(fā)布 - 訂閱(pub-sub)消息系統(tǒng)。 大多數(shù)消息模式遵循 pub-sub 。

1.1 點對點消息系統(tǒng)

在點對點系統(tǒng)中，消息被保留在隊列中。 一個或多個消費(fèi)者可以消耗隊列中的消息，但是特定消息只能由最多一個消費(fèi)者消費(fèi)。 一旦消費(fèi)者讀取隊列中的消息，它就從該隊列中消失。 該系統(tǒng)的典型示例是訂單處理系統(tǒng)，其中每個訂單將由一個訂單處理器處理，但多個訂單處理器也可以同時工作。 下圖描述了結(jié)構(gòu)

1.2 發(fā)布 - 訂閱消息系統(tǒng)

在發(fā)布 - 訂閱系統(tǒng)中，消息被保留在主題中。 與點對點系統(tǒng)不同，消費(fèi)者可以訂閱一個或多個主題并使用該主題中的所有消息。 在發(fā)布 - 訂閱系統(tǒng)中，消息生產(chǎn)者稱為發(fā)布者，消息使用者稱為訂閱者。

二、什么是Kafka

Apache Kafka 是一個分布式發(fā)布 - 訂閱消息系統(tǒng)和一個強(qiáng)大的隊列，可以處理大量的數(shù)據(jù)，并使你能夠?qū)⑾囊粋€端點傳遞到另一個端點。 Kafka 適合離線和在線消息消費(fèi)。 Kafka 消息保留在磁盤上，并在群集內(nèi)復(fù)制以防止數(shù)據(jù)丟失。 Kafka 構(gòu)建在 ZooKeeper 同步服務(wù)之上。 它與 Apache Storm 和 Spark 非常好地集成，用于實時流式數(shù)據(jù)分析。 Kafka 是一個分布式消息隊列，具有高性能、持久化、多副本備份、橫向擴(kuò)展能力。生產(chǎn)者往隊列里寫消息，消費(fèi)者從隊列里取消息進(jìn)行業(yè)務(wù)邏輯。一般在架構(gòu)設(shè)計中起到解耦、削峰、異步處理的作用。

2.1 關(guān)鍵術(shù)語

（1）生產(chǎn)者和消費(fèi)者（producer和consumer）：消息的發(fā)送者叫 Producer，消息的使用者和接受者是 Consumer，生產(chǎn)者將數(shù)據(jù)保存到 Kafka 集群中，消費(fèi)者從中獲取消息進(jìn)行業(yè)務(wù)的處理。

（2）broker：Kafka 集群中有很多臺 Server，其中每一臺 Server 都可以存儲消息，將每一臺 Server 稱為一個 kafka 實例，也叫做 broker。

（3）主題（topic）：一個 topic 里保存的是同一類消息，相當(dāng)于對消息的分類，每個 producer 將消息發(fā)送到 kafka 中，都需要指明要存的 topic 是哪個，也就是指明這個消息屬于哪一類。

（4）分區(qū)（partition）：每個 topic 都可以分成多個 partition，每個 partition 在存儲層面是 append log 文件。任何發(fā)布到此 partition 的消息都會被直接追加到 log 文件的尾部。為什么要進(jìn)行分區(qū)呢？最根本的原因就是：kafka基于文件進(jìn)行存儲，當(dāng)文件內(nèi)容大到一定程度時，很容易達(dá)到單個磁盤的上限，因此，采用分區(qū)的辦法，一個分區(qū)對應(yīng)一個文件，這樣就可以將數(shù)據(jù)分別存儲到不同的server上去，另外這樣做也可以負(fù)載均衡，容納更多的消費(fèi)者。

（5）偏移量（Offset）：一個分區(qū)對應(yīng)一個磁盤上的文件，而消息在文件中的位置就稱為 offset（偏移量），offset 為一個 long 型數(shù)字，它可以唯一標(biāo)記一條消息。由于kafka 并沒有提供其他額外的索引機(jī)制來存儲 offset，文件只能順序的讀寫，所以在kafka中幾乎不允許對消息進(jìn)行“隨機(jī)讀寫”。

2.2 Kafka要點

（1）kafka 是一個基于發(fā)布-訂閱的分布式消息系統(tǒng)（消息隊列）

（2）Kafka 面向大數(shù)據(jù)，消息保存在主題中，而每個 topic 有分為多個分區(qū)

（3）kafka 的消息數(shù)據(jù)保存在磁盤，每個 partition 對應(yīng)磁盤上的一個文件，消息寫入就是簡單的文件追加，文件可以在集群內(nèi)復(fù)制備份以防丟失

（4）即使消息被消費(fèi)，kafka 也不會立即刪除該消息，可以通過配置使得過一段時間后自動刪除以釋放磁盤空間

（5）kafka依賴分布式協(xié)調(diào)服務(wù)Zookeeper，適合離線/在線信息的消費(fèi)，與 storm 和 spark 等實時流式數(shù)據(jù)分析常常結(jié)合使用

三、Kafka基本原理

Kafka的設(shè)計初衷是建立一個統(tǒng)一的信息收集平臺，使其可以做到對信息的實時反饋。

3.1 分布式和分區(qū)（distributed、partitioned）

我們說 kafka 是一個分布式消息系統(tǒng)，所謂的分布式，實際上我們已經(jīng)大致了解。消息保存在 Topic 中，而為了能夠?qū)崿F(xiàn)大數(shù)據(jù)的存儲，一個 topic 劃分為多個分區(qū)，每個分區(qū)對應(yīng)一個文件，可以分別存儲到不同的機(jī)器上，以實現(xiàn)分布式的集群存儲。另外，每個 partition 可以有一定的副本，備份到多臺機(jī)器上，以提高可用性。 總結(jié)起來就是：一個 topic 對應(yīng)的多個 partition 分散存儲到集群中的多個 broker 上，存儲方式是一個 partition 對應(yīng)一個文件，每個 broker 負(fù)責(zé)存儲在自己機(jī)器上的 partition 中的消息讀寫。

3.2 副本（replicated ）

kafka 還可以配置 partitions 需要備份的個數(shù)(replicas),每個 partition 將會被備份到多臺機(jī)器上,以提高可用性，備份的數(shù)量可以通過配置文件指定。 這種冗余備份的方式在分布式系統(tǒng)中是很常見的，那么既然有副本，就涉及到對同一個文件的多個備份如何進(jìn)行管理和調(diào)度。kafka 采取的方案是：每個 partition 選舉一個 server 作為“leader”，由 leader 負(fù)責(zé)所有對該分區(qū)的讀寫，其他 server 作為 follower 只需要簡單的與 leader 同步，保持跟進(jìn)即可。如果原來的 leader 失效，會重新選舉由其他的 follower 來成為新的 leader。 至于如何選取 leader，實際上如果我們了解 ZooKeeper，就會發(fā)現(xiàn)其實這正是 Zookeeper 所擅長的，Kafka 使用 ZK 在 Broker 中選出一個 Controller，用于 Partition 分配和 Leader 選舉。 另外，這里我們可以看到，實際上作為 leader 的 server 承擔(dān)了該分區(qū)所有的讀寫請求，因此其壓力是比較大的，從整體考慮，有多少個 partition 就意味著會有多少個leader，kafka 會將 leader 分散到不同的 broker 上，確保整體的負(fù)載均衡。

3.3 整體數(shù)據(jù)流程

Kafka 的總體數(shù)據(jù)流滿足下圖，該圖可以說是概括了整個 kafka 的基本原理。

（1）數(shù)據(jù)生產(chǎn)過程（Produce）  

對于生產(chǎn)者要寫入的一條記錄，可以指定四個參數(shù)：

分別是 topic、partition、key 和 value，其中 topic 和 value（要寫入的數(shù)據(jù)）是必須要指定的，而 key 和 partition 是可選的。 對于一條記錄，先對其進(jìn)行序列化，然后按照 Topic 和 Partition，放進(jìn)對應(yīng)的發(fā)送隊列中。

如果 Partition 沒填，那么情況會是這樣的：

a、Key 有填。按照 Key 進(jìn)行哈希，相同 Key 去一個 Partition。

b、Key 沒填。Round-Robin 來選 Partition。

producer 將會和Topic下所有 partition leader 保持 socket 連接，消息由 producer 直接通過 socket 發(fā)送到 broker。其中 partition leader 的位置( host : port )注冊在 zookeeper 中，producer 作為 zookeeper client，已經(jīng)注冊了 watch 用來監(jiān)聽 partition leader 的變更事件，因此，可以準(zhǔn)確的知道誰是當(dāng)前的 leader。 producer 端采用異步發(fā)送：將多條消息暫且在客戶端 buffer 起來，并將他們批量的發(fā)送到 broker，小數(shù)據(jù) IO 太多，會拖慢整體的網(wǎng)絡(luò)延遲，批量延遲發(fā)送事實上提升了網(wǎng)絡(luò)效率。

（2）數(shù)據(jù)消費(fèi)過程（Consume）

對于消費(fèi)者，不是以單獨的形式存在的，每一個消費(fèi)者屬于一個 consumer group，一個 group 包含多個 consumer。特別需要注意的是：訂閱 Topic 是以一個消費(fèi)組來訂閱的，發(fā)送到 Topic 的消息，只會被訂閱此 Topic 的每個 group 中的一個 consumer 消費(fèi)。

如果所有的 Consumer 都具有相同的 group，那么就像是一個點對點的消息系統(tǒng)；如果每個 consumer 都具有不同的 group，那么消息會廣播給所有的消費(fèi)者。 具體說來，這實際上是根據(jù) partition 來分的，一個 Partition，只能被消費(fèi)組里的一個消費(fèi)者消費(fèi)，但是可以同時被多個消費(fèi)組消費(fèi)，消費(fèi)組里的每個消費(fèi)者是關(guān)聯(lián)到一個 partition 的

因此有這樣的說法：對于一個 topic,同一個 group 中不能有多于 partitions 個數(shù)的 consumer 同時消費(fèi),否則將意味著某些 consumer 將無法得到消息。 同一個消費(fèi)組的兩個消費(fèi)者不會同時消費(fèi)一個 partition。

在 kafka 中，采用了 pull 方式，即 consumer 在和 broker 建立連接之后，主動去 pull(或者說 fetch )消息，首先 consumer 端可以根據(jù)自己的消費(fèi)能力適時的去 fetch 消息并處理，且可以控制消息消費(fèi)的進(jìn)度(offset)。

partition 中的消息只有一個 consumer 在消費(fèi)，且不存在消息狀態(tài)的控制，也沒有復(fù)雜的消息確認(rèn)機(jī)制，可見 kafka broker 端是相當(dāng)輕量級的。當(dāng)消息被 consumer 接收之后，需要保存 Offset 記錄消費(fèi)到哪，以前保存在 ZK 中，由于 ZK 的寫性能不好，以前的解決方法都是 Consumer 每隔一分鐘上報一次，在 0.10 版本后，Kafka 把這個 Offset 的保存，從 ZK 中剝離，保存在一個名叫 consumeroffsets topic 的 Topic 中，由此可見，consumer 客戶端也很輕量級。

3.4 消息傳送機(jī)制

Kafka 支持 3 種消息投遞語義,在業(yè)務(wù)中，常常都是使用 At least once 的模型。

At most once：最多一次，消息可能會丟失，但不會重復(fù)。At least once：最少一次，消息不會丟失，可能會重復(fù)。Exactly once：只且一次，消息不丟失不重復(fù)，只且消費(fèi)一次。

四、Kafka 集群架構(gòu)

Kafka的集群圖：

（1）Broker（代理）：Kafka集群通常由多個代理組成以保持負(fù)載平衡。 Kafka代理是無狀態(tài)的，所以他們使用ZooKeeper來維護(hù)它們的集群狀態(tài)。 一個Kafka代理實例可以每秒處理數(shù)十萬次讀取和寫入，每個Broker可以處理TB的消息，而沒有性能影響。 Kafka經(jīng)紀(jì)人領(lǐng)導(dǎo)選舉可以由ZooKeeper完成。

（2）ZooKeeper： ZooKeeper用于管理和協(xié)調(diào)Kafka代理。 ZooKeeper服務(wù)主要用于通知生產(chǎn)者和消費(fèi)者Kafka系統(tǒng)中存在任何新代理或Kafka系統(tǒng)中代理失敗。 根據(jù)Zookeeper接收到關(guān)于代理的存在或失敗的通知，然后生產(chǎn)者和消費(fèi)者采取決定并開始與某些其他代理協(xié)調(diào)他們的任務(wù)。

（3）Producers（生產(chǎn)者） 生產(chǎn)者將數(shù)據(jù)推送給經(jīng)紀(jì)人。 當(dāng)新代理啟動時，所有生產(chǎn)者搜索它并自動向該新代理發(fā)送消息。 Kafka生產(chǎn)者不等待來自代理的確認(rèn)，并且發(fā)送消息的速度與代理可以處理的一樣快。

（4）Consumers（消費(fèi)者） 因為Kafka代理是無狀態(tài)的，這意味著消費(fèi)者必須通過使用分區(qū)偏移來維護(hù)已經(jīng)消耗了多少消息。 如果消費(fèi)者確認(rèn)特定的消息偏移，則意味著消費(fèi)者已經(jīng)消費(fèi)了所有先前的消息。 消費(fèi)者向代理發(fā)出異步拉取請求，以具有準(zhǔn)備好消耗的字節(jié)緩沖區(qū)。 消費(fèi)者可以簡單地通過提供偏移值來快退或跳到分區(qū)中的任何點。 消費(fèi)者偏移值由ZooKeeper通知。

五、Kafka 工作流程

Kafka 只是分為一個或多個分區(qū)的主題的集合。Kafka 分區(qū)是消息的線性有序序列，其中每個消息由它們的索引(稱為偏移)來標(biāo)識。Kafka 集群中的所有數(shù)據(jù)都是不相連的分區(qū)聯(lián)合。 傳入消息寫在分區(qū)的末尾，消息由消費(fèi)者順序讀取。 通過將消息復(fù)制到不同的代理提供持久性。

Kafka 以快速，可靠，持久，容錯和零停機(jī)的方式提供基于pub-sub 和隊列的消息系統(tǒng)。 在這兩種情況下，生產(chǎn)者只需將消息發(fā)送到主題，消費(fèi)者可以根據(jù)自己的需要選擇任何一種類型的消息傳遞系統(tǒng)。

5.1 發(fā)布 - 訂閱消息的工作流程

Pub-Sub 消息的逐步工作流程 -

生產(chǎn)者定期向主題發(fā)送消息。Kafka 代理存儲為該特定主題配置的分區(qū)中的所有消息。 它確保消息在分區(qū)之間平等共享。 如果生產(chǎn)者發(fā)送兩個消息并且有兩個分區(qū)，Kafka 將在第一分區(qū)中存儲一個消息，在第二分區(qū)中存儲第二消息。消費(fèi)者訂閱特定主題。一旦消費(fèi)者訂閱主題，Kafka 將向消費(fèi)者提供主題的當(dāng)前偏移，并且還將偏移保存在 Zookeeper 系統(tǒng)中。

消費(fèi)者將定期請求 Kafka (如100 Ms)新消息。一旦 Kafka 收到來自生產(chǎn)者的消息，它將這些消息轉(zhuǎn)發(fā)給消費(fèi)者。消費(fèi)者將收到消息并進(jìn)行處理。一旦消息被處理，消費(fèi)者將向 Kafka 代理發(fā)送確認(rèn)。一旦 Kafka 收到確認(rèn)，它將偏移更改為新值，并在 Zookeeper 中更新它。 由于偏移在 Zookeeper 中維護(hù)，消費(fèi)者可以正確地讀取下一封郵件，即使在服務(wù)器暴力期間。以上流程將重復(fù)，直到消費(fèi)者停止請求。消費(fèi)者可以隨時回退/跳到所需的主題偏移量，并閱讀所有后續(xù)消息。

5.2 隊列消息/用戶組的工作流

在隊列消息傳遞系統(tǒng)而不是單個消費(fèi)者中，具有相同組 ID 的一組消費(fèi)者將訂閱主題。

簡單來說，訂閱具有相同 Group ID 的主題的消費(fèi)者被認(rèn)為是單個組，并且消息在它們之間共享。

生產(chǎn)者以固定間隔向某個主題發(fā)送消息。Kafka存儲在為該特定主題配置的分區(qū)中的所有消息，類似于前面的方案。單個消費(fèi)者訂閱特定主題，假設(shè) Topic-01 為 Group ID 為 Group-1 。Kafka 以與發(fā)布 - 訂閱消息相同的方式與消費(fèi)者交互，直到新消費(fèi)者以相同的組 ID 訂閱相同主題Topic-01 1 。一旦新消費(fèi)者到達(dá)，

Kafka 將其操作切換到共享模式，并在兩個消費(fèi)者之間共享數(shù)據(jù)。 此共享將繼續(xù)，直到用戶數(shù)達(dá)到為該特定主題配置的分區(qū)數(shù)。一旦消費(fèi)者的數(shù)量超過分區(qū)的數(shù)量，新消費(fèi)者將不會接收任何進(jìn)一步的消息，直到現(xiàn)有消費(fèi)者取消訂閱任何一個消費(fèi)者。 出現(xiàn)這種情況是因為 Kafka 中的每個消費(fèi)者將被分配至少一個分區(qū)，并且一旦所有分區(qū)被分配給現(xiàn)有消費(fèi)者，新消費(fèi)者將必須等待。此功能也稱為使用者組。 同樣，Kafka 將以非常簡單和高效的方式提供兩個系統(tǒng)中最好的。

六、ZooKeeper 的作用

Apache Kafka 的一個關(guān)鍵依賴是 Apache Zookeeper，它是一個分布式配置和同步服務(wù)。Zookeeper 是 Kafka 代理和消費(fèi)者之間的協(xié)調(diào)接口。Kafka 服務(wù)器通過 Zookeeper 集群共享信息。Kafka 在 Zookeeper 中存儲基本元數(shù)據(jù)，例如關(guān)于主題，代理，消費(fèi)者偏移(隊列讀取器)等的信息。

由于所有關(guān)鍵信息存儲在 Zookeeper 中，并且它通常在其整體上復(fù)制此數(shù)據(jù)，因此Zookeeper 代理的故障不會影響 Kafka 集群的狀態(tài)。一旦 Zookeeper 重新啟動，Kafka 將恢復(fù)狀態(tài)。 這為Kafka帶來了零停機(jī)時間。Kafka 代理之間的領(lǐng)導(dǎo)者選舉也通過使用 Zookeeper 在領(lǐng)導(dǎo)者失敗的情況下完成。

到此這篇關(guān)于Java消息隊列Kafka的簡單概述的文章就介紹到這了,更多相關(guān)Java消息隊列Kafka內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論