詳細解析kafka之kafka分區和副本

本篇主要介紹kafka的分區和副本，由於這二者是有些關聯的，因此就放在一塊兒來說了，後面順便會給出一些對應的配置以及具體的實現代碼，以供參考~

1.kafka分區機制

分區機制是kafka實現高吞吐的祕密武器，但這個武器用得很差的話也容易出問題，今天主要就來介紹分區的機制以及相關的部分配置。

首先，從數據組織形式來講，kafka有三層形式，kafka有多個主題，每一個主題有多個分區，每一個分區又有多條消息。

而每一個分區能夠分佈到不一樣的機器上，這樣一來，從服務端來講，分區能夠實現高伸縮性，以及負載均衡，動態調節的能力。

固然多分區就意味着每條消息都難以按照順序存儲，那麼是否是意味着這樣的業務場景kafka就無能爲力呢？不是的，最簡單的作法可使用單個分區，單個分區，全部消息天然都順序寫入到一個分區中，就跟順序隊列同樣了。而複雜些的，還有其餘辦法，那就是使用按消息鍵，將須要順序保存的消息存儲的單獨的分區，其餘消息存儲其餘分區，這個在下面會介紹。

咱們能夠經過replication-factor指定建立topic時候所建立的分區數。

bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test

好比這裏就是建立了1個分區，的主題。值得注意的是，還有一種建立主題的方法，是使用zookeeper參數的，那種是比較舊的建立方法，這裏是使用bootstrap參數的。

1.1 分區個數選擇

既然分區效果這麼好，是否是越多分區越好呢？顯而易見並不是如此。

分區越多，所須要消耗的資源就越多。甚至若是足夠大的時候，還會觸發到操做系統的一些參數限制。好比linux中的文件描述符限制，通常在建立線程，建立socket，打開文件的場景下，linux默認的文件描述符參數，只有1024，超過則會報錯。

看到這裏有讀者就會不耐煩了，說這麼多有啥用，能不能直接告訴我分區分多少個比較好？很遺憾，暫時沒有。

由於每一個業務場景都不一樣，只能結合具體業務來看。假如每秒鐘須要從主題寫入和讀取1GB數據，而消費者1秒鐘最多處理50MB的數據，那麼這個時候就能夠設置20-25個分區，固然還要結合具體的物理資源狀況。

而如何沒法估算出大概的處理速度和時間，那麼就用基準測試來測試吧。建立不一樣分區的topic，逐步壓測測出最終的結果。若是實在是懶得測，那比較無腦的肯定分區數的方式就是broker機器數量的2~3倍。

1.2 分區寫入策略

所謂分區寫入策略，便是生產者將數據寫入到kafka主題後，kafka如何將數據分配到不一樣分區中的策略。

常見的有三種策略，輪詢策略，隨機策略，和按鍵保存策略。其中輪詢策略是默認的分區策略，而隨機策略則是較老版本的分區策略，不過因爲其分配的均衡性不如輪詢策略，故然後來改爲了輪詢策略爲默認策略。

輪詢策略

所謂輪詢策略，即按順序輪流將每條數據分配到每一個分區中。

舉個例子，假設主題test有三個分區，分別是分區A，分區B和分區C。那麼主題對接收到的第一條消息寫入A分區，第二條消息寫入B分區，第三條消息寫入C分區，第四條消息則又寫入A分區，依此類推。

輪詢策略是默認的策略，故而也是使用最頻繁的策略，它能最大限度保證全部消息都平均分配到每個分區。除非有特殊的業務需求，不然使用這種方式便可。

隨機策略

隨機策略，也就是每次都隨機地將消息分配到每一個分區。其實大概就是先得出分區的數量，而後每次獲取一個隨機數，用該隨機數肯定消息發送到哪一個分區。

在比較早的版本，默認的分區策略就是隨機策略，但其實使用隨機策略也是爲了更好得將消息均衡寫入每一個分區。但後來發現對這一需求而言，輪詢策略的表現更優，因此社區後來的默認策略就是輪詢策略了。

按鍵保存策略

按鍵保存策略，就是當生產者發送數據的時候，能夠指定一個key，計算這個key的hashCode值，按照hashCode的值對不一樣消息進行存儲。

至於要如何實現，那也簡單，只要讓生產者發送的時候指定key就行。欸剛剛不是說默認的是輪詢策略嗎？其實啊，kafka默認是實現了兩個策略，沒指定key的時候就是輪詢策略，有的話那激素按鍵保存策略了。

上面有說到一個場景，那就是要順序發送消息到kafka。前面提到的方案是讓全部數據存儲到一個分區中，但其實更好的作法，就是使用這種按鍵保存策略。

讓須要順序存儲的數據都指定相同的鍵，而不須要順序存儲的數據指定不一樣的鍵，這樣一來，即實現了順序存儲的需求，又可以享受到kafka多分區的優點，豈不美哉。

1.3 實現自定義分區

說了這麼多，那麼到底要如何自定義分區呢？

kafka提供了兩種讓咱們本身選擇分區的方法，第一種是在發送producer的時候，在ProducerRecord中直接指定，但須要知道具體發送的分區index，因此並不推薦。

第二種則是須要實現Partitioner.class類，並重寫類中的partition(String topic, Object key, byte[] keyBytes,Object value, byte[] valueBytes, Cluster cluster) 方法。後面在生成kafka producer客戶端的時候直接指定新的分區類就能夠了。

package kafkaconf;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;


public class MyParatitioner implements Partitioner {
    @Override
    public void configure(Map<String, ?> configs) {
    }

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        //key不能空，若是key爲空的會經過輪詢的方式 選擇分區
        if(keyBytes == null || (!(key instanceof String))){
            throw new RuntimeException("key is null");
        }
        //獲取分區列表
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);

        //如下是上述各類策略的實現，不能共存
        //隨機策略
        return ThreadLocalRandom.current().nextInt(partitions.size());

        //按消息鍵保存策略
        return Math.abs(key.hashCode()) % partitions.size();

        //自定義分區策略, 好比key爲123的消息，選擇放入最後一個分區
        if(key.toString().equals("123")){
            return partitions.size()-1;
        }else{
            //不然隨機
            ThreadLocalRandom.current().nextInt(partitions.size());
        }
    }

    @Override
    public void close() {
    }
}

而後須要在生成kafka producer客戶端的時候指定該類就行：

val properties = new Properties()
    ......
    props.put("partitioner.class", "kafkaconf.MyParatitioner");  //主要這個配置指定分區類
    ......其餘配置
    val producer = new KafkaProducer[String, String](properties)

2.kafka副本機制

說完了分區，再來講說副本。先說說副本的基本內容，在kafka中，每一個主題能夠有多個分區，每一個分區又能夠有多個副本。這多個副本中，只有一個是leader，而其餘的都是follower副本。僅有leader副本能夠對外提供服務。

多個follower副本一般存放在和leader副本不一樣的broker中。經過這樣的機制實現了高可用，當某臺機器掛掉後，其餘follower副本也能迅速」轉正「，開始對外提供服務。

這裏經過問題來整理這部份內容。

kafka的副本都有哪些做用？

在kafka中，實現副本的目的就是冗餘備份，且僅僅是冗餘備份，全部的讀寫請求都是由leader副本進行處理的。follower副本僅有一個功能，那就是從leader副本拉取消息，儘可能讓本身跟leader副本的內容一致。

說說follower副本爲何不對外提供服務？

這個問題本質上是對性能和一致性的取捨。試想一下，若是follower副本也對外提供服務那會怎麼樣呢？首先，性能是確定會有所提高的。但同時，會出現一系列問題。相似數據庫事務中的幻讀，髒讀。

好比你如今寫入一條數據到kafka主題a，消費者b從主題a消費數據，卻發現消費不到，由於消費者b去讀取的那個分區副本中，最新消息還沒寫入。而這個時候，另外一個消費者c卻能夠消費到最新那條數據，由於它消費了leader副本。

看吧，爲了提升那麼些性能而致使出現數據不一致問題，那顯然是不值得的。

leader副本掛掉後，如何選舉新副本？

若是你對zookeeper選舉機制有所瞭解，就知道zookeeper每次leader節點掛掉時，都會經過內置id，來選舉處理了最新事務的那個follower節點。

從結果上來講，kafka分區副本的選舉也是相似的，都是選擇最新的那個follower副本，但它是經過一個In-sync（ISR）副本集合實現。

kafka會將與leader副本保持同步的副本放到ISR副本集合中。固然，leader副本是一直存在於ISR副本集合中的，在某些特殊狀況下，ISR副本中甚至只有leader一個副本。

當leader掛掉時，kakfa經過zookeeper感知到這一狀況，在ISR副本中選取新的副本成爲leader，對外提供服務。

但這樣還有一個問題，前面提到過，有可能ISR副本集合中，只有leader，當leader副本掛掉後，ISR集合就爲空，這時候怎麼辦呢？這時候若是設置unclean.leader.election.enable參數爲true，那麼kafka會在非同步，也就是不在ISR副本集合中的副本中，選取出副本成爲leader，但這樣意味這消息會丟失，這又是可用性和一致性的一個取捨了。

ISR副本集合保存的副本的條件是什麼？

上面一直說ISR副本集合中的副本就是和leader副本是同步的，那這個同步的標準又是什麼呢？

答案其實跟一個參數有關：replica.lag.time.max.ms。

前面說到follower副本的任務，就是從leader副本拉取消息，若是持續拉取速度慢於leader副本寫入速度，慢於時間超過replica.lag.time.max.ms後，它就變成「非同步」副本，就會被踢出ISR副本集合中。但後面如何follower副本的速度慢慢提上來，那就又可能會從新加入ISR副本集合中了。

producer的acks參數

前面說了那麼多理論的知識，那麼就能夠來看看如何在實際應用中使用這些知識。

跟副本關係最大的，那天然就是acks機制，acks決定了生產者如何在性能與數據可靠之間作取捨。

配置acks的代碼其實很簡單，只須要在新建producer的時候多加一個配置：

val properties = new Properties()
    ......
    props.put("acks", "0/1/-1");  //配置acks，有三個可選值
    ......其餘配置
    val producer = new KafkaProducer[String, String](properties)

acks這個配置能夠指定三個值，分別是0，1和-1。咱們分別來講三者表明什麼：

• acks爲0：這意味着producer發送數據後，不會等待broker確認，直接發送下一條數據，性能最快
• acks爲1：爲1意味着producer發送數據後，須要等待leader副本確認接收後，纔會發送下一條數據，性能中等
• acks爲-1：這個表明的是all，意味着發送的消息寫入全部的ISR集合中的副本（注意不是所有副本）後，纔會發送下一條數據，性能最慢，但可靠性最強

還有一點值得一提，kafka有一個配置參數，min.insync.replicas，默認是1（也就是隻有leader，實際生產應該調高），該屬性規定了最小的ISR數。這意味着當acks爲-1（即all）的時候，這個參數規定了必須寫入的ISR集中的副本數，若是沒達到，那麼producer會產生異常。

以上~