Kafka 的生產(chǎn)者與消費者機制+分區(qū)策略，你這還不懂？

作者：003 2021-12-28 12:01:59

開發(fā)

架構(gòu)

Kafka 目前 Kafka 已經(jīng)定位為一個分布式流式處理平臺，它以高吞吐、可持久化、可水平擴展、支持流數(shù)據(jù)處理等多種特性而被廣泛使用。

成都創(chuàng)新互聯(lián)始終堅持【策劃先行，效果至上】的經(jīng)營理念，通過多達10多年累計超上千家客戶的網(wǎng)站建設(shè)總結(jié)了一套系統(tǒng)有效的全網(wǎng)營銷解決方案，現(xiàn)已廣泛運用于各行各業(yè)的客戶，其中包括：成都柴油發(fā)電機等企業(yè)，備受客戶表揚。

 [[442535]]

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

什么是Kafka

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

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

生產(chǎn)者與消費者機制

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

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

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

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

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

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

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

  
 
 
 

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

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

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

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

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

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

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

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

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

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

  
 
 
 

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

Kafka的Consumer消費者機制和分區(qū)策略講解

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

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

消費者從哪個分區(qū)進行消費?

我們知道一個topic有多個partition，一個消費者組里面就有多個消費者，那是怎么分配的呢?一個主題topic可以有多個消費者，因為里面有多個partition分區(qū)(leader分區(qū))，一個partition leader可以由一個消費組里的一個消費者來消費。

那么消費者從哪個分區(qū)來進行消費呢?

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

  
 
 
 

   
  
  
  
#有七個分區(qū)，同組里有兩個消費者 
   
  
  
  
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 （消費者1） 
   
  
  
  
 
   
  
  
  
c-2:topic-p1/topic-p3/topic-p5 （消費者2） 
  
 
 
 

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

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

【面試題】Consumer消費者重新分配策略與offset維護機制

什么是Rebalance操作

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

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

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

 

網(wǎng)站欄目：Kafka的生產(chǎn)者與消費者機制+分區(qū)策略，你這還不懂？
標(biāo)題網(wǎng)址：http://www.5511xx.com/article/cdcjcoi.html