[[442535]]

本文轉(zhuǎn)載自微信公眾號(hào)「零零后程序員小三」，作者003 。轉(zhuǎn)載本文請(qǐng)聯(lián)系零零后程序員小三公眾號(hào)。

什么是Kafka

Kafka是最初由Linkedin公司開發(fā)，Linkedin于2010年貢獻(xiàn)給了Apache基金會(huì)并成為頂級(jí)開源項(xiàng)目，也是一個(gè)開源【分布式流處理平臺(tái)】，由Scala和Java編寫，(也當(dāng)做MQ系統(tǒng)，但不是純粹的消息系統(tǒng))

目前 Kafka 已經(jīng)定位為一個(gè)分布式流式處理平臺(tái)，它以高吞吐、可持久化、可水平擴(kuò)展、支持流數(shù)據(jù)處理等多種特性而被廣泛使用。目前越來越多的開源分布式處理系統(tǒng)如 Cloudera、Storm、Spark、Flink 等都支持與 Kafka 集成

生產(chǎn)者與消費(fèi)者機(jī)制

在Kafka中，生產(chǎn)者(producer)將消息發(fā)送給Broker，Broker將生產(chǎn)者發(fā)送的消息存儲(chǔ)到磁盤當(dāng)中，而消費(fèi)者(Consumer)負(fù)責(zé)從Broker訂閱并且消費(fèi)消息，消費(fèi)者(Consumer)使用pull這種模式從服務(wù)端拉取消息。而zookeeper是負(fù)責(zé)整個(gè)集群的元數(shù)據(jù)管理與控制器的選舉。具體如下圖所示。

Kafka的producer生產(chǎn)者發(fā)送到Broker分區(qū)策略

發(fā)布訂閱的對(duì)象是主題(Topic)，生產(chǎn)者將消息發(fā)送到指定的主題，消費(fèi)者再對(duì)負(fù)責(zé)訂閱的主題來進(jìn)行消費(fèi)。在Kafka里的分區(qū)機(jī)制是怎么樣的呢?它是將主題劃分成了多個(gè)分區(qū)(partition)，每一個(gè)分區(qū)又有多個(gè)副本，在同一個(gè)主題下的不同分區(qū)里的消息也是不一樣的。在生產(chǎn)者生產(chǎn)出來的每一條消息都只會(huì)發(fā)送到一個(gè)分區(qū)里，Kafka里的分區(qū)編號(hào)都是從0開始的，如果生產(chǎn)者向兩個(gè)分區(qū)的主題發(fā)送一條消息，那么這條不是在分區(qū)0里，就是在分區(qū)1里。

那么如何指定消息到指定的分區(qū)里呢?

這時(shí)候就可以看看生產(chǎn)者的發(fā)送邏輯了，在此之前我們需要知道一個(gè)叫ProducerRecord的玩意，這個(gè)是什么?

ProducerRecord就是發(fā)送給Broker的Key/value鍵值對(duì)，封裝基礎(chǔ)數(shù)據(jù)信息，簡(jiǎn)稱為PR。

內(nèi)部結(jié)構(gòu)

Topic（名字） 
PartitionID（可選） 
Key（可選） 
Value 

生產(chǎn)者發(fā)送邏輯

1、如果指定了Partition ID的話，那么PR就會(huì)被發(fā)送到指定的Partition里。

2、如果沒有指定Partition ID，但是指定了Key，那么PR就會(huì)按照hash(key)發(fā)送到相對(duì)應(yīng)的Partition里

3、如果沒有指定Partition ID，也沒有指定Key，PR就會(huì)使用默認(rèn)的round-robin輪訓(xùn)發(fā)送到每一個(gè)Partition里(消費(fèi)者消費(fèi)partition分區(qū)默認(rèn)是range模式)

4、如果同時(shí)指定了Partition ID與Key的話，PR只會(huì)發(fā)送到指定的Partition(這時(shí)候的Key不起作用，代碼邏輯決定)

注意：Partition有多個(gè)副本，但是的話只有一個(gè)replicationLeader來負(fù)責(zé)這個(gè)Partition和生產(chǎn)者消費(fèi)者交互

生產(chǎn)者到Broker的發(fā)送流程

kafka的客戶端發(fā)送數(shù)據(jù)到服務(wù)器里(并不是來一條發(fā)一條)，會(huì)經(jīng)過內(nèi)存的緩沖區(qū)，在通過KafkaProducer發(fā)送出去的消息都是先進(jìn)入到客戶端的本地緩存里，然后再把消息收集到Batch里，再一次性的發(fā)送到Broker上去的，這樣的性能才可能提高。

生產(chǎn)者常見的配置

#kafka地址,即broker地址 
bootstrap.servers   
 
#當(dāng)producer向leader發(fā)送數(shù)據(jù)時(shí)，可以通過request.required.acks參數(shù)來設(shè)置數(shù)據(jù)可靠性的級(jí)別,分別是0, 1，all。 
acks 
 
#請(qǐng)求失敗，生產(chǎn)者會(huì)自動(dòng)重試，指定是0次，如果啟用重試，則會(huì)有重復(fù)消息的可能性 
retries 
 
#每個(gè)分區(qū)未發(fā)送消息總字節(jié)大小,單位：字節(jié)，超過設(shè)置的值就會(huì)提交數(shù)據(jù)到服務(wù)端，默認(rèn)值是16KB 
batch.size 
 
# 默認(rèn)值就是0，消息是立刻發(fā)送的，即便batch.size緩沖空間還沒有滿，如果想減少請(qǐng)求的數(shù)量，可以設(shè)置 linger.ms 大于#0，即消息在緩沖區(qū)保留的時(shí)間，超過設(shè)置的值就會(huì)被提交到服務(wù)端 
# 通俗解釋是，本該早就發(fā)出去的消息被迫至少等待了linger.ms時(shí)間，相對(duì)于這時(shí)間內(nèi)積累了更多消息，批量發(fā)送 減少請(qǐng)求 
#如果batch被填滿或者linger.ms達(dá)到上限，滿足其中一個(gè)就會(huì)被發(fā)送 
linger.ms 
 
# buffer.memory的用來約束Kafka Producer能夠使用的內(nèi)存緩沖的大小的，默認(rèn)值32MB。 
# 如果buffer.memory設(shè)置的太小，可能導(dǎo)致消息快速的寫入內(nèi)存緩沖里，但Sender線程來不及把消息發(fā)送到Kafka服務(wù)器 
# 會(huì)造成內(nèi)存緩沖很快就被寫滿，而一旦被寫滿，就會(huì)阻塞用戶線程，不讓繼續(xù)往Kafka寫消息了 
# buffer.memory要大于batch.size，否則會(huì)報(bào)申請(qǐng)內(nèi)存不足的錯(cuò)誤，不要超過物理內(nèi)存，根據(jù)實(shí)際情況調(diào)整 
buffer.memory 
 
# key的序列化器，將用戶提供的 key和value對(duì)象ProducerRecord 進(jìn)行序列化處理，key.serializer必須被設(shè)置，即使 
#消息中沒有指定key，序列化器必須是一個(gè)實(shí)現(xiàn)org.apache.kafka.common.serialization.Serializer接口的類，將#key序列化成字節(jié)數(shù)組。 
key.serializer 
value.serializer  

Kafka的Consumer消費(fèi)者機(jī)制和分區(qū)策略講解

消費(fèi)者根據(jù)什么模式從broker獲取數(shù)據(jù)的?為什么是pull模式，而不是broker主動(dòng)push?

答案可以看文章一開始的圖，消費(fèi)者是采用Pull拉取方式從broker的partition獲取數(shù)據(jù)，那為什么是pull模式而不是push呢?pull模式可以根據(jù)消費(fèi)者的消費(fèi)能力來進(jìn)行自己調(diào)整，不同的消費(fèi)者性能不一樣。如果broker沒有數(shù)據(jù)的話，消費(fèi)者可以配置timeout的世界，進(jìn)行阻塞等待一段時(shí)間后再返回。但如果是broker主動(dòng)Push，push的優(yōu)點(diǎn)是可以快速的處理消息，但是容易對(duì)消費(fèi)者處理不過來，造成消息的堆積和延遲。

消費(fèi)者從哪個(gè)分區(qū)進(jìn)行消費(fèi)?

我們知道一個(gè)topic有多個(gè)partition，一個(gè)消費(fèi)者組里面就有多個(gè)消費(fèi)者，那是怎么分配的呢?一個(gè)主題topic可以有多個(gè)消費(fèi)者，因?yàn)槔锩嬗卸鄠€(gè)partition分區(qū)(leader分區(qū))，一個(gè)partition leader可以由一個(gè)消費(fèi)組里的一個(gè)消費(fèi)者來消費(fèi)。

那么消費(fèi)者從哪個(gè)分區(qū)來進(jìn)行消費(fèi)呢?

策略一、round-robin (RoundRobinAssignor非默認(rèn)策略)輪訓(xùn)，按照消費(fèi)者組來進(jìn)行輪訓(xùn)分配，同個(gè)消費(fèi)者組監(jiān)聽不同的主題也是一樣，是把所有的partition和所有的consumer都列出來，所以的話消費(fèi)者組里面的訂閱主題是一樣的才可以，主題不一樣的話會(huì)出現(xiàn)分配不均勻的問題。比如下面這個(gè)例子：

#有七個(gè)分區(qū)，同組里有兩個(gè)消費(fèi)者 
topic-p0/topic-p1/topic-p2/topic-p3/topic-p4/topic-p5/topic-p6  （分區(qū)） 
 
c-1: topic-p0/topic-p2/topic-p4/topic-p6 （消費(fèi)者1） 
 
c-2:topic-p1/topic-p3/topic-p5 （消費(fèi)者2） 

這樣會(huì)有什么弊端，如果是同一消費(fèi)者組里，所訂閱的消息是不相同的，在執(zhí)行分區(qū)的時(shí)候分配不是輪詢分配，這樣可能會(huì)導(dǎo)致分區(qū)分配的不均勻。例如現(xiàn)在有三個(gè)消費(fèi)者C0、C1、C2，它們共訂閱了3個(gè)主題：t0、t1、t2。這時(shí)候t0有1個(gè)分區(qū)(p0)，t1有2個(gè)分區(qū)(p0，p1)，t2有3個(gè)分區(qū)(p0,p1,p2)。消費(fèi)者C0訂閱了主題t0，消費(fèi)者C1訂閱主題t0和t1，消費(fèi)者C2訂閱的是t0,t1,t2。因?yàn)槭禽喸兊臋C(jī)制，當(dāng)C0訂閱到T0后，C1就訂閱不了到T0了，但是可以訂閱到T1，C2也一樣的訂閱不了T0，但是T1和T2都能訂閱到，這時(shí)候T2也就只有C2訂閱，其他的C0與C1是不可見的，這時(shí)候T2的的消息也就給C2這個(gè)消費(fèi)者來消費(fèi)了。這個(gè)情況就是分配不均的問題。

策略二、range(RangeAssignor默認(rèn)策略)范圍，按照主題來進(jìn)行分配，如果不平均分配的話，則第一個(gè)消費(fèi)者會(huì)分配比較多的分區(qū)，一個(gè)消費(fèi)者監(jiān)聽不同的主題也不影響，這一種策略有什么弊端呢，只是針對(duì)一個(gè)topic來說的話，c-1多消費(fèi)一個(gè)分區(qū)的話影響并不大，如果有多個(gè)topic，那么針對(duì)每一個(gè)topic的話，消費(fèi)者C-1都將多消費(fèi)1個(gè)分區(qū)，topic越多的話那么久消費(fèi)的分區(qū)也越多，性能會(huì)有所下降。

【面試題】Consumer消費(fèi)者重新分配策略與offset維護(hù)機(jī)制

什么是Rebalance操作

Kafka怎么均勻的分配某一個(gè)topic下所有的partition到各個(gè)消費(fèi)者的呢，從而使得消息的消費(fèi)速度達(dá)到了最快，這就是平衡。而rebalance(重平衡)其實(shí)就是重新進(jìn)行partition的分配，從而使得partition的分配重新達(dá)到了平衡的狀態(tài)。如下圖，有兩個(gè)Consumer，A和B，當(dāng)?shù)谌齻€(gè)成員C加入時(shí)，Kafka就會(huì)觸發(fā)Rebalance，重新分配策略為A、B、C重新分區(qū)，Rebalance之后的分配依舊還是公平的，每個(gè)Consumer實(shí)例都獲取了兩個(gè)分區(qū)的消費(fèi)權(quán)。

當(dāng)消費(fèi)者在消費(fèi)過程突然宕機(jī)了，重新恢復(fù)后是從哪里消費(fèi)，會(huì)有什么問題?

消費(fèi)者會(huì)記錄offset，故障恢復(fù)后會(huì)從這里繼續(xù)消費(fèi)，那么這個(gè)offset記錄在哪里呢?記錄在zookeeper和本地，新版的默認(rèn)將offset保證在kafka的內(nèi)置topic中，名稱為_consumer_offsets。在這個(gè)topic默認(rèn)會(huì)有50個(gè)Partition，每一個(gè)Partition都有3個(gè)副本，分區(qū)數(shù)量就是由參數(shù)offset.topic.num.partition配置的。通過groupid的哈希值和該參數(shù)的取模方式來確定某個(gè)消費(fèi)者組已消費(fèi)的offset保存到_consumer_offsets主題的哪個(gè)分區(qū)中。這個(gè)由消費(fèi)者組名+主題+分區(qū)，來確定唯一的offset的key，從而獲取對(duì)應(yīng)的值。