[[442535]]

本文轉載自微信公眾號「零零后程序員小三」，作者003 。轉載本文請聯系零零后程序員小三公眾號。

什么是Kafka

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

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

生產者與消費者機制

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

Kafka的producer生產者發送到Broker分區策略

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

那么如何指定消息到指定的分區里呢?

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

ProducerRecord就是發送給Broker的Key/value鍵值對，封裝基礎數據信息，簡稱為PR。

內部結構

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

生產者發送邏輯

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

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

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

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

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

生產者到Broker的發送流程

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

生產者常見的配置

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

Kafka的Consumer消費者機制和分區策略講解

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

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

消費者從哪個分區進行消費?

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

那么消費者從哪個分區來進行消費呢?

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

#有七個分區，同組里有兩個消費者 
topic-p0/topic-p1/topic-p2/topic-p3/topic-p4/topic-p5/topic-p6  （分區） 
 
c-1: topic-p0/topic-p2/topic-p4/topic-p6 （消費者1） 
 
c-2:topic-p1/topic-p3/topic-p5 （消費者2） 

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

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

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

什么是Rebalance操作

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

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

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