kafka：一文讀懂消費(fèi)者背后的那點(diǎn)"貓膩"

前言

經(jīng)過前幾篇文章的介紹，大致了解了生產(chǎn)者背后的運(yùn)行原理。消息有生產(chǎn)就得有人去消費(fèi)，今天我們就來介紹下消費(fèi)端消費(fèi)消息背后發(fā)生的那點(diǎn)事兒。

文章概覽

1. 消費(fèi)者與消費(fèi)組的“父子關(guān)系”。
2. Repartition 觸發(fā)時機(jī)。
3. 消費(fèi)者與 ZK 的關(guān)系。
4. 消費(fèi)端工作流程。
5. 消費(fèi)者的三種消費(fèi)情況。

消費(fèi)者與消費(fèi)組的“父子關(guān)系”

Kafka 消費(fèi)端確保一個 Partition 在一個消費(fèi)者組內(nèi)只能被一個消費(fèi)者消費(fèi)。這句話改怎么理解呢？

1. 在同一個消費(fèi)者組內(nèi)，一個 Partition 只能被一個消費(fèi)者消費(fèi)。
2. 在同一個消費(fèi)者組內(nèi)，所有消費(fèi)者組合起來必定可以消費(fèi)一個 Topic 下的所有 Partition。
3. 在同一個消費(fèi)組內(nèi)，一個消費(fèi)者可以消費(fèi)多個 Partition 的信息。
4. 在不同消費(fèi)者組內(nèi)，同一個分區(qū)可以被多個消費(fèi)者消費(fèi)。
5. 每個消費(fèi)者組一定會完整消費(fèi)一個 Topic 下的所有 Partition。

消費(fèi)組存在的意義

了解了消費(fèi)者與消費(fèi)組的關(guān)系后，有朋友會比較疑惑消費(fèi)者組有啥實(shí)際存在的意義呢？或者說消費(fèi)組的作用是什么？

作者對消費(fèi)組的作用歸結(jié)了如下兩點(diǎn)。

1. 在實(shí)際生產(chǎn)中，對于同一個 Topic，可能有 A、B、C 等 N 個消費(fèi)方想要消費(fèi)。比如一份用戶點(diǎn)擊日志，A 消費(fèi)方想用來做一個用戶近 N 天點(diǎn)擊過哪些商品；B 消費(fèi)方想用來做一個用戶近 N 天點(diǎn)擊過前 TopN 個相似的商品；C 消費(fèi)方想用來做一個根據(jù)用戶點(diǎn)擊過的商品推薦相關(guān)周邊的商品需求。對于多應(yīng)用場景，就可以使用消費(fèi)組來隔離不同的業(yè)務(wù)使用場景，從而達(dá)到一個 Topic 可以被多個消費(fèi)組重復(fù)消費(fèi)的目的。
2. 消費(fèi)組與 Partition 的消費(fèi)進(jìn)度綁定。當(dāng)有新的消費(fèi)者加入或者有消費(fèi)者從消費(fèi)組退出時，會觸發(fā)消費(fèi)組的 Repartition 操作（后面會詳細(xì)介紹 Repartition）；在 Repartition 前，Partition1 被消費(fèi)組的消費(fèi)者 A 進(jìn)行消費(fèi)，Repartition 后，Partition1 消費(fèi)組的消費(fèi)者 B 進(jìn)行消費(fèi)，為了避免消息被重復(fù)消費(fèi)，需要從消費(fèi)組記錄的 Partition 消費(fèi)進(jìn)度讀取當(dāng)前消費(fèi)到的位置（即 OffSet 位置），然后在繼續(xù)消費(fèi)，從而達(dá)到消費(fèi)者的平滑遷移，同時也提高了系統(tǒng)的可用性。

Repartition 觸發(fā)時機(jī)

使用過 Kafka 消費(fèi)者客戶端的同學(xué)肯定知道，消費(fèi)者組內(nèi)偶爾會觸發(fā) Repartition 操作，所謂 Repartition 即 Partition 在某些情況下重新被分配給參與消費(fèi)的消費(fèi)者。基本可以分為如下幾種情況。

1. 消費(fèi)組內(nèi)某消費(fèi)者宕機(jī)，觸發(fā) Repartition 操作，如下圖所示。

2. 消費(fèi)組內(nèi)新增消費(fèi)者，觸發(fā) Repartition 操作，如下圖所示。一般這種情況是為了提高消費(fèi)端的消費(fèi)能力，從而加快消費(fèi)進(jìn)度。

3. Topic 下的 Partition 增多，觸發(fā) Repartition 操作，如下圖所示。一般這種調(diào)整 Partition 個數(shù)的情況也是為了提高消費(fèi)端消費(fèi)速度的，因?yàn)楫?dāng)消費(fèi)者個數(shù)大于等于 Partition 個數(shù)時，在增加消費(fèi)者個數(shù)是沒有用的（原因是：在一個消費(fèi)組內(nèi)，消費(fèi)者:Partition = 1:N，當(dāng) N 小于 1 時，相當(dāng)于消費(fèi)者過剩了），所以一方面增加 Partition 個數(shù)同時增加消費(fèi)者個數(shù)可以提高消費(fèi)端的消費(fèi)速度。

消費(fèi)者與 ZK 的關(guān)系

眾所周知，ZK 不僅保存了消費(fèi)者消費(fèi) partition 的進(jìn)度，同時也保存了消費(fèi)組的成員列表、partition 的所有者。消費(fèi)者想要消費(fèi) Partition，需要從 ZK 中獲取該消費(fèi)者對應(yīng)的分區(qū)信息及當(dāng)前分區(qū)對應(yīng)的消費(fèi)進(jìn)度，即 OffSert 信息。那么 Partition 應(yīng)該由那個消費(fèi)者進(jìn)行消費(fèi)，決定因素有哪些呢？從之前的圖中不難得出，兩個重要因素分別是：消費(fèi)組中存活的消費(fèi)者列表和 Topic 對應(yīng)的 Partition 列表。通過這兩個因素結(jié)合 Partition 分配算法，即可得出消費(fèi)者與 Partition 的對應(yīng)關(guān)系，然后將信息存儲到 ZK 中。Kafka 有高級 API 和低級 API，如果不需要操作 OffSet 偏移量的提交，可通過高級 API 直接使用，從而降低使用者的難度。對于一些比較特殊的使用場景，比如想要消費(fèi)特定 Partition 的信息，Kafka 也提供了低級 API 可進(jìn)行手動操作。

消費(fèi)端工作流程

在介紹消費(fèi)端工作流程前，先來熟悉一下用到的一些組件。

• KakfaConsumer：消費(fèi)端，用于啟動消費(fèi)者進(jìn)程來消費(fèi)消息。
• ConsumerConfig：消費(fèi)端配置管理，用于給消費(fèi)端配置相關(guān)參數(shù)，比如指定 Kafka 集群，設(shè)置自動提交和自動提交時間間隔等等參數(shù)，都由其來管理。
• ConsumerConnector：消費(fèi)者連接器，通過消費(fèi)者連接器可以獲得 Kafka 消息流，然后通過消息流就能獲得消息從而使得客戶端開始消費(fèi)消息。

以上三者之間的關(guān)系可以概括為：消費(fèi)端使用消費(fèi)者配置管理創(chuàng)建出了消費(fèi)者連接器，通過消費(fèi)者連接器創(chuàng)建隊(duì)列（這個隊(duì)列的作用也是為了緩存數(shù)據(jù)），其中隊(duì)列中的消息由專門的拉取線程從服務(wù)端拉取然后寫入，最后由消費(fèi)者客戶端輪詢隊(duì)列中的消息進(jìn)行消費(fèi)。具體操作流程如下圖所示。

我們在從消費(fèi)者與 ZK 的角度來看看其工作流程是什么樣的？

從上圖可以看出，首先拉取線程每拉取一次消息，同步更新一次拉取狀態(tài)，其作用是為了下一次拉取消息時能夠拉取到最新產(chǎn)生的消息；拉取線程將拉取到的消息寫入到隊(duì)列中等待消費(fèi)消費(fèi)線程去真正讀取處理。消費(fèi)線程以輪詢的方式持續(xù)讀取隊(duì)列中的消息，只要發(fā)現(xiàn)隊(duì)列中有消息就開始消費(fèi)，消費(fèi)完消息后更新消費(fèi)進(jìn)度，此處需要注意的是，消費(fèi)線程不是每次都和 ZK 同步消費(fèi)進(jìn)度，而是將消費(fèi)進(jìn)度暫時寫入本地。這樣做的目的是為了減少消費(fèi)者與 ZK 的頻繁同步消息，從而降低 ZK 的壓力。

消費(fèi)者的三種消費(fèi)情況

消費(fèi)者從服務(wù)端的 Partition 上拉取到消息，消費(fèi)消息有三種情況，分別如下：

1. 至少一次。即一條消息至少被消費(fèi)一次，消息不可能丟失，但是可能會被重復(fù)消費(fèi)。
2. 至多一次。即一條消息最多可以被消費(fèi)一次，消息不可能被重復(fù)消費(fèi)，但是消息有可能丟失。
3. 正好一次。即一條消息正好被消費(fèi)一次，消息不可能丟失也不可能被重復(fù)消費(fèi)。

1.至少一次

消費(fèi)者讀取消息，先處理消息，在保存消費(fèi)進(jìn)度。消費(fèi)者拉取到消息，先消費(fèi)消息，然后在保存偏移量，當(dāng)消費(fèi)者消費(fèi)消息后還沒來得及保存偏移量，則會造成消息被重復(fù)消費(fèi)。如下圖所示：

2.至多一次

消費(fèi)者讀取消息，先保存消費(fèi)進(jìn)度，在處理消息。消費(fèi)者拉取到消息，先保存了偏移量，當(dāng)保存了偏移量后還沒消費(fèi)完消息，消費(fèi)者掛了，則會造成未消費(fèi)的消息丟失。如下圖所示：

3.正好一次

正好消費(fèi)一次的辦法可以通過將消費(fèi)者的消費(fèi)進(jìn)度和消息處理結(jié)果保存在一起。只要能保證兩個操作是一個原子操作，就能達(dá)到正好消費(fèi)一次的目的。通常可以將兩個操作保存在一起，比如 HDFS 中。正好消費(fèi)一次流程如下圖所示。

總結(jié)

本文講解了消費(fèi)組與消費(fèi)者之間的關(guān)系，及 Repartition 的觸發(fā)時機(jī)，然后講述了消費(fèi)端的基本工作流程，最后提出了一條消息被重復(fù)消費(fèi)的幾種情況。下篇文章我們來講講消息在服務(wù)端是怎么存儲的，敬請期待。