本篇主要介紹kafka的分區(qū)和副本蝙眶，因?yàn)檫@兩者是有些關(guān)聯(lián)的季希，所以就放在一起來講了，后面順便會(huì)給出一些對(duì)應(yīng)的配置以及具體的實(shí)現(xiàn)代碼幽纷，以供參考~
1.kafka分區(qū)機(jī)制
分區(qū)機(jī)制是kafka實(shí)現(xiàn)高吞吐的秘密武器式塌，但這個(gè)武器用得不好的話也容易出問題，今天主要就來介紹分區(qū)的機(jī)制以及相關(guān)的部分配置友浸。
首先峰尝，從數(shù)據(jù)組織形式來說，kafka有三層形式尾菇，kafka有多個(gè)主題境析，每個(gè)主題有多個(gè)分區(qū)，每個(gè)分區(qū)又有多條消息派诬。
而每個(gè)分區(qū)可以分布到不同的機(jī)器上劳淆，這樣一來，從服務(wù)端來說默赂，分區(qū)可以實(shí)現(xiàn)高伸縮性沛鸵，以及負(fù)載均衡，動(dòng)態(tài)調(diào)節(jié)的能力缆八。
當(dāng)然多分區(qū)就意味著每條消息都難以按照順序存儲(chǔ)曲掰，那么是不是意味著這樣的業(yè)務(wù)場(chǎng)景kafka就無能為力呢？不是的奈辰，最簡(jiǎn)單的做法可以使用單個(gè)分區(qū)栏妖，單個(gè)分區(qū)，所有消息自然都順序?qū)懭氲揭粋€(gè)分區(qū)中奖恰，就跟順序隊(duì)列一樣了吊趾。而復(fù)雜些的，還有其他辦法瑟啃，那就是使用按消息鍵论泛，將需要順序保存的消息存儲(chǔ)的單獨(dú)的分區(qū)，其他消息存儲(chǔ)其他分區(qū)蛹屿，這個(gè)在下面會(huì)介紹屁奏。
我們可以通過replication-factor指定創(chuàng)建topic時(shí)候所創(chuàng)建的分區(qū)數(shù)。
bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test
比如這里就是創(chuàng)建了1個(gè)分區(qū)错负，的主題坟瓢。值得注意的是，還有一種創(chuàng)建主題的方法湿颅，是使用zookeeper參數(shù)的载绿，那種是比較舊的創(chuàng)建方法，這里是使用bootstrap參數(shù)的油航。
1.1 分區(qū)個(gè)數(shù)選擇
既然分區(qū)效果這么好崭庸，是不是越多分區(qū)越好呢？顯而易見并非如此谊囚。
分區(qū)越多怕享，所需要消耗的資源就越多。甚至如果足夠大的時(shí)候镰踏，還會(huì)觸發(fā)到操作系統(tǒng)的一些參數(shù)限制函筋。比如linux中的文件描述符限制，一般在創(chuàng)建線程奠伪，創(chuàng)建socket跌帐，打開文件的場(chǎng)景下首懈，linux默認(rèn)的文件描述符參數(shù)，只有1024谨敛，超過則會(huì)報(bào)錯(cuò)究履。
看到這里有讀者就會(huì)不耐煩了，說這么多有啥用脸狸，能不能直接告訴我分區(qū)分多少個(gè)比較好最仑？很遺憾，暫時(shí)沒有炊甲。
因?yàn)槊總€(gè)業(yè)務(wù)場(chǎng)景都不同泥彤，只能結(jié)合具體業(yè)務(wù)來看。假如每秒鐘需要從主題寫入和讀取1GB數(shù)據(jù)卿啡，而消費(fèi)者1秒鐘最多處理50MB的數(shù)據(jù)吟吝，那么這個(gè)時(shí)候就可以設(shè)置20-25個(gè)分區(qū)，當(dāng)然還要結(jié)合具體的物理資源情況牵囤。
而如何無法估算出大概的處理速度和時(shí)間爸黄，那么就用基準(zhǔn)測(cè)試來測(cè)試吧。創(chuàng)建不同分區(qū)的topic揭鳞，逐步壓測(cè)測(cè)出最終的結(jié)果炕贵。如果實(shí)在是懶得測(cè)，那比較無腦的確定分區(qū)數(shù)的方式就是broker機(jī)器數(shù)量的2~3倍野崇。
1.2 分區(qū)寫入策略
所謂分區(qū)寫入策略称开，即是生產(chǎn)者將數(shù)據(jù)寫入到kafka主題后，kafka如何將數(shù)據(jù)分配到不同分區(qū)中的策略乓梨。
常見的有三種策略鳖轰，輪詢策略，隨機(jī)策略扶镀，和按鍵保存策略蕴侣。其中輪詢策略是默認(rèn)的分區(qū)策略，而隨機(jī)策略則是較老版本的分區(qū)策略臭觉，不過由于其分配的均衡性不如輪詢策略昆雀，故而后來改成了輪詢策略為默認(rèn)策略。
輪詢策略
所謂輪詢策略蝠筑，即按順序輪流將每條數(shù)據(jù)分配到每個(gè)分區(qū)中狞膘。
舉個(gè)例子，假設(shè)主題test有三個(gè)分區(qū)什乙，分別是分區(qū)A挽封，分區(qū)B和分區(qū)C。那么主題對(duì)接收到的第一條消息寫入A分區(qū)臣镣，第二條消息寫入B分區(qū)辅愿，第三條消息寫入C分區(qū)智亮，第四條消息則又寫入A分區(qū)，依此類推点待。
輪詢策略是默認(rèn)的策略鸽素，故而也是使用最頻繁的策略，它能最大限度保證所有消息都平均分配到每一個(gè)分區(qū)亦鳞。除非有特殊的業(yè)務(wù)需求，否則使用這種方式即可棒坏。
隨機(jī)策略
隨機(jī)策略燕差，也就是每次都隨機(jī)地將消息分配到每個(gè)分區(qū)。其實(shí)大概就是先得出分區(qū)的數(shù)量坝冕，然后每次獲取一個(gè)隨機(jī)數(shù)徒探，用該隨機(jī)數(shù)確定消息發(fā)送到哪個(gè)分區(qū)。
在比較早的版本喂窟，默認(rèn)的分區(qū)策略就是隨機(jī)策略测暗，但其實(shí)使用隨機(jī)策略也是為了更好得將消息均衡寫入每個(gè)分區(qū)。但后來發(fā)現(xiàn)對(duì)這一需求而言磨澡，輪詢策略的表現(xiàn)更優(yōu)碗啄，所以社區(qū)后來的默認(rèn)策略就是輪詢策略了。
按鍵保存策略
按鍵保存策略稳摄，就是當(dāng)生產(chǎn)者發(fā)送數(shù)據(jù)的時(shí)候稚字，可以指定一個(gè)key，計(jì)算這個(gè)key的hashCode值厦酬，按照hashCode的值對(duì)不同消息進(jìn)行存儲(chǔ)胆描。
至于要如何實(shí)現(xiàn)，那也簡(jiǎn)單仗阅，只要讓生產(chǎn)者發(fā)送的時(shí)候指定key就行昌讲。欸剛剛不是說默認(rèn)的是輪詢策略嗎？其實(shí)啊减噪，kafka默認(rèn)是實(shí)現(xiàn)了兩個(gè)策略短绸，沒指定key的時(shí)候就是輪詢策略，有的話那激素按鍵保存策略了旋廷。
上面有說到一個(gè)場(chǎng)景鸠按，那就是要順序發(fā)送消息到kafka。前面提到的方案是讓所有數(shù)據(jù)存儲(chǔ)到一個(gè)分區(qū)中饶碘，但其實(shí)更好的做法目尖，就是使用這種按鍵保存策略。
讓需要順序存儲(chǔ)的數(shù)據(jù)都指定相同的鍵扎运，而不需要順序存儲(chǔ)的數(shù)據(jù)指定不同的鍵瑟曲，這樣一來饮戳，即實(shí)現(xiàn)了順序存儲(chǔ)的需求，又能夠享受到kafka多分區(qū)的優(yōu)勢(shì)洞拨，豈不美哉扯罐。
1.3 實(shí)現(xiàn)自定義分區(qū)
說了這么多，那么到底要如何自定義分區(qū)呢烦衣？
kafka提供了兩種讓我們自己選擇分區(qū)的方法歹河，第一種是在發(fā)送producer的時(shí)候，在ProducerRecord中直接指定花吟，但需要知道具體發(fā)送的分區(qū)index秸歧，所以并不推薦。
第二種則是需要實(shí)現(xiàn)Partitioner.class類衅澈，并重寫類中的partition(String topic, Object key, byte[] keyBytes,Object value, byte[] valueBytes, Cluster cluster) 方法键菱。后面在生成kafka producer客戶端的時(shí)候直接指定新的分區(qū)類就可以了。
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為空的會(huì)通過輪詢的方式 選擇分區(qū)
if(keyBytes == null || (!(key instanceof String))){
throw new RuntimeException("key is null");
}
//獲取分區(qū)列表
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
//以下是上述各種策略的實(shí)現(xiàn)经备，不能共存
//隨機(jī)策略
return ThreadLocalRandom.current().nextInt(partitions.size());
//按消息鍵保存策略
return Math.abs(key.hashCode()) % partitions.size();
//自定義分區(qū)策略, 比如key為123的消息，選擇放入最后一個(gè)分區(qū)
if(key.toString().equals("123")){
return partitions.size()-1;
}else{
//否則隨機(jī)
ThreadLocalRandom.current().nextInt(partitions.size());
}
}
@Override
public void close() {
}
}
然后需要在生成kafka producer客戶端的時(shí)候指定該類就行：
val properties = new Properties()
......
props.put("partitioner.class", "kafkaconf.MyParatitioner"); //主要這個(gè)配置指定分區(qū)類
......其他配置
val producer = new KafkaProducer[String, String](properties)
2.kafka副本機(jī)制
說完了分區(qū)部默，再來說說副本侵蒙。先說說副本的基本內(nèi)容，在kafka中傅蹂，每個(gè)主題可以有多個(gè)分區(qū)蘑志，每個(gè)分區(qū)又可以有多個(gè)副本。這多個(gè)副本中贬派，只有一個(gè)是leader急但，而其他的都是follower副本。僅有l(wèi)eader副本可以對(duì)外提供服務(wù)搞乏。
多個(gè)follower副本通常存放在和leader副本不同的broker中波桩。通過這樣的機(jī)制實(shí)現(xiàn)了高可用，當(dāng)某臺(tái)機(jī)器掛掉后请敦，其他follower副本也能迅速”轉(zhuǎn)正“镐躲，開始對(duì)外提供服務(wù)。
這里通過問題來整理這部分內(nèi)容侍筛。
kafka的副本都有哪些作用萤皂？
在kafka中，實(shí)現(xiàn)副本的目的就是冗余備份匣椰，且僅僅是冗余備份裆熙，所有的讀寫請(qǐng)求都是由leader副本進(jìn)行處理的。follower副本僅有一個(gè)功能，那就是從leader副本拉取消息入录，盡量讓自己跟leader副本的內(nèi)容一致蛤奥。
說說follower副本為什么不對(duì)外提供服務(wù)？
這個(gè)問題本質(zhì)上是對(duì)性能和一致性的取舍僚稿。試想一下凡桥，如果follower副本也對(duì)外提供服務(wù)那會(huì)怎么樣呢？首先蚀同，性能是肯定會(huì)有所提升的缅刽。但同時(shí)，會(huì)出現(xiàn)一系列問題蠢络。類似數(shù)據(jù)庫事務(wù)中的幻讀拷恨，臟讀。
比如你現(xiàn)在寫入一條數(shù)據(jù)到kafka主題a谢肾，消費(fèi)者b從主題a消費(fèi)數(shù)據(jù)，卻發(fā)現(xiàn)消費(fèi)不到小泉，因?yàn)橄M(fèi)者b去讀取的那個(gè)分區(qū)副本中芦疏，最新消息還沒寫入。而這個(gè)時(shí)候微姊，另一個(gè)消費(fèi)者c卻可以消費(fèi)到最新那條數(shù)據(jù)酸茴，因?yàn)樗M(fèi)了leader副本。
看吧兢交，為了提高那么些性能而導(dǎo)致出現(xiàn)數(shù)據(jù)不一致問題薪捍，那顯然是不值得的。
leader副本掛掉后配喳，如何選舉新副本酪穿？
如果你對(duì)zookeeper選舉機(jī)制有所了解，就知道zookeeper每次leader節(jié)點(diǎn)掛掉時(shí)晴裹，都會(huì)通過內(nèi)置id被济，來選舉處理了最新事務(wù)的那個(gè)follower節(jié)點(diǎn)。
從結(jié)果上來說涧团，kafka分區(qū)副本的選舉也是類似的只磷，都是選擇最新的那個(gè)follower副本，但它是通過一個(gè)In-sync（ISR）副本集合實(shí)現(xiàn)泌绣。
kafka會(huì)將與leader副本保持同步的副本放到ISR副本集合中钮追。當(dāng)然，leader副本是一直存在于ISR副本集合中的阿迈，在某些特殊情況下元媚，ISR副本中甚至只有l(wèi)eader一個(gè)副本。
當(dāng)leader掛掉時(shí)，kakfa通過zookeeper感知到這一情況惠毁，在ISR副本中選取新的副本成為leader犹芹，對(duì)外提供服務(wù)。
但這樣還有一個(gè)問題鞠绰，前面提到過腰埂，有可能ISR副本集合中，只有l(wèi)eader蜈膨，當(dāng)leader副本掛掉后屿笼，ISR集合就為空，這時(shí)候怎么辦呢翁巍？這時(shí)候如果設(shè)置unclean.leader.election.enable參數(shù)為true驴一，那么kafka會(huì)在非同步，也就是不在ISR副本集合中的副本中灶壶，選取出副本成為leader肝断，但這樣意味這消息會(huì)丟失，這又是可用性和一致性的一個(gè)取舍了驰凛。
ISR副本集合保存的副本的條件是什么胸懈？
上面一直說ISR副本集合中的副本就是和leader副本是同步的，那這個(gè)同步的標(biāo)準(zhǔn)又是什么呢恰响？
答案其實(shí)跟一個(gè)參數(shù)有關(guān)：replica.lag.time.max.ms趣钱。
前面說到follower副本的任務(wù)，就是從leader副本拉取消息胚宦，如果持續(xù)拉取速度慢于leader副本寫入速度首有，慢于時(shí)間超過replica.lag.time.max.ms后，它就變成“非同步”副本枢劝，就會(huì)被踢出ISR副本集合中井联。但后面如何follower副本的速度慢慢提上來，那就又可能會(huì)重新加入ISR副本集合中了您旁。
producer的acks參數(shù)
前面說了那么多理論的知識(shí)低矮，那么就可以來看看如何在實(shí)際應(yīng)用中使用這些知識(shí)。
跟副本關(guān)系最大的被冒，那自然就是acks機(jī)制军掂，acks決定了生產(chǎn)者如何在性能與數(shù)據(jù)可靠之間做取舍。
配置acks的代碼其實(shí)很簡(jiǎn)單昨悼，只需要在新建producer的時(shí)候多加一個(gè)配置：
val properties = new Properties()
......
props.put("acks", "0/1/-1"); //配置acks蝗锥，有三個(gè)可選值
......其他配置
val producer = new KafkaProducer[String, String](properties)
acks這個(gè)配置可以指定三個(gè)值，分別是0率触，1和-1终议。我們分別來說三者代表什么：
- acks為0：這意味著producer發(fā)送數(shù)據(jù)后，不會(huì)等待broker確認(rèn)，直接發(fā)送下一條數(shù)據(jù)穴张，性能最快
- acks為1：為1意味著producer發(fā)送數(shù)據(jù)后细燎，需要等待leader副本確認(rèn)接收后，才會(huì)發(fā)送下一條數(shù)據(jù)皂甘，性能中等
- acks為-1：這個(gè)代表的是all玻驻，意味著發(fā)送的消息寫入所有的ISR集合中的副本（注意不是全部副本）后，才會(huì)發(fā)送下一條數(shù)據(jù)偿枕，性能最慢璧瞬，但可靠性最強(qiáng)
還有一點(diǎn)值得一提，kafka有一個(gè)配置參數(shù)渐夸，min.insync.replicas嗤锉，默認(rèn)是1（也就是只有l(wèi)eader，實(shí)際生產(chǎn)應(yīng)該調(diào)高），該屬性規(guī)定了最小的ISR數(shù)。這意味著當(dāng)acks為-1（即all）的時(shí)候坐漏，這個(gè)參數(shù)規(guī)定了必須寫入的ISR集中的副本數(shù)，如果沒達(dá)到访诱，那么producer會(huì)產(chǎn)生異常。
以上~
