本篇主要介紹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 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 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會產生異常。