之前我們介紹過了 Kafka 整體架構，Kafka 生產者，Kafka 生產的訊息最終流向哪裡呢？當然是需要消費了，要不只產生一系列資料沒有任何作用啊，如果把 Kafka 比作餐廳的話，那麼生產者就是廚師的角色，消費者就是客人，只有廚師的話，那麼炒出來的菜沒有人吃也沒有意義，如果只有客人沒有廚師的話，誰會去這個店吃飯呢？！所以如果你看完前面的文章意猶未盡的話，可以繼續讓你爽一爽。如果你沒看過前面的文章，那就從現在開始讓你爽。

Kafka 消費者概念

應用程式使用 KafkaConsumer 從 Kafka 中訂閱主題並接收來自這些主題的訊息，然後再把他們儲存起來。應用程式首先需要建立一個 KafkaConsumer 物件，訂閱主題並開始接受訊息，驗證訊息並儲存結果。一段時間後，生產者往主題寫入的速度超過了應用程式驗證資料的速度，這時候該如何處理？如果只使用單個消費者的話，應用程式會跟不上訊息生成的速度，就像多個生產者像相同的主題寫入訊息一樣，這時候就需要多個消費者共同參與消費主題中的訊息，對訊息進行分流處理。

Kafka 消費者從屬於消費者群組。一個群組中的消費者訂閱的都是相同的主題，每個消費者接收主題一部分分割槽的訊息。下面是一個 Kafka 分割槽消費示意圖

上圖中的主題 T1 有四個分割槽，分別是分割槽0、分割槽1、分割槽2、分割槽3，我們建立一個消費者群組1，消費者群組中只有一個消費者，它訂閱主題T1，接收到 T1 中的全部訊息。由於一個消費者處理四個生產者傳送到分割槽的訊息，壓力有些大，需要幫手來幫忙分擔任務，於是就演變為下圖

這樣一來，消費者的消費能力就大大提高了，但是在某些環境下比如使用者產生訊息特別多的時候，生產者產生的訊息仍舊讓消費者吃不消，那就繼續增加消費者。

如上圖所示，每個分割槽所產生的訊息能夠被每個消費者群組中的消費者消費，如果向消費者群組中增加更多的消費者，那麼多餘的消費者將會閒置，如下圖所示

向群組中增加消費者是橫向伸縮消費能力的主要方式。總而言之，我們可以通過增加消費組的消費者來進行水平擴充套件提升消費能力。這也是為什麼建議建立主題時使用比較多的分割槽數，這樣可以在消費負載高的情況下增加消費者來提升效能。另外，消費者的數量不應該比分割槽數多，因為多出來的消費者是空閒的，沒有任何幫助。

Kafka 一個很重要的特性就是，只需寫入一次訊息，可以支援任意多的應用讀取這個訊息。換句話說，每個應用都可以讀到全量的訊息。為了使得每個應用都能讀到全量訊息，應用需要有不同的消費組。對於上面的例子，假如我們新增了一個新的消費組 G2，而這個消費組有兩個消費者，那麼就演變為下圖這樣

在這個場景中，消費組 G1 和消費組 G2 都能收到 T1 主題的全量訊息，在邏輯意義上來說它們屬於不同的應用。

總結起來就是如果應用需要讀取全量訊息，那麼請為該應用設定一個消費組；如果該應用消費能力不足，那麼可以考慮在這個消費組裡增加消費者。

消費者組和分割槽重平衡

消費者組是什麼

消費者組（Consumer Group）是由一個或多個消費者例項（Consumer Instance）組成的群組，具有可擴充套件性和可容錯性的一種機制。消費者組內的消費者共享一個消費者組ID，這個ID 也叫做 Group ID，組內的消費者共同對一個主題進行訂閱和消費，同一個組中的消費者只能消費一個分割槽的訊息，多餘的消費者會閒置，派不上用場。

我們在上面提到了兩種消費方式

• 一個消費者群組消費一個主題中的訊息，這種消費模式又稱為點對點的消費方式，點對點的消費方式又被稱為訊息佇列
• 一個主題中的訊息被多個消費者群組共同消費，這種消費模式又稱為釋出-訂閱模式

消費者重平衡

我們從上面的消費者演變圖中可以知道這麼一個過程：最初是一個消費者訂閱一個主題並消費其全部分割槽的訊息，後來有一個消費者加入群組，隨後又有更多的消費者加入群組，而新加入的消費者例項分攤了最初消費者的部分訊息，這種把分割槽的所有權通過一個消費者轉到其他消費者的行為稱為重平衡，英文名也叫做 Rebalance 。如下圖所示

重平衡非常重要，它為消費者群組帶來了高可用性 和 伸縮性，我們可以放心的新增消費者或移除消費者，不過在正常情況下我們並不希望發生這樣的行為。在重平衡期間，消費者無法讀取訊息，造成整個消費者組在重平衡的期間都不可用。另外，當分割槽被重新分配給另一個消費者時，訊息當前的讀取狀態會丟失，它有可能還需要去重新整理快取，在它重新恢復狀態之前會拖慢應用程式。

消費者通過向組織協調者（Kafka Broker）傳送心跳來維護自己是消費者組的一員並確認其擁有的分割槽。對於不同不的消費群體來說，其組織協調者可以是不同的。只要消費者定期傳送心跳，就會認為消費者是存活的並處理其分割槽中的訊息。當消費者檢索記錄或者提交它所消費的記錄時就會發送心跳。

如果過了一段時間 Kafka 停止傳送心跳了，會話（Session）就會過期，組織協調者就會認為這個 Consumer 已經死亡，就會觸發一次重平衡。如果消費者宕機並且停止傳送訊息，組織協調者會等待幾秒鐘，確認它死亡了才會觸發重平衡。在這段時間裡，死亡的消費者將不處理任何訊息。在清理消費者時，消費者將通知協調者它要離開群組，組織協調者會觸發一次重平衡，儘量降低處理停頓。

重平衡是一把雙刃劍，它為消費者群組帶來高可用性和伸縮性的同時，還有有一些明顯的缺點(bug)，而這些 bug 到現在社群還無法修改。

重平衡的過程對消費者組有極大的影響。因為每次重平衡過程中都會導致萬物靜止，參考 JVM 中的垃圾回收機制，也就是 Stop The World ，STW，(引用自《深入理解 Java 虛擬機器》中 p76 關於 Serial 收集器的描述)：

更重要的是它在進行垃圾收集時，必須暫停其他所有的工作執行緒。直到它收集結束。Stop The World 這個名字聽起來很帥，但這項工作實際上是由虛擬機器在後臺自動發起並完成的，在使用者不可見的情況下把使用者正常工作的執行緒全部停掉，這對很多應用來說都是難以接受的。

也就是說，在重平衡期間，消費者組中的消費者例項都會停止消費，等待重平衡的完成。而且重平衡這個過程很慢......

建立消費者

上面的理論說的有點多，下面就通過程式碼來講解一下消費者是如何消費的

在讀取訊息之前，需要先建立一個 KafkaConsumer 物件。建立 KafkaConsumer 物件與建立 KafkaProducer 物件十分相似 --- 把需要傳遞給消費者的屬性放在 properties 物件中，後面我們會著重討論 Kafka 的一些配置，這裡我們先簡單的建立一下，使用3個屬性就足矣，分別是 bootstrap.server，key.deserializer，value.deserializer 。

這三個屬性我們已經用過很多次了，如果你還不是很清楚的話，可以參考 帶你漲姿勢是認識一下Kafka Producer

還有一個屬性是 group.id 這個屬性不是必須的，它指定了 KafkaConsumer 是屬於哪個消費者群組。建立不屬於任何一個群組的消費者也是可以的

Properties properties = new Properties();
        properties.put("bootstrap.server","192.168.1.9:9092");     properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");   properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
KafkaConsumer<String,String> consumer = new KafkaConsumer<>(properties);

主題訂閱

建立好消費者之後，下一步就開始訂閱主題了。subscribe() 方法接受一個主題列表作為引數，使用起來比較簡單

consumer.subscribe(Collections.singletonList("customerTopic"));

為了簡單我們只訂閱了一個主題 customerTopic，引數傳入的是一個正則表示式，正則表示式可以匹配多個主題，如果有人建立了新的主題，並且主題的名字與正則表示式相匹配，那麼會立即觸發一次重平衡，消費者就可以讀取新的主題。

要訂閱所有與 test 相關的主題，可以這樣做

consumer.subscribe("test.*");

輪詢

我們知道，Kafka 是支援訂閱/釋出模式的，生產者傳送資料給 Kafka Broker，那麼消費者是如何知道生產者傳送了資料呢？其實生產者產生的資料消費者是不知道的，KafkaConsumer 採用輪詢的方式定期去 Kafka Broker 中進行資料的檢索，如果有資料就用來消費，如果沒有就再繼續輪詢等待，下面是輪詢等待的具體實現

try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(100));
    for (ConsumerRecord<String, String> record : records) {
      int updateCount = 1;
      if (map.containsKey(record.value())) {
        updateCount = (int) map.get(record.value() + 1);
      }
      map.put(record.value(), updateCount);
    }
  }
}finally {
  consumer.close();
}

• 這是一個無限迴圈。消費者實際上是一個長期執行的應用程式，它通過輪詢的方式向 Kafka 請求資料。
• 第三行程式碼非常重要，Kafka 必須定期迴圈請求資料，否則就會認為該 Consumer 已經掛了，會觸發重平衡，它的分割槽會移交給群組中的其它消費者。傳給 poll() 方法的是一個超市時間，用 java.time.Duration 類來表示，如果該引數被設定為 0 ，poll() 方法會立刻返回，否則就會在指定的毫秒數內一直等待 broker 返回資料。
• poll() 方法會返回一個記錄列表。每條記錄都包含了記錄所屬主題的資訊，記錄所在分割槽的資訊、記錄在分割槽中的偏移量，以及記錄的鍵值對。我們一般會遍歷這個列表，逐條處理每條記錄。
• 在退出應用程式之前使用 close() 方法關閉消費者。網路連線和 socket 也會隨之關閉，並立即觸發一次重平衡，而不是等待群組協調器發現它不再發送心跳並認定它已經死亡。

執行緒安全性

在同一個群組中，我們無法讓一個執行緒執行多個消費者，也無法讓多個執行緒安全的共享一個消費者。按照規則，一個消費者使用一個執行緒，如果一個消費者群組中多個消費者都想要執行的話，那麼必須讓每個消費者在自己的執行緒中執行，可以使用 Java 中的 ExecutorService 啟動多個消費者進行進行處理。

消費者配置

到目前為止，我們學習瞭如何使用消費者 API，不過只介紹了幾個最基本的屬性，Kafka 文件列出了所有與消費者相關的配置說明。大部分引數都有合理的預設值，一般不需要修改它們，下面我們就來介紹一下這些引數。

• fetch.min.bytes

該屬性指定了消費者從伺服器獲取記錄的最小位元組數。broker 在收到消費者的資料請求時，如果可用的資料量小於 fetch.min.bytes 指定的大小，那麼它會等到有足夠的可用資料時才把它返回給消費者。這樣可以降低消費者和 broker 的工作負載，因為它們在主題使用頻率不是很高的時候就不用來回處理訊息。如果沒有很多可用資料，但消費者的 CPU 使用率很高，那麼就需要把該屬性的值設得比預設值大。如果消費者的數量比較多，把該屬性的值調大可以降低 broker 的工作負載。

• fetch.max.wait.ms

我們通過上面的 fetch.min.bytes 告訴 Kafka，等到有足夠的資料時才會把它返回給消費者。而 fetch.max.wait.ms 則用於指定 broker 的等待時間，預設是 500 毫秒。如果沒有足夠的資料流入 kafka 的話，消費者獲取的最小資料量要求就得不到滿足，最終導致 500 毫秒的延遲。如果要降低潛在的延遲，就可以把引數值設定的小一些。如果 fetch.max.wait.ms 被設定為 100 毫秒的延遲，而 fetch.min.bytes 的值設定為 1MB，那麼 Kafka 在收到消費者請求後，要麼返回 1MB 的資料，要麼在 100 ms 後返回所有可用的資料。就看哪個條件首先被滿足。

• max.partition.fetch.bytes

該屬性指定了伺服器從每個分割槽裡返回給消費者的最大位元組數。它的預設值時 1MB，也就是說，KafkaConsumer.poll() 方法從每個分割槽裡返回的記錄最多不超過 max.partition.fetch.bytes 指定的位元組。如果一個主題有20個分割槽和5個消費者，那麼每個消費者需要至少4 MB的可用記憶體來接收記錄。在為消費者分配記憶體時，可以給它們多分配一些，因為如果群組裡有消費者發生崩潰，剩下的消費者需要處理更多的分割槽。max.partition.fetch.bytes 的值必須比 broker 能夠接收的最大訊息的位元組數(通過 max.message.size 屬性配置大)，否則消費者可能無法讀取這些訊息，導致消費者一直掛起重試。 在設定該屬性時，另外一個考量的因素是消費者處理資料的時間。消費者需要頻繁的呼叫 poll() 方法來避免會話過期和發生分割槽再平衡，如果單次呼叫poll() 返回的資料太多，消費者需要更多的時間進行處理，可能無法及時進行下一個輪詢來避免會話過期。如果出現這種情況，可以把 max.partition.fetch.bytes 值改小，或者延長會話過期時間。

• session.timeout.ms

這個屬性指定了消費者在被認為死亡之前可以與伺服器斷開連線的時間，預設是 3s。如果消費者沒有在 session.timeout.ms 指定的時間內傳送心跳給群組協調器，就會被認定為死亡，協調器就會觸發重平衡。把它的分割槽分配給消費者群組中的其它消費者，此屬性與 heartbeat.interval.ms 緊密相關。heartbeat.interval.ms 指定了 poll() 方法向群組協調器傳送心跳的頻率，session.timeout.ms 則指定了消費者可以多久不傳送心跳。所以，這兩個屬性一般需要同時修改，heartbeat.interval.ms 必須比 session.timeout.ms 小，一般是 session.timeout.ms 的三分之一。如果 session.timeout.ms 是 3s，那麼 heartbeat.interval.ms 應該是 1s。把 session.timeout.ms 值設定的比預設值小，可以更快地檢測和恢復崩憤的節點，不過長時間的輪詢或垃圾收集可能導致非預期的重平衡。把該屬性的值設定得大一些，可以減少意外的重平衡，不過檢測節點崩潰需要更長的時間。

• auto.offset.reset

該屬性指定了消費者在讀取一個沒有偏移量的分割槽或者偏移量無效的情況下的該如何處理。它的預設值是 latest，意思指的是，在偏移量無效的情況下，消費者將從最新的記錄開始讀取資料。另一個值是 earliest，意思指的是在偏移量無效的情況下，消費者將從起始位置處開始讀取分割槽的記錄。

• enable.auto.commit

我們稍後將介紹幾種不同的提交偏移量的方式。該屬性指定了消費者是否自動提交偏移量，預設值是 true，為了儘量避免出現重複資料和資料丟失，可以把它設定為 false，由自己控制何時提交偏移量。如果把它設定為 true，還可以通過 auto.commit.interval.ms 屬性來控制提交的頻率

• partition.assignment.strategy

我們知道，分割槽會分配給群組中的消費者。PartitionAssignor 會根據給定的消費者和主題，決定哪些分割槽應該被分配給哪個消費者，Kafka 有兩個預設的分配策略Range 和 RoundRobin

• client.id

該屬性可以是任意字串，broker 用他來標識從客戶端傳送過來的訊息，通常被用在日誌、度量指標和配額中

• max.poll.records

該屬性用於控制單次呼叫 call() 方法能夠返回的記錄數量，可以幫你控制在輪詢中需要處理的資料量。

• receive.buffer.bytes 和 send.buffer.bytes

socket 在讀寫資料時用到的 TCP 緩衝區也可以設定大小。如果它們被設定為 -1，就使用作業系統預設值。如果生產者或消費者與 broker 處於不同的資料中心內，可以適當增大這些值，因為跨資料中心的網路一般都有比較高的延遲和比較低的頻寬。

提交和偏移量的概念

特殊偏移

我們上面提到，消費者在每次呼叫poll() 方法進行定時輪詢的時候，會返回由生產者寫入 Kafka 但是還沒有被消費者消費的記錄，因此我們可以追蹤到哪些記錄是被群組裡的哪個消費者讀取的。消費者可以使用 Kafka 來追蹤訊息在分割槽中的位置（偏移量）

消費者會向一個叫做 _consumer_offset 的特殊主題中傳送訊息，這個主題會儲存每次所傳送訊息中的分割槽偏移量，這個主題的主要作用就是消費者觸發重平衡後記錄偏移使用的，消費者每次向這個主題傳送訊息，正常情況下不觸發重平衡，這個主題是不起作用的，當觸發重平衡後，消費者停止工作，每個消費者可能會分到對應的分割槽，這個主題就是讓消費者能夠繼續處理訊息所設定的。

如果提交的偏移量小於客戶端最後一次處理的偏移量，那麼位於兩個偏移量之間的訊息就會被重複處理

如果提交的偏移量大於最後一次消費時的偏移量，那麼處於兩個偏移量中間的訊息將會丟失

既然_consumer_offset 如此重要，那麼它的提交方式是怎樣的呢？下面我們就來說一下

提交方式

KafkaConsumer API 提供了多種方式來提交偏移量

自動提交

最簡單的方式就是讓消費者自動提交偏移量。如果 enable.auto.commit 被設定為true，那麼每過 5s，消費者會自動把從 poll() 方法輪詢到的最大偏移量提交上去。提交時間間隔由 auto.commit.interval.ms 控制，預設是 5s。與消費者裡的其他東西一樣，自動提交也是在輪詢中進行的。消費者在每次輪詢中會檢查是否提交該偏移量了，如果是，那麼就會提交從上一次輪詢中返回的偏移量。

提交當前偏移量

把 auto.commit.offset 設定為 false，可以讓應用程式決定何時提交偏移量。使用 commitSync() 提交偏移量。這個 API 會提交由 poll() 方法返回的最新偏移量，提交成功後馬上返回，如果提交失敗就丟擲異常。

commitSync() 將會提交由 poll() 返回的最新偏移量，如果處理完所有記錄後要確保呼叫了 commitSync()，否則還是會有丟失訊息的風險，如果發生了在均衡，從最近一批訊息到發生在均衡之間的所有訊息都將被重複處理。

非同步提交

非同步提交 commitAsync() 與同步提交 commitSync() 最大的區別在於非同步提交不會進行重試，同步提交會一致進行重試。

同步和非同步組合提交

一般情況下，針對偶爾出現的提交失敗，不進行重試不會有太大的問題，因為如果提交失敗是因為臨時問題導致的，那麼後續的提交總會有成功的。但是如果在關閉消費者或再均衡前的最後一次提交，就要確保提交成功。

因此，在消費者關閉之前一般會組合使用commitAsync和commitSync提交偏移量。

提交特定的偏移量

消費者API允許呼叫 commitSync() 和 commitAsync() 方法時傳入希望提交的 partition 和 offset 的 map，即提交特定的偏移量。

文章參考：

Kafka 消費者解讀

《Kafka-the-definitive》

《極客時間-Kafka核心技術與實戰》

《Kafka 權威指南》

https://docs.confluent.io/current/clients/consumer.html

https://kafka.apache.org/23/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html

關注公眾號獲取更多優質電子書，關注一下你就知道資源是有多好了

相關推薦