1. Kafka 概述

?

1.1?定義

?

Kakfa 是一個分布式的基于發布/訂閱模式的消息隊列（message queue），主要應用于大數據的實時處理領域。

?

1.2?消息隊列

?

1.2.1 傳統消息隊列與新式消息隊列模式

?

?

上面是傳統的消息隊列，比如一個用戶要注冊信息，當用戶信息寫入數據庫后，后面還有一些其他流程，比如發送短信，則需要等這些流程處理完成后，再返回給用戶。而新式隊列，比如一個用戶注冊信息，數據直接丟進數據庫，就直接返回給用戶成功。

?

1.2.2?使用消息隊列的好處

?

• 解耦

• 可恢復性

• 緩沖

• 靈活性與峰值處理能力

• 異步通信

?

1.2.3?消息隊列的模式

?

1)?點對點模式

?

消息生產者發送消息到消息隊列中，然后消息消費者從隊列中取出并且消費消息，消息被消費后，隊列中不在存儲。所以消息消費者不可能消費到已經被消費的消息；隊列支持存在多個消費者，但是對于一個消息而言，只會 有一個消費者可以消費；如果想發給多個消費者，則需要多次發送該條消息。

?

2)?發布/訂閱模式（一對多，消費者消費數據之后不會清除消息）

?

消息生產者將消息發布到 topic 中，同時有多個消息消費者（訂閱）消費該消息。和點對點的方式不同，發布到 topic 的消息會被所有的訂閱者消費；但是數據保留是期限的，默認是 7 天，因為它不是存儲系統。Kafka 就是這種模式的。有兩種方式，一種是是消費者去主動去消費（拉取）消息，而不是生產者推送消息給消費者；另外一種就是生產者主動推送消息給消費者，類似公眾號。

?

1.3 Kafka 基礎架構

?

?

Kafka 的基礎架構主要有 broker、生產者、消費者組構成，當前還包括 ZooKeeper。

?

?

生產者負責發送消息，broker?負責緩沖消息，broker 中可以創建 topic，每個 topic 又有 partition 和 replication 的概念。

?

消費者組負責處理消息，同一個消費者組的中消費者不能消費同一個 partition 中的數據。消費者組主要是提高消費能力，比如之前是一個消費者消費 100?條數據，現在是 2 個消費者消費 100 條數據，可以提高消費能力。所以消費者組的消費者的個數要小于 partition 的個數，不然就會有消費者沒有 partition 可以消費，造成資源的浪費。

?

注意：不同消費者組的消費者是可以消費相同的 partition 數據。

?

Kakfa?如果要組件集群，則只需要注冊到一個 ZooKeeper?中就可以了，ZooKeeper?中還保留消息消費的進度或者說偏移量或者消費位置。

?

• 0.9?之前的版本偏移量存儲在?ZooKeeper；

• 0.9?之后的版本偏移量存儲在 Kafka中。Kafka 定義了一個系統 topic，專用用來存儲偏移量的數據。

?

為什么要改？

?

主要是考慮到頻繁更改偏移量，對?ZooKeeper?的壓力較大，而且 Kafka 本身自己的處理也較復雜。

?

1.4 安裝 Kafka

?

1) Kafka 的安裝只需要解壓安裝包就可以完成安裝。

?

tar -zxvf kafka_2.11-2.1.1.tgz -C /usr/local/

?

2) 查看配置文件。

?

[root@es1 config]# pwd /usr/local/kafka/config [root@es1 config]# ll total 84 -rw-r--r--. 1 root root 906 Feb 8 2019 connect-console-sink.properties -rw-r--r--. 1 root root 909 Feb 8 2019 connect-console-source.properties -rw-r--r--. 1 root root 5321 Feb 8 2019 connect-distributed.properties -rw-r--r--. 1 root root 883 Feb 8 2019 connect-file-sink.properties -rw-r--r--. 1 root root 881 Feb 8 2019 connect-file-source.properties -rw-r--r--. 1 root root 1111 Feb 8 2019 connect-log4j.properties -rw-r--r--. 1 root root 2262 Feb 8 2019 connect-standalone.properties -rw-r--r--. 1 root root 1221 Feb 8 2019 consumer.properties -rw-r--r--. 1 root root 4727 Feb 8 2019 log4j.properties -rw-r--r--. 1 root root 1925 Feb 8 2019 producer.properties -rw-r--r--. 1 root root 6865 Jan 16 22:00 server-1.properties -rw-r--r--. 1 root root 6865 Jan 16 22:00 server-2.properties -rw-r--r--. 1 root root 6873 Jan 16 03:57 server.properties -rw-r--r--. 1 root root 1032 Feb 8 2019 tools-log4j.properties -rw-r--r--. 1 root root 1169 Feb 8 2019 trogdor.conf -rw-r--r--. 1 root root 1023 Feb 8 2019 zookeeper.properties

?

3) 修改配置文件 server.properties。

?

設置 broker.id 這個是 Kafka 集群區分每個節點的唯一標志符。

?

?

4) 設置 Kafka 的數據存儲路徑。

?

?

注意：這個目錄下不能有其他非 Kafka 目錄，不然會導致 Kafka 集群無法啟動。

?

5) 設置是否可以刪除 topic，默認 Kafka 的 topic 是不允許刪除的。

?

?

6) Kafka 的數據保留的時間，默認是 7 天。

?

?

7) Log 文件最大的大小，如果 log 文件超過 1 G 會創建一個新的文件。

?

?

?

8) Kafka 連接的 ZooKeeper 的地址和連接 Kafka 的超時時間。

?

?

9) 默認的 partition 的個數。

?

?

1.5 啟動 Kafka

?

1) 啟動方式一，Kafka 只能單節點啟動，所以每個 Kakfa 節點都需要手動啟動，下面的方式阻塞的方式啟動。

?

?

2) 啟動方式二，守護的方式啟動，推薦使用。

?

?

1.6 Kafka 操作

?

1) 查看當前 Kafka 集群已有的 topic。

?

?

注意：這里連接的 ZooKeeper，而不是連接的 Kafka。

?

2) 創建 topic，指定分片和副本個數。

?

?

說明：replication-factor?副本數，replication-factor?分區數，topic?主題名。

?

?

如果當前 Kafka 集群只有 3 個 broker 節點，則 replication-factor 最大就是 3 了，下面的例子創建副本為 4，則會報錯。

?

?

3) 刪除 topic。

?

?

4） 查看 topic 信息。

?

?

?

1.7 啟動生產者生產消息，Kafka 自帶一個生產者和消費者的客戶端

?

1) 啟動一個生產者，注意此時連的 9092 端口，連接的 Kafka 集群。

?

?

2) 啟動一個消費者，注意此時連接的還是 9092 端口，在 0.9 版本之前連接的還是 2181 端口。

?

?

這里我們啟動 2 個消費者來測試一下。

?

?

說明：如果不指定的消費者組的配置文件的話，默認每個消費者都屬于不同的消費者組。

?

3) 發送消息，可以看到每個消費者都能收到消息。

?

?

?

?

4）Kakfa 中的實際數據。

?

?

?

2. Kafka 架構深入

?

?

Kafka 不能保證消息的全局有序，只能保證消息在 partition 內有序，因為消費者消費消息是在不同的 partition 中隨機的。

?

2.1 Kafka 的工作流程

?

Kafka 中的消息是以 topic 進行分類的，生產者生成消息、消費者消費消息都面向 topic。

?

?

Topic 是一個邏輯上的概念，而 partition 是物理上的概念。每個 partition 又有副本的概念。每個 partition 對應于一個 log 文件，該 log 文件中存儲的就是生產者生成的數據，生產者生成的數據會不斷的追加到該 log 的文件末端，且每條數據都有自己的 offset，消費者都會實時記錄自己消費到了那個 offset，以便出錯的時候從上次的位置繼續消費，這個 offset 就保存在 index 文件中。Kafka 的 offset 是分區內有序的，但是在不同分區中是無順序的，Kafka 不保證數據的全局有序。

?

2.2 Kafka 原理

?

由于生產者生產的消息會不斷追加到 log 文件的末尾，為防止 log 文件過大導致數據定位效率低下，Kafka 采用分片和索引的機制，將每個 partition 分為多個 segment，每個 segment 對應2個文件 — index 文件和 log 文件，這 2 個文件位于一個相同的文件夾下，文件夾的命名規則為：topic 名稱 + 分區序號。

?

?

Index 和 log 的文件的文件名是當前這個索引是最小的數據的 offset。Kafka?如何快速的消費數據呢？

?

?

Index 文件中存儲的數據的索引信息，第一列是 offset，第二列這這個數據所對應的 log 文件中的偏移量，就像我們去讀文件，使用 seek() 設置當前鼠標的位置一樣，可以更快的找到數據。

?

如果要去消費 offset 為 3 的數據，首先通過二分法找到數據在哪個 index 文件中，然后在通過 index 中 offset 找到數據在 log 文件中的 offset；這樣就可以快速的定位到數據，并消費。

?

所以，Kakfa 雖然把數據存儲在磁盤中，但是他的讀取速度還是非?？斓?。

?

3. Kafka 生產者和消費者

?

3.1 Kafka 生產者

?

Kafka 的 partition 分區的作用

?

Kafka 分區的原因主要就是提供并發提高性能，因為讀寫是 partition 為單位讀寫的；那生產者發送消息是發送到哪個?partition?中呢？

?

在客戶端中指定 partition；

輪詢（推薦）消息1去 p1，消息2去 p2，消息3去 p3，消息4去 p1，消息5去 p2，消息6去 p3……

?

3.2 Kafka 如何保證數據可靠性呢？通過 ack 來保證

?

為保證生產者發送的數據，能可靠的發送到指定的 topic，topic 的每個 partition 收到生產者發送的數據后，都需要向生產者發送 ack（確認收到），如果生產者收到 ack，就會進行下一輪的發送，否則重新發送數據。

?

?

?

那么 Kafka 什么時候向生產者發送 ack？

?

確保 follower 和 leader 同步完成，leader 在發送 ack 給生產者，這樣才能確保 leader 掛掉之后，能在 follower 中選舉出新的 leader 后，數據不會丟失。

?

那多少個 follower 同步完成后發送 ack？

?

• 方案1：半數已經完成同步，就發送 ack；

• 方案2：全部完成同步，才發送 ack（Kafka采用這種方式）

?

采用第二種方案后，設想以下場景：leader 收到數據，所有的 follower 都開始同步數據，但是有一個 follower 因為某種故障，一直無法完成同步，那 leader 就要一直等下，直到他同步完成，才能發送 ack。這樣就非常影響效率，這個問題怎么解決？

?

?

Leader 維護了一個動態的 ISR 列表（同步副本的作用），只需要這個列表的中的 follower 和 leader 同步；當 ISR 中的 follower 完成數據的同步之后，leader 就會給生產者發送 ack，如果 follower 長時間未向 leader 同步數據，則該 follower 將被剔除ISR，這個時間閾值也是自定義的；同樣 leader 故障后，就會從 ISR 中選舉新的 leader。

?

怎么選擇 ISR 的節點呢？

?

首先通信的時間要快，要和 leader 要可以很快的完成通信，這個時間默認是 10s

然后就看 leader 數據差距，消息條數默認是 10000 條（后面版本被移除）

為什么移除：因為 Kafka 發送消息是批量發送的，所以會一瞬間 leader 接受完成，但是 follower 還沒有拉取，所以會頻繁踢出和加入ISR，這個數據會保存到 ZooKeeper 和內存中，所以會頻繁更新?ZooKeeper?和內存。

?

但是對于某些不太重要的數據，對數據的可靠性要求不是很高，能夠容忍數據的少量丟失，所以沒必要等 ISR 中的 follower 全部接受成功。所以 Kafka 為用戶提供了三種可靠性級別，用戶可以根據可靠性和延遲進行權衡，這個設置在kafka的生成中設置：ack 參數設置。

?

?

1) acks 為 0

?

生產者不等 ack，只管往 topic 丟數據就可以了，這個丟數據的概率非常高。

?

2) ack為 1

?

leader 落盤后就會返回 ack，會有數據丟失的現象，如果 leader 在同步完成后出現故障，則會出現數據丟失。

?

3) ack為-1（all）

?

leader 和 follower（ISR）落盤才會返回 ack，會有數據重復現象，如果在 leader 已經寫完成，且 follower 同步完成，但是在返回ack的出現故障，則會出現數據重復現象；極限情況下，這個也會有數據丟失的情況，比如 follower 和 leader 通信都很慢，所以 ISR 中只有一個 leader 節點，這個時候，leader 完成落盤，就會返回 ack，如果此時 leader 故障后，就會導致丟失數據。

?

3.3 Kafka 如何保證消費數據的一致性？通過HW來保證

?

?

• LEO：指每個 follower 的最大的 offset；

• HW（高水位）：指消費者能見到的最大的 offset，LSR 隊列中最小的 LEO，也就是說消費者只能看到1~6的數據，后面的數據看不到，也消費不了。

?

避免 leader 掛掉后，比如當前消費者消費8這條數據后，leader 掛了，此時比如 f2 成為 leader，f2 根本就沒有9這條數據，那么消費者就會報錯，所以設計了 HW 這個參數，只暴露最少的數據給消費者，避免上面的問題。

?

3.3.1 HW 保證數據存儲的一致性

?

1) follower 故障

?

follower 發生故障后會被臨時提出 LSR，待該 follower 恢復后，follower 會讀取本地的磁盤記錄的上次的 HW，并將該 log 文件高于 HW 的部分截取掉，從 HW 開始向 leader 進行同步，等該 follower 的 LEO 大于等于該 partition 的 hw，即 follower 追上leader后，就可以重新加入 LSR。

?

2) leader 故障

?

leader 發生故障后，會從 ISR 中選出一個新的 leader，之后，為了保證多個副本之間的數據一致性，其余的 follower 會先將各自的 log 文件高于 hw 的部分截掉（新 leader 自己不會截掉），然后從新的 leader 同步數據。

?

注意：這個是為了保證多個副本間的數據存儲的一致性，并不能保證數據不丟失或者不重復。

?

3.3.2 精準一次（冪等性），保證數據不重復

?

• ack 設置為 -1，則可以保證數據不丟失，但是會出現數據重復（at least once）

• ack 設置為 0，則可以保證數據不重復，但是不能保證數據不丟失（at most once）

?

但是如果魚和熊掌兼得，該怎么辦？這個時候就就引入了 Exact once（精準一次）。

?

在 0.11 版本后，引入冪等性解決 Kakfa 集群內部的數據重復，在 0.11 版本之前，在消費者處自己做處理。如果啟用了冪等性，則 ack 默認就是-1，Kafka 就會為每個生產者分配一個 pid，并未每條消息分配 seqnumber，如果 pid、partition、seqnumber 三者一樣，則 Kafka 認為是重復數據，就不會落盤保存；但是如果生產者掛掉后，也會出現有數據重復的現象；所以冪等性解決在單次會話的單個分區的數據重復，但是在分區間或者跨會話的是數據重復的是無法解決的。

?

?

3.4 Kafka 消費者

?

3.4.1 消費方式

?

消息隊列有兩種消費消息的方式，push（微信公眾號）、pull（kafka）。push 模式很難適應消費速率不同的消費者，因為消費發送速率是由 broker 決定的，他的目標是盡可能以最快的的速度傳遞消息，但是這樣很容易造成消費者來不及處理消息，典型的表現就是拒絕服務以及網絡擁塞。而 pull 的方式可以消費者的消費能力以適當的速率消費消息。

?

pull 模式的不足之處是如果 Kafka 沒有數據，消費者可能會陷入死循環，一直返回空數據，針對這一點，Kafka 消費者在消費數據時候回傳遞一個 timeout 參數，如果當時沒有數據可供消費，消費者會等待一段時間在返回。

?

3.4.2 分區分配策略

?

一個消費者組有多個消費者，一個 topic 有多個 partition。所以必然會涉及到 partition 的分配問題，即確定哪個 partition 由哪個消費者來消費。Kafka 提供兩種方式，一種是輪詢（RountRobin）對于 topic 組生效，一種是（Range）對于單個topic生效

?

輪詢：前置條件是需要一個消費者里的消費者訂閱的是相同的 topic。不然就會出現問題；非默認的的方式。

?

同一個消費者組里的消費者不能同時消費同一個分區，比如三個消費者消費一個 topic 的 9 個分區。

?

?

?

如果一個消費者組里有2個消費者，這個消費者組里同時消費 2 個 topic，每個 topic 又有三個 partition。首先會把 2 個 topic 當做一個主題，然后根據 topic 和 partition 做 hash，然后在按照 hash 排序。然后輪詢分配給一個消費者組中的 2 個消費者。

?

?

如果是下面這樣的方式訂閱的呢？

?

比如有 3 個 topic，每個 topic 有 3 個 partition，一個消費者組中有 2 個消費者。消費者1訂閱 topic1 和 topic2，消費者2訂閱 topic2 和 topic3。那么這樣的場景，使用輪訓的方式訂閱 topic 就會有問題。

?

如果是下面這種方式訂閱呢？

?

比如有2個 topic，每個 topic 有3個 partition，一個消費者組 有2個消費者，消費者1訂閱 topic1，消費者2訂閱 topic2，這樣使用輪訓的方式訂閱 topic 也會有問題。

?

所以我們一直強調，使用輪訓的方式訂閱 topic 的前提是一個消費者組中的所有消費者訂閱的主題是一樣的；所以輪詢的方式不是 Kafka 默認的方式；Range?是按照單個 topic 來劃分的，默認的分配方式。

?

?

?

?

Range 的問題會出現消費者數據不均衡的問題。比如下面的例子，一個消費者組訂閱了 2 個 topic，就會出現消費者1消費 4 個 partition，而另外一個消費者只消費 2 個 partition。

?

?

分區策略什么時候會觸發呢？當消費者組里的消費者個數變化的時候，會觸發分區策略調整，比如消費者里增加消費者，或者減少消費者。

?

3.4.3?維護?offset

?

由于消費者在消費過程中可能會出現斷電宕機等故障，消費者恢復后，需要從故障前的位置繼續消費，所以消費者需要實施記錄自己消費哪個 offset，以便故障恢復后繼續消費。

?

Offset保存的位置有2個，一個 ZooKeeper，一個是 Kafka。首先看下 offset 保存到?ZooKeeper，由消費者組、topic、partition 三個元素確定唯一的 offset。

?

?

所以消費者組中的某個消費者掛掉之后，或者的消費者還是可以拿到這個 offset。

?

?

Controller 這個節點和?ZooKeeper?通信，同步數據，這個節點就是誰先起來，誰就先注冊 controller，誰就是 controller。其他節點和 controller 信息保持同步。

?

3.4.5 消費者組的案例

?

修改消費者組 id

?

?

啟動一個消費者發送 3 條數據。

?

?

指定消費者組啟動消費者，啟動三個消費者，可以看到每個消費者消費了一條數據。

?

?

?

?

在演示下不同組可以消費同一個 topic 的，我們看到 2 個消費者的消費者都消費到同一條數據。再次啟動一個消費者，這個消費者屬于另外一個消費者組。

?

?

?

4?Kafka 的高效讀寫機制

?

4.1 分布式部署

?

多節點并行操作。

?

4.2、順序寫磁盤

?

Kafka 的 producer 生產數據，要寫入到 log 文件中，寫的過程中一直追加到文件末尾，為順序寫，官網有數據表明。同樣的磁盤，順序寫能到 600M/S，而隨機寫只有 100K/S。這與磁盤的機械結構有關，順序寫之所以快，是因為其省去了大量磁頭尋址的時間。

?

4.3、零復制技術

?

正常情況下，先把數據讀到內核空間，在從內核空間把數據讀到用戶空間，然后在調操作系統的 IO 接口寫到內核空間，最終在寫到硬盤中。

?

?

Kafka 是這樣做的，直接在內核空間流轉 IO 流，所以 Kafka 的性能非常高。

?

?

?

5. ZooKeeper 在 Kafka 中的作用

?

Kafka 集群中有一個 broker 會被選舉為 controller，負責管理集群 broker 的上下線，所有的 topic 的分區副本分配和 leader 選舉等工作。

總結

以上是生活随笔為你收集整理的临近春招，Kafka是不是忘完了，给你一文概括Kafka的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。