最近看了kafka2.4新版本的一些功能特性，不得不說，在kafka2.0以后，kafka自身就比較少推出一些新的feature了，基本都是一些修修補補的東西，倒是kafka connect和kafka stream相關的開發作業做的比較多，可能kafka的野心也不局限于要當一個中間件，而是要實作一個流處理系統的生態了，

這次要介紹的是我覺得比較有意思的兩個特性，一個是kafka支持從follower副本讀取資料，當然這個功能并不是為了提供讀取性能，后面再詳細介紹，另一個則是新推出的sticky partitioner功能，我猜是從rebalance的StickyAssignor中得到靈感，發現producer的磁區策略也可以這樣搞，233，這個feature主要作用是提高性能，

這兩個feature都是kafka2.4.0版本推出的，如果想使用這些新feature，那么不妨升級下吧～

follower副本讀取資料（consumer fetch from closest replica）

背景

在早先kafka的設計中，為了使consumer讀取資料能夠保持一致，是只允許consumer讀取leader副本的資料的，即follower replica只是單純地備份資料的作用，那推出follower replica fetch功能的背景是什么呢？

舉個比較常見的場景，kafka存在多個資料中心，不同資料中心存在于不同的機房，當其中一個資料中心需要向另一個資料中心同步資料的時候，由于只能從leader replica消費資料，那么它不得不進行跨機房獲取資料，而這些流量帶寬通常是比較昂貴的（尤其是云服務器），即無法利用本地性來減少昂貴的跨機房流量，

所以kafka推出這一個功能，就是幫助類似這種場景，節約流量資源，并且這種功能似憾訓可以和新推出的mirror maker2相互配合，實作多個資料源的資料同步，不過我自己還沒測驗過，

rack功能介紹

要說follower replica fetch，那就不得不先說rack功能，這個是kafka比較早就推出的功能，是Kafka對機架感知提供了的基本支持，可以將其用于控制副本的放置，詳細內容可以參閱這篇Kafka機架感知文章，

使用方式，其實就是一個broker端的引數，broker.rack，這個引數可以說明當前broker在哪個機房，

舉上面文章中的例子，如果一個資料中心的集群分布如下：
kafka-rack功能

那么可以這樣配置：

broker0 -> rack1
broker1 -> rack1
broker2 -> rack2
broker3 -> rack2

這樣其實就是相當于給broker打一個標簽，當新建topic，比如新建一個兩個副本 & 兩個磁區的topic，kafka至少會自動給rack1或rack2分配全部磁區的一個副本，什么，你說要是創建兩個磁區一個副本的topic該怎么分，，，抱歉，我給不了答案，等你自己實踐然后評論跟我說下答案 =，=

replica fetch功能測驗

OK，上面介紹的rack功能，我們就能發現，這個其實跟跨機房讀資料的這種場景是很搭的，在跨機房多資料中心場景中，如果資料中心A，一個副本放在資料中心B的機房中，只要讓資料中心B的consumer能夠讀資料中心A的那個replica的資料（follower副本）讀資料，那不就萬事大吉，

社區也是這樣想的，所以就推出了這個功能，讓消費者可以指定rack id，然后可以不從消費者讀取資料，要實作這個目的，需要先配置兩個引數：

replica.selector.class

broker端配置
配置名：replica.selector.class
解釋：ReplicaSelector實作類的全名，包括路徑 (比如 RackAwareReplicaSelector 即按 rack id 指定消費)
默認：從 Leader 消費的 LeaderSelector

為了支持這個功能，kafka修改了這部分的介面實作，原始碼中新增一個ReplicaSelector介面，如果用戶有自定義消費策略的需求，也可以繼承這個介面實作自己的功能，

目前這個介面有兩個實作類，一個是LeaderSelector，即從leader副本讀資料，另一個則是RackAwareReplicaSelector，會去到指定的rack id讀資料，

client.rack

consumer端配置
配置名：client.rack
解釋：這個引數需要和broker端指定的broker.rack相同，表示去哪個rack中獲取資料，
默認：null

這個引數只有在上面的replica.selector.class指定為RackAwareReplicaSelector且broekr指定了broker.rack才會生效，

這個功能要測驗也挺簡單的，可以直接搭建一個兩個broker的kafka集群，配置broker.rack，然后使用consumer客戶端指定client.rack發送到非leader的節點查資料就行了，另外，可以使用這條命令查看網卡流量資訊：

sar -n DEV 1 300

存在問題

從follower replica讀取資料肯定有問題，最可能的問題就是落后節點的問題，從這樣的節點讀取資料會面臨什么樣的情況呢？官方給出了幾種場景及解決辦法，先看看這張圖

主要有四種可能出現問題的情況，我們分別來看看應該如何解決：

Case 1（uncommitted offset）

這個場景是follower接收到資料但還未committed offset，這個時候，若消費者的offet消費到high watemark到log end offset之間的那段（Case 1黃色那段），會回傳空資料，而不是一個錯誤資訊，直到這段內容 committed，

case 2（unavailable offset）

這種場景應該發生于慢節點的情況下，滿節點的broker還未接收到實際資料，但已經跟leader通信知道有部分資料committed了（case 2黃色部分），當遇到這種情況，consumer 消費到時候，會回傳 OFFSET_NOT_AVAILABLE 錯誤資訊，

case 3（offset too small）

這種情況可能出現在消費者指定了 offset 的情況，那么在指定不同auto.offset.reset的時候有不同的情況，

If the reset policy is "earliest," fetch the log start offset of the current replica that raised the out of range error.
If the reset policy is "latest," fetch the log end offset from the leader.
If the reset policy is "none," raise an exception.

case 4（offset too large）

遇到這種情況，會回傳一個 broker 會回傳一個 OFFSET_OUT_OF_RANGE 的錯誤，

但 OFFSET_OUT_OF_RANGE 遇到這種錯誤的時候也有多種可能，官方給出當 consumer 遇到這種問題的解決思路，

Use the OffsetForLeaderEpoch API to verify the current position with the leader.

If the fetch offset is still valid, refresh metadata and continue fetching
If truncation was detected, follow the steps in KIP-320 to either reset the offset or raise the truncation error
Otherwise, follow the same steps above as in case 3.

sticky partitioner功能

背景

kafka producer發送資料并不是一個一個訊息發送，而是取決于兩個producer端引數，一個是linger.ms，默認是0ms，當達到這個時間后，kafka producer就會立刻向broker發送資料，另一個引數是batch.size，默認是16kb，當產生的訊息數達到這個大小后，就會立即向broker發送資料，

按照這個設計，從直觀上思考，肯定是希望每次都盡可能填滿一個batch再發送到一個磁區，但實際決定batch如何形成的一個因素是磁區策略（partitioner strategy），在Kafka2.4版本之前，在producer發送資料默認的磁區策略是輪詢策略（沒指定keyd的情況），這在我以前的文章有說到過詳細決議kafka之kafka磁區和副本，如果多條訊息不是被發送到相同的磁區，它們就不能被放入到一個batch中，

所以如果使用默認的輪詢partition策略，可能會造成一個大的batch被輪詢成多個小的batch的情況，鑒于此，kafka2.4的時候推出一種新的磁區策略，即Sticky Partitioning Strategy，Sticky Partitioning Strategy會隨機地選擇另一個磁區并會盡可能地堅持使用該磁區——即所謂的粘住這個磁區，

鑒于小batch可能導致延時增加，之前對于無Key訊息的磁區策略效率很低，社區于2.4版本引入了黏性磁區策略（Sticky Partitioning Strategy），該策略是一種全新的策略，能夠顯著地降低給訊息指定磁區程序中的延時，

使用Sticky Partitioner有助于改進訊息批處理，減少延遲，并減少broker的負載，

功能決議

sticky Partitioner實作的代碼是在UniformStickyPartitioner里面，貼下代碼看看：

public class UniformStickyPartitioner implements Partitioner {

    private final StickyPartitionCache stickyPartitionCache = new StickyPartitionCache();

    public void configure(Map<String, ?> configs) {}

    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        return stickyPartitionCache.partition(topic, cluster);
    }

    public void close() {}

    public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
        stickyPartitionCache.nextPartition(topic, cluster, prevPartition);
    }
}

我們主要關注UniformStickyPartitioner#partition()方法，可以看到，它是直接通過一個cache類獲取相同的磁區，這表示新的record會一直發送到同一個磁區中，除非生成新的batch，觸發了UniformStickyPartitioner#onNewBatch()方法才會換磁區，

可以看看RoundRobinPartitioner#partition()方法（即輪詢磁區策略）進行對比，就能發現比較明顯的對比，

這個sticky partitioner最大的好處就是性能較好，按照官方給出的測驗結果，使用sticky partitioner測驗可以減少50%的延時，吞吐也有相對應的提高，我自己測了下資料基本出入不大，

另外說明下，在kafka2.4以后，默認的partitioner磁區策略，已經包含了sticky partitioner了，所以升級到kafka2.4以后，并不需要任何修改就能享受到性能到極大提升，這里可以看下kafka2.4版本的策略說明：

/**
 * The default partitioning strategy:
 * <ul>
 * <li>If a partition is specified in the record, use it
 * <li>If no partition is specified but a key is present choose a partition based on a hash of the key
 * <li>If no partition or key is present choose the sticky partition that changes when the batch is full.
 * 
 * See KIP-480 for details about sticky partitioning.
 */
public class DefaultPartitioner implements Partitioner {

有一點挺奇怪到，在測驗程序中（使用bin/kafka-producer-perf-test.sh測驗），發現DefaultPartitioner的性能要比UniformStickyPartitioner的性能要好一些，不確定是什么原因，知道到小伙伴可以在評論區給出答案:)

參考：
KIP-392: Allow consumers to fetch from closest replica
KIP-480: Sticky Partitioner
以上～

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/228384.html

標籤：Java

上一篇：思維導圖學 Kotlin

下一篇：從入門到應用再到原始碼，京東技術官撰寫的Redis進階寶典，不愧為跳槽漲薪，首選秘籍！

什么,kafka能夠從follower副本讀資料了 —kafka新功能介紹