前言

消息的消費(fèi)一般有兩種模式，推模式和拉模式。推模式是服務(wù)端主動(dòng)將消息推送給消費(fèi)者，而拉模式是消費(fèi)者主動(dòng)向服務(wù)端發(fā)起請(qǐng)求來(lái)拉取消息。kakfa采用的是拉模式，這樣可以很好的控制消費(fèi)速率。那么kafka消費(fèi)的具體工作流程是什么樣的呢？kafka的位移管理又是怎么樣的呢？

消費(fèi)者消費(fèi)規(guī)則

kafka是以消費(fèi)者組進(jìn)行消費(fèi)，一個(gè)消費(fèi)者組，由多個(gè)consumer組成，他們和topic的消費(fèi)規(guī)則如下：

• topic的一個(gè)分區(qū)只能被消費(fèi)組中的一個(gè)消費(fèi)者消費(fèi)。
• 消費(fèi)者組中的一個(gè)消費(fèi)者可以消費(fèi)topic一個(gè)或者多個(gè)分區(qū)。

通過(guò)這種分組、分區(qū)的消費(fèi)方式，可以提高消費(fèi)者的吞吐量，同時(shí)也能夠?qū)崿F(xiàn)消息的發(fā)布/訂閱模式和點(diǎn)對(duì)點(diǎn)兩種模式。

消費(fèi)者整體工作流程

消費(fèi)者消費(fèi)總體分為兩個(gè)步驟，第一步是制定消費(fèi)的方案，就是這個(gè)組下哪個(gè)消費(fèi)者消費(fèi)哪個(gè)分區(qū)，第二個(gè)是建立網(wǎng)絡(luò)連接，獲取消息數(shù)據(jù)。

一、制定消費(fèi)方案

1. 消費(fèi)者consumerA，consumerB, consumerC向kafka集群中的協(xié)調(diào)器coordinator發(fā)送JoinGroup的請(qǐng)求。coordinator主要是用來(lái)輔助實(shí)現(xiàn)消費(fèi)者組的初始化和分區(qū)的分配。

• coordinator老大節(jié)點(diǎn)選擇 = groupid的hashcode值 % 50（ __consumer_offsets內(nèi)置主題位移的分區(qū)數(shù)量）例如： groupid的hashcode值 為1，1% 50 = 1，那么__consumer_offsets 主題的1號(hào)分區(qū)，在哪個(gè)broker上，就選擇這個(gè)節(jié)點(diǎn)的coordinator作為這個(gè)消費(fèi)者組的老大。消費(fèi)者組下的所有的消費(fèi)者提交offset的時(shí)候就往這個(gè)分區(qū)去提交offset。

2. 選出一個(gè) consumer作為消費(fèi)中的leader，比如上圖中的consumerB。
3. 消費(fèi)者leader制定出消費(fèi)方案，比如誰(shuí)來(lái)消費(fèi)哪個(gè)分區(qū)等，有Range分區(qū)策略、RoundRobin分區(qū)策略等。
4. 把消費(fèi)方案告訴給coordinator
5. 最后coordinator就把消費(fèi)方案下發(fā)給各個(gè)consumer, 圖中只畫了一條線，實(shí)際上是會(huì)下發(fā)到各個(gè)consumer。

二、消費(fèi)者消費(fèi)細(xì)節(jié)

現(xiàn)在已經(jīng)初始化消費(fèi)者組信息，知道哪個(gè)消費(fèi)者消費(fèi)哪個(gè)分區(qū)，接著我們來(lái)看看消費(fèi)者細(xì)節(jié)。

1. 消費(fèi)者創(chuàng)建一個(gè)網(wǎng)絡(luò)連接客戶端ConsumerNetworkClient, 發(fā)送消費(fèi)請(qǐng)求，可以進(jìn)行如下配置：

• fetch.min.bytes: 每批次最小抓取大小，默認(rèn)1字節(jié)
• fetch.max.bytes: 每批次最大抓取大小，默認(rèn)50M
• fetch.max.wait.ms：最大超時(shí)時(shí)間，默認(rèn)500ms

2. 發(fā)送請(qǐng)求到kafka集群
3. 獲取數(shù)據(jù)成功，會(huì)將數(shù)據(jù)保存到completedFetches隊(duì)列中
4. 消費(fèi)者從隊(duì)列中抓取數(shù)據(jù)，根據(jù)配置max.poll.records一次拉取數(shù)據(jù)返回消息的最大條數(shù)，默認(rèn)500條。
5. 獲取到數(shù)據(jù)后，經(jīng)過(guò)反序列化器、攔截器后，得到最終的消息。
6. 最后一步是提交保存消費(fèi)的位移offset，也就是這個(gè)消費(fèi)者消費(fèi)到什么位置了，這樣下次重啟也可以繼續(xù)從這個(gè)位置開始消費(fèi)，關(guān)于offset的管理后面詳細(xì)介紹。

消費(fèi)者分區(qū)策略

前面簡(jiǎn)單提到了消費(fèi)者組初始化的時(shí)候會(huì)對(duì)分區(qū)進(jìn)行分配，那么具體的分配策略是什么呢，也就是哪個(gè)消費(fèi)者消費(fèi)哪個(gè)分區(qū)數(shù)據(jù)？

kafka有四種主流的分區(qū)分配策略： Range、RoundRobin、Sticky、CooperativeSticky?？梢酝ㄟ^(guò)配置參數(shù)partition.assignment.strategy，修改分區(qū)的分配策略。默認(rèn)策略是Range + CooperativeSticky。Kafka可以同時(shí)使用多個(gè)分區(qū)分配策略。

Range 分區(qū)策略

• Range分區(qū) 是對(duì)每個(gè) topic 而言的。對(duì)同一個(gè) topic 里面的分區(qū)按照序號(hào)進(jìn)行排序，并對(duì)消費(fèi)者按照字母順序進(jìn)行排序。
• 通過(guò) partitions數(shù)/consumer數(shù) 來(lái)決定每個(gè)消費(fèi)者應(yīng)該消費(fèi)幾個(gè)分區(qū)。如果除不盡，那么前面幾個(gè)消費(fèi)者將會(huì)多消費(fèi) 1 個(gè)分區(qū)。

如上圖所示：有 7 個(gè)分區(qū)，3 個(gè)消費(fèi)者，排序后的分區(qū)將會(huì)是0,1,2,3,4,5,6；消費(fèi)者排序完之后將會(huì)是C0,C1,C2。7/3 = 2 余 1 ，除不盡，那么 消費(fèi)者 C0 便會(huì)多消費(fèi) 1 個(gè)分區(qū)。 8/3=2余2，除不盡，那么C0和C1分別多消費(fèi)一個(gè)。

這種方式容易造成數(shù)據(jù)傾斜！如果有 N 多個(gè) topic，那么針對(duì)每個(gè) topic，消費(fèi)者 C0都將多消費(fèi) 1 個(gè)分區(qū)，topic越多，C0消費(fèi)的分區(qū)會(huì)比其他消費(fèi)者明顯多消費(fèi) N 個(gè)分區(qū)。

RoundRobin 分區(qū)策略

RoundRobin 針對(duì)集群中所有topic而言，RoundRobin 輪詢分區(qū)策略，是把所有的 partition 和所有的consumer 都列出來(lái)，然后按照 hashcode 進(jìn)行排序，最后通過(guò)輪詢算法來(lái)分配 partition 給到各個(gè)消費(fèi)者。

Sticky 和Cooperative Sticky分區(qū)策略

Sticky是粘性的意思，它是從 0.11.x 版本開始引入這種分配策略，首先會(huì)盡量均衡的放置分區(qū)到消費(fèi)者上面，在出現(xiàn)同一消費(fèi)者組內(nèi)消費(fèi)者出現(xiàn)問(wèn)題的時(shí)候，在rebalance會(huì)盡量保持原有分配的分區(qū)不變化，這樣可以節(jié)省開銷。

Cooperative Sticky和Sticky類似，但是它會(huì)將原來(lái)的一次大規(guī)模rebalance操作，拆分成了多次小規(guī)模的rebalance，直至最終平衡完成，所以體驗(yàn)上會(huì)更好。

關(guān)于什么是rebalance繼續(xù)往下看你就知道了。

消費(fèi)者再均衡

上面也提到了rebalance，也就是再均衡。當(dāng)kafka發(fā)生下面的情況會(huì)進(jìn)行在均衡，也就是重新給消費(fèi)者分配分區(qū)：

• 有新的消費(fèi)者加入消費(fèi)組。 ?
• 有消費(fèi)者宕機(jī)下線，消費(fèi)者并不一定需要真正下線，例如遇到長(zhǎng)時(shí)間的 GC 、網(wǎng)絡(luò)延遲導(dǎo)致消費(fèi)者長(zhǎng)時(shí)間未向Group Coordinator發(fā)送心跳等情況時(shí)，GroupCoordinator 會(huì)認(rèn)為消費(fèi)者己下線。 ?
• 有消費(fèi)者主動(dòng)退出消費(fèi)組。
• 消費(fèi)組所對(duì)應(yīng)的Group Coorinator節(jié)點(diǎn)發(fā)生了變更。 ?
• 消費(fèi)組內(nèi)所訂閱的任一主題或者主題的分區(qū)數(shù)量發(fā)生變化。

消費(fèi)者位移offset管理

消費(fèi)者需要保存當(dāng)前消費(fèi)到分區(qū)的什么位置了，這樣哪怕消費(fèi)者故障，重啟后也能繼續(xù)消費(fèi)，這就是消費(fèi)者的維護(hù)offset管理。

一、消費(fèi)者位移offset存儲(chǔ)位置

消費(fèi)者位移offset存儲(chǔ)在哪呢？

• kafka0.9版本之前，consumer默認(rèn)將offset保存在Zookeeper中
• 從0.9版本開始，consumer默認(rèn)將offset保存在Kafka一個(gè)內(nèi)置的topic中，該topic為__consumer_offsets，這樣可以大量減少和zookeeper的交互。
• __consumer_offsets 主題里面采用 key 和 value 的方式存儲(chǔ)數(shù)據(jù)。key 是 group.id+topic+分區(qū)號(hào)，value 就是當(dāng)前 offset 的值。

如何查看__consumer_offsets主題內(nèi)容？

• 在配置文件 config/consumer.properties 中添加配置 exclude.internal.topics=false，默認(rèn)是 true，表示不能消費(fèi)系統(tǒng)主題。為了查看該系統(tǒng)主題數(shù)據(jù)，所以該參數(shù)修改為 false。
• 查看消費(fèi)者消費(fèi)主題__consumer_offsets。

bin/kafka-console-consumer.sh --topic 
__consumer_offsets --bootstrap-server hadoop102:9092 --
consumer.config config/consumer.properties --formatter 
"kafka.coordinator.group.GroupMetadataManager$OffsetsMessageForm
atter" --from-beginning
## topic1 1號(hào)分區(qū)
[offset,topic1,1]::OffsetAndMetadata(offset=7, 
leaderEpoch=Optional[0], metadata=, commitTimestamp=1622442520203, 
expireTimestamp=None)
## topic1 0號(hào)分區(qū)
[offset,topic1,0]::OffsetAndMetadata(offset=8, 
leaderEpoch=Optional[0], metadata=, commitTimestamp=1622442520203, 
expireTimestamp=None)

二、消費(fèi)者位移offset提交保存模式

消費(fèi)者是如何提交保存位移offset呢？

自動(dòng)提交

為了使我們能夠?qū)Ｗ⒂谧约旱臉I(yè)務(wù)邏輯，kafka默認(rèn)提供了自動(dòng)提交offset的功能。這個(gè)由消費(fèi)者客戶端參數(shù) enable.auto.commit 配置， 默認(rèn)值為 true 。當(dāng)然這個(gè)默認(rèn)的自動(dòng)提交不是每消費(fèi)一條消息就提交一次，而是定期提交，這個(gè)定期的周期時(shí)間由客戶端參數(shù) auto.commit.interval.ms 配置，默認(rèn)值為 5 秒。

• 消費(fèi)者每隔 5 秒會(huì)將拉取到的每個(gè)分區(qū)中最大的消息位移進(jìn)行提交。
• 自動(dòng)位移提交 的動(dòng)作是在 poll() 方法的邏輯里完成的，在每次真正向服務(wù)端發(fā)起拉取請(qǐng)求之前會(huì)檢查是否可以進(jìn)行位移提交，如果可以，那么就會(huì)提交上一次輪詢的位移。

自動(dòng)提交會(huì)帶來(lái)什么問(wèn)題？

自動(dòng)提交消費(fèi)位移的方式非常簡(jiǎn)便，但會(huì)帶來(lái)是重復(fù)消費(fèi)的問(wèn)題。

假設(shè)剛剛提交完一次消費(fèi)位移，然后拉取一批消息進(jìn)行消費(fèi)，在下一次自動(dòng)提交消費(fèi)位移之前，消費(fèi)者崩潰了，那么又得從上一次位移提交的地方重新開始消費(fèi)，這樣便發(fā)生了重復(fù)消費(fèi)的現(xiàn)象。

我們可以通過(guò)減小位移提交的時(shí)間間隔來(lái)減小重復(fù)消息的窗口大小，但這樣 并不能避免重復(fù)消費(fèi)的發(fā)送，而且也會(huì)使位移提交更加頻繁。

手動(dòng)提交

很多時(shí)候并不是說(shuō)拉取到消息就算消費(fèi)完成，而是需要將消息寫入數(shù)據(jù)庫(kù)、寫入本地緩存，或者是更 加復(fù)雜的業(yè)務(wù)處理。在這些場(chǎng)景下，所有的業(yè)務(wù)處理完成才能認(rèn)為消息被成功消費(fèi)。手動(dòng)的提交方式可以讓開發(fā)人員根據(jù)程序的邏輯在合適的地方進(jìn)行位移提交。

// 是否自動(dòng)提交 offset
 properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

手動(dòng)提交可以細(xì)分為同步提交和異步提交，對(duì)應(yīng)于 KafkaConsumer 中的 commitSync()和 commitAsync()兩種類型的方法。

• 同步提交方式

同步提交會(huì)阻塞當(dāng)前線程，一直到提交成功，并且會(huì)自動(dòng)失敗重試（由不可控因素導(dǎo)致，也會(huì)出現(xiàn)提交失敗），它必須等待offset提交完畢，再去消費(fèi)下一批數(shù)據(jù)。

// 同步提交 offset
consumer.commitSync();

• 異步提交方式

異步提交則沒有失敗重試機(jī)制，故有可能提交失敗。它發(fā)送完提交offset請(qǐng)求后，就開始消費(fèi)下一批數(shù)據(jù)了。

// 異步提交 offset
consumer.commitAsync();

那么手動(dòng)提交會(huì)帶來(lái)什么問(wèn)題呢？可能會(huì)出現(xiàn)"漏消息"的情況。

設(shè)置offset為手動(dòng)提交，當(dāng)offset被提交時(shí)，數(shù)據(jù)還在內(nèi)存中未落盤，此時(shí)剛好消費(fèi)者線程被kill掉，那么offset已經(jīng)提交，但是數(shù)據(jù)未處理，導(dǎo)致這部分內(nèi)存中的數(shù)據(jù)丟失。

我們可以通過(guò)消費(fèi)者事物來(lái)解決這樣的問(wèn)題。

其實(shí)無(wú)論是手動(dòng)提交還是自動(dòng)提交，都有可能出現(xiàn)消息重復(fù)和是漏消息，與我們的編程模型有關(guān)，需要我們開發(fā)的時(shí)候根據(jù)消息的重要程度來(lái)選擇合適的消費(fèi)方案。

消費(fèi)者API

一個(gè)正常的消費(fèi)邏輯需要具備以下幾個(gè)步驟：

（1）配置消費(fèi)者客戶端參數(shù)及創(chuàng)建相應(yīng)的消費(fèi)者實(shí)例；

（2）訂閱主題；

（3）拉取消息并消費(fèi)；

（4）提交消費(fèi)位移 offset；

（5）關(guān)閉消費(fèi)者實(shí)例。

public class MyConsumer { 
    public static void main(String[] args) { 
        Properties props = new Properties(); 
        // 定義 kakfa 服務(wù)的地址，不需要將所有 broker 指定上 
        props.put("bootstrap.servers", "doitedu01:9092"); 
        // 制定 consumer group 
        props.put("group.id", "g1"); 
        // 是否自動(dòng)提交 offset 
        props.put("enable.auto.commit", "true"); 
        // 自動(dòng)提交 offset 的時(shí)間間隔 
        props.put("auto.commit.interval.ms", "1000");
        // key 的反序列化類 
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); 
        // value 的反序列化類
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); 
        // 如果沒有消費(fèi)偏移量記錄，則自動(dòng)重設(shè)為起始 offset：latest, earliest, none 
        props.put("auto.offset.reset","earliest");
    
    	// 定義 consumer 
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); 
        // 消費(fèi)者訂閱的 topic, 可同時(shí)訂閱多個(gè) 
        consumer.subscribe(Arrays.asList("first", "test","test1"));
        while (true) { 
            // 讀取數(shù)據(jù)，讀取超時(shí)時(shí)間為 100ms 
            ConsumerRecords<String, String> records = consumer.poll(100); 
            for (ConsumerRecord<String, String> record : records) 
            	System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); 
        } 
	} 
}

訂閱主題

• 指定集合方式訂閱主題

consumer.subscribe(Arrays.asList(topicl )); 
consumer subscribe(Arrays.asList(topic2))

• 正則方式訂閱主題

如果消費(fèi)者采用的是正則表達(dá)式的方式（subscribe(Pattern)）訂閱， 在之后的過(guò)程中，如果 有人又創(chuàng)建了新的主題，并且主題名字與正表達(dá)式相匹配，那么這個(gè)消費(fèi)者就可以消費(fèi)到 新添加的主題中的消息。

consumer.subscribe(Pattern.compile ("topic.*" ));

• 訂閱主題指定分區(qū)

消費(fèi)者不僅可以通過(guò) KafkaConsumer.subscribe()方法訂閱主題，還可直接訂閱某些主題的指定分區(qū)。

consumer.assign(Arrays.asList(new TopicPartition ("tpc_1" , 0),new TopicPartition(“tpc_2”,1))) ;

取消訂閱

通過(guò)unsubscribe()方法采取消主題的訂閱。

consumer.unsubscribe();

poll()拉取消息

kafka 中的消息消費(fèi)是一個(gè)不斷輪詢的過(guò)程，消費(fèi)者所要做的就是重復(fù)地調(diào)用 poll() 方法， poll()方法返回的是所訂閱的主題（分區(qū)）上的一組消息。

對(duì)于 poll () 方法而言，如果某些分區(qū)中沒有可供消費(fèi)的消息，那么此分區(qū)對(duì)應(yīng)的消息拉取的結(jié)果就為空。

public ConsumerRecords<K, V> poll(final Duration timeout)

超時(shí)時(shí)間參數(shù) timeout ，用來(lái)控制 poll() 方法的阻塞時(shí)間，在消費(fèi)者的緩沖區(qū)里沒有可用數(shù)據(jù)時(shí)會(huì)發(fā)生阻塞。

指定位移消費(fèi)

有些時(shí)候，我們需要一種更細(xì)粒度的掌控，可以讓我們從特定的位移處開始拉取消息，而 KafkaConsumer 中的 seek（ 方法正好提供了這個(gè)功能，讓我們可以追前消費(fèi)或回溯消費(fèi)。

public void seek(TopicPartiton partition,long offset)

消費(fèi)者重要參數(shù)

最后我們總結(jié)一下消費(fèi)者中重要的參數(shù)配置。

總結(jié)

kafka消費(fèi)是很重要的一個(gè)環(huán)節(jié)，本文總結(jié)kafka消費(fèi)者的一些重要機(jī)制，包括消費(fèi)者的整個(gè)流程，消費(fèi)的分區(qū)策略，消費(fèi)的再平衡以及消費(fèi)的位移管理。在明白這些機(jī)制以后，簡(jiǎn)單講解了如何使用消費(fèi)者consumer的API以及消費(fèi)者中重要的參數(shù)。