摘自Jason’s Blog滋觉，原文鏈接 http://www.jasongj.com/2015/01/02/Kafka深度解析
一 kafka的簡單理解
Kafka是Apache下的一個子項目签夭，是一個高性能跨語言分布式發(fā)布/訂閱消息隊列系統(tǒng)，而Jafka是在Kafka之上孵化而來的椎侠，即Kafka的一個升級版第租。具有以下特性：快速持久化，可以在O(1)的系統(tǒng)開銷下進行消息持久化我纪；高吞吐慎宾，在一臺普通的服務器上既可以達到10W/s的吞吐速率；完全的分布式系統(tǒng)浅悉，Broker趟据、Producer、Consumer都原生自動支持分布式术健，自動實現(xiàn)負載均衡汹碱；支持Hadoop數(shù)據(jù)并行加載，對于像Hadoop的一樣的日志數(shù)據(jù)和離線分析系統(tǒng)荞估，但又要求實時處理的限制咳促，這是一個可行的解決方案。Kafka通過Hadoop的并行加載機制來統(tǒng)一了在線和離線的消息處理勘伺。Apache Kafka相對于ActiveMQ是一個非常輕量級的消息系統(tǒng)跪腹，除了性能非常好之外，還是一個工作良好的分布式系統(tǒng)飞醉。
二 Terminology
- Broker：Kafka集群包含一個或多個服務器冲茸，這種服務器被稱為broker
- Topic：每條發(fā)布到Kafka集群的消息都有一個類別，這個類別被稱為topic。（物理上不同topic的消息分開存儲噪裕，邏輯上一個topic的消息雖然保存于一個或多個broker上但用戶只需指定消息的topic即可生產(chǎn)或消費數(shù)據(jù)而不必關(guān)心數(shù)據(jù)存于何處）
- Partition：parition是物理上的概念蹲盘，每個topic包含一個或多個partition，創(chuàng)建topic時可指定parition數(shù)量膳音。每個partition對應于一個文件夾，該文件夾下存儲該partition的數(shù)據(jù)和索引文件
- Producer：負責發(fā)布消息到Kafka broker
- Consumer：消費消息铃诬。每個consumer屬于一個特定的consumer group（可為每個consumer指定group name祭陷，若不指定group name則屬于默認的group）。使用consumer high level API時趣席，同一topic的一條消息只能被同一個consumer group內(nèi)的一個consumer消費兵志，但多個consumer group可同時消費這一消息。
三 kafka架構(gòu)
一個典型的kafka集群中包含若干producer（可以是web前端產(chǎn)生的page view宣肚，或者是服務器日志想罕，系統(tǒng)CPU、memory等）霉涨，若干broker（Kafka支持水平擴展按价，一般broker數(shù)量越多，集群吞吐率越高）笙瑟，若干consumer group楼镐，以及一個Zookeeper集群。Kafka通過Zookeeper管理集群配置往枷，選舉leader框产，以及在consumer group發(fā)生變化時進行rebalance。producer使用push模式將消息發(fā)布到broker错洁，consumer使用pull模式從broker訂閱并消費消息秉宿。
四 Push vs. Pull
作為一個messaging system，Kafka遵循了傳統(tǒng)的方式屯碴，選擇由producer向broker push消息并由consumer從broker pull消息描睦。一些logging-centric system，比如Facebook的Scribe和Cloudera的Flume,采用非常不同的push模式窿锉。事實上酌摇，push模式和pull模式各有優(yōu)劣。
push模式很難適應消費速率不同的消費者嗡载，因為消息發(fā)送速率是由broker決定的窑多。push模式的目標是盡可能以最快速度傳遞消息，但是這樣很容易造成consumer來不及處理消息洼滚，典型的表現(xiàn)就是拒絕服務以及網(wǎng)絡擁塞埂息。而pull模式則可以根據(jù)consumer的消費能力以適當?shù)乃俾氏M消息。
五 Topic & Partition
- Topic在邏輯上可以被認為是一個queue。每條消費都必須指定它的topic千康，可以簡單理解為必須指明把這條消息放進哪個queue里享幽。為了使得Kafka的吞吐率可以水平擴展，物理上把topic分成一個或多個partition拾弃，每個partition在物理上對應一個文件夾值桩，該文件夾下存儲這個partition的所有消息和索引文件。
- 因為每條消息都被append到該partition中豪椿，是順序?qū)懘疟P奔坟，因此效率非常高（經(jīng)驗證，順序?qū)懘疟P效率比隨機寫內(nèi)存還要高搭盾，這是Kafka高吞吐率的一個很重要的保證）咳秉。
- 每一條消息被發(fā)送到broker時，會根據(jù)paritition規(guī)則選擇被存儲到哪一個partition鸯隅。如果partition規(guī)則設置的合理澜建，所有消息可以均勻分布到不同的partition里，這樣就實現(xiàn)了水平擴展蝌以。（如果一個topic對應一個文件炕舵，那這個文件所在的機器I/O將會成為這個topic的性能瓶頸另玖，而partition解決了這個問題）吉嫩。在創(chuàng)建topic時可以在$KAFKA_HOME/config/server.properties中指定這個partition的數(shù)量(如下所示)，當然也可以在topic創(chuàng)建之后去修改parition數(shù)量敏释。
num.partitions=3 - 在發(fā)送一條消息時碍彭，可以指定這條消息的key晤硕，producer根據(jù)這個key和partition機制來判斷將這條消息發(fā)送到哪個parition。paritition機制可以通過指定producer的paritition. class這一參數(shù)來指定庇忌，該class必須實現(xiàn)kafka.producer.Partitioner接口舞箍。本例中如果key可以被解析為整數(shù)則將對應的整數(shù)與partition總數(shù)取余，該消息會被發(fā)送到該數(shù)對應的partition皆疹。（每個parition都會有個序號)
import kafka.producer.Partitioner;
import kafka.utils.VerifiableProperties;
public class JasonPartitioner<T> implements Partitioner {
public JasonPartitioner(VerifiableProperties verifiableProperties) {}
@Override
public int partition(Object key, int numPartitions) {
try {
int partitionNum = Integer.parseInt((String) key);
return Math.abs(Integer.parseInt((String) key) % numPartitions);
} catch (Exception e) {
return Math.abs(key.hashCode() % numPartitions);
}
}}
- 對于傳統(tǒng)的message queue而言疏橄，一般會刪除已經(jīng)被消費的消息，而Kafka集群會保留所有的消息略就，無論其被消費與否捎迫。當然，因為磁盤限制表牢，不可能永久保留所有數(shù)據(jù)（實際上也沒必要）窄绒，因此Kafka提供兩種策略去刪除舊數(shù)據(jù)。一是基于時間崔兴，二是基于partition文件大小彰导。例如可以通過配置$KAFKA_HOME/config/server.properties蛔翅，讓Kafka刪除一周前的數(shù)據(jù)，也可通過配置讓Kafka在partition文件超過1GB時刪除舊數(shù)據(jù)
# The minimum age of a log file to be eligible for deletion
log.retention.hours=168
# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824
# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000
# If log.cleaner.enable=true is set the cleaner will be enabled and individual logs
#can then be marked for log compaction.
log.cleaner.enable=false
- 這里要注意位谋，因為Kafka讀取特定消息的時間復雜度為O(1)山析，即與文件大小無關(guān)，所以這里刪除文件與Kafka性能無關(guān)掏父，選擇怎樣的刪除策略只與磁盤以及具體的需求有關(guān)笋轨。另外，Kafka會為每一個consumer group保留一些metadata信息–當前消費的消息的position赊淑，也即offset翩腐。這個offset由consumer控制。正常情況下consumer會在消費完一條消息后線性增加這個offset膏燃。當然，consumer也可將offset設成一個較小的值何什，重新消費一些消息组哩。因為offet由consumer控制，所以Kafka broker是無狀態(tài)的处渣，它不需要標記哪些消息被哪些consumer過伶贰，不需要通過broker去保證同一個consumer group只有一個consumer能消費某一條消息，因此也就不需要鎖機制罐栈，這也為Kafka的高吞吐率提供了有力保障黍衙。
六 Replication & Leader election
- Kafka從0.8開始提供partition級別的replication，replication的數(shù)量可在$KAFKA_HOME/config/server.properties中配置
default.replication.factor = 1
該 Replication與leader election配合提供了自動的failover機制荠诬。replication對Kafka的吞吐率是有一定影響的琅翻，但極大的增強了可用性。默認情況下柑贞，Kafka的replication數(shù)量為1方椎。
- 每個partition都有一個唯一的leader，所有的讀寫操作都在leader上完成钧嘶，follower批量從leader上pull數(shù)據(jù)棠众。一般情況下partition的數(shù)量大于等于broker的數(shù)量，并且所有partition的leader均勻分布在broker上有决。follower上的日志和其leader上的完全一樣闸拿。
- 和大部分分布式系統(tǒng)一樣，Kakfa處理失敗需要明確定義一個broker是否alive书幕。對于Kafka而言新荤，Kafka存活包含兩個條件，一是它必須維護與Zookeeper的session(這個通過Zookeeper的heartbeat機制來實現(xiàn))按咒。二是follower必須能夠及時將leader的writing復制過來迟隅，不能“落后太多”但骨。
- leader會track“in sync”的node list。如果一個follower宕機智袭，或者落后太多奔缠，leader將把它從”in sync” list中移除。這里所描述的“落后太多”指follower復制的消息落后于leader后的條數(shù)超過預定值吼野，該值可在$KAFKA_HOME/config/server.properties中配置
#If a replica falls more than this many messages behind the leader, the leader will remove the follower from ISR and treat it as dead
replica.lag.max.messages=4000
#If a follower hasn't sent any fetch requests for this window of time, the leader will remove the follower from ISR (in-sync replicas) and treat it as dead
replica.lag.time.max.ms=10000
- 需要說明的是校哎，Kafka只解決”fail/recover”，不處理“Byzantine”（“拜占庭”）問題瞳步。一條消息只有被“in sync” list里的所有follower都從leader復制過去才會被認為已提交闷哆。這樣就避免了部分數(shù)據(jù)被寫進了leader，還沒來得及被任何follower復制就宕機了单起，而造成數(shù)據(jù)丟失（consumer無法消費這些數(shù)據(jù)）抱怔。而對于producer而言，它可以選擇是否等待消息commit嘀倒，這可以通過request.required.acks來設置屈留。這種機制確保了只要“in sync” list有一個或以上的flollower，一條被commit的消息就不會丟失测蘑。
- 這里的復制機制即不是同步復制灌危，也不是單純的異步復制。事實上碳胳，同步復制要求“活著的”follower都復制完勇蝙，這條消息才會被認為commit，這種復制方式極大的影響了吞吐率（高吞吐率是Kafka非常重要的一個特性）挨约。而異步復制方式下味混，follower異步的從leader復制數(shù)據(jù)，數(shù)據(jù)只要被leader寫入log就被認為已經(jīng)commit烫罩，這種情況下如果follwer都落后于leader惜傲，而leader突然宕機，則會丟失數(shù)據(jù)贝攒。而Kafka的這種使用“in sync” list的方式則很好的均衡了確保數(shù)據(jù)不丟失以及吞吐率盗誊。follower可以批量的從leader復制數(shù)據(jù)，這樣極大的提高復制性能（批量寫磁盤）隘弊，極大減少了follower與leader的差距（前文有說到哈踱，只要follower落后leader不太遠，則被認為在“in sync” list里）梨熙。
- 另外一個很重要的問題是當leader宕機了开镣，怎樣在follower中選舉出新的leader。因為follower可能落后許多或者crash了咽扇，所以必須確保選擇“最新”的follower作為新的leader邪财。一個基本的原則就是陕壹，如果leader不在了，新的leader必須擁有原來的leader commit的所有消息树埠。這就需要作一個折衷糠馆，如果leader在標明一條消息被commit前等待更多的follower確認，那在它die之后就有更多的follower可以作為新的leader怎憋，但這也會造成吞吐率的下降又碌。
- 一種非常常用的選舉leader的方式是“majority vote”（“少數(shù)服從多數(shù)”），但Kafka并未采用這種方式绊袋。這種模式下毕匀，如果我們有2f+1個replica（包含leader和follower），那在commit之前必須保證有f+1個replica復制完消息癌别，為了保證正確選出新的leader皂岔，fail的replica不能超過f個。因為在剩下的任意f+1個replica里展姐，至少有一個replica包含有最新的所有消息凤薛。這種方式有個很大的優(yōu)勢，系統(tǒng)的latency只取決于最快的幾臺server诞仓，也就是說，如果replication factor是3速兔，那latency就取決于最快的那個follower而非最慢那個墅拭。majority vote也有一些劣勢，為了保證leader election的正常進行涣狗，它所能容忍的fail的follower個數(shù)比較少谍婉。如果要容忍1個follower掛掉，必須要有3個以上的replica镀钓，如果要容忍2個follower掛掉穗熬，必須要有5個以上的replica。也就是說丁溅，在生產(chǎn)環(huán)境下為了保證較高的容錯程度唤蔗，必須要有大量的replica，而大量的replica又會在大數(shù)據(jù)量下導致性能的急劇下降窟赏。這就是這種算法更多用在Zookeeper這種共享集群配置的系統(tǒng)中而很少在需要存儲大量數(shù)據(jù)的系統(tǒng)中使用的原因妓柜。例如HDFS的HA feature是基于majority-vote-based journal，但是它的數(shù)據(jù)存儲并沒有使用這種expensive的方式涯穷。
- Kafka在Zookeeper中動態(tài)維護了一個ISR（in-sync replicas） set棍掐，這個set里的所有replica都跟上了leader，只有ISR里的成員才有被選為leader的可能拷况。在這種模式下作煌，對于f+1個replica掘殴，一個Kafka topic能在保證不丟失已經(jīng)ommit的消息的前提下容忍f個replica的失敗。在大多數(shù)使用場景中粟誓，這種模式是非常有利的奏寨。事實上，為了容忍f個replica的失敗努酸，majority vote和ISR在commit前需要等待的replica數(shù)量是一樣的服爷，但是ISR需要的總的replica的個數(shù)幾乎是majority vote的一半。
- 雖然majority vote與ISR相比有不需等待最慢的server這一優(yōu)勢获诈，但是Kafka作者認為Kafka可以通過producer選擇是否被commit阻塞來改善這一問題仍源，并且節(jié)省下來的replica和磁盤使得ISR模式仍然值得。
- 上文提到舔涎，在ISR中至少有一個follower時笼踩，Kafka可以確保已經(jīng)commit的數(shù)據(jù)不丟失，但如果某一個partition的所有replica都掛了亡嫌，就無法保證數(shù)據(jù)不丟失了嚎于。這種情況下有兩種可行的方案：
- 等待ISR中的任一個replica“活”過來，并且選它作為leader
- 選擇第一個“活”過來的replica（不一定是ISR中的）作為leader
- 這就需要在可用性和一致性當中作出一個簡單的平衡挟冠。如果一定要等待ISR中的replica“活”過來于购，那不可用的時間就可能會相對較長。而且如果ISR中的所有replica都無法“活”過來了知染，或者數(shù)據(jù)都丟失了肋僧，這個partition將永遠不可用。選擇第一個“活”過來的replica作為leader控淡，而這個replica不是ISR中的replica嫌吠，那即使它并不保證已經(jīng)包含了所有已commit的消息，它也會成為leader而作為consumer的數(shù)據(jù)源（前文有說明掺炭，所有讀寫都由leader完成）辫诅。Kafka0.8.*使用了第二種方式。根據(jù)Kafka的文檔涧狮，在以后的版本中炕矮，Kafka支持用戶通過配置選擇這兩種方式中的一種，從而根據(jù)不同的使用場景選擇高可用性還是強一致性者冤。
- Kafka集群需要管理成百上千個partition吧享，Kafka通過round-robin的方式來平衡partition從而避免大量partition集中在了少數(shù)幾個節(jié)點上。同時Kafka也需要平衡leader的分布譬嚣，盡可能的讓所有partition的leader均勻分布在不同broker上钢颂。另一方面，優(yōu)化leadership election的過程也是很重要的拜银，畢竟這段時間相應的partition處于不可用狀態(tài)殊鞭。一種簡單的實現(xiàn)是暫停宕機的broker上的所有partition遭垛，并為之選舉leader。實際上操灿，Kafka選舉一個broker作為controller锯仪，這個controller通過watch Zookeeper檢測所有的broker failure，并負責為所有受影響的parition選舉leader趾盐，再將相應的leader調(diào)整命令發(fā)送至受影響的broker
七 Consumer group
- 每一個consumer實例都屬于一個consumer group庶喜，每一條消息只會被同一個consumer group里的一個consumer實例消費。（不同consumer group可以同時消費同一條消息）
- 很多傳統(tǒng)的message queue都會在消息被消費完后將消息刪除救鲤，一方面避免重復消費久窟，另一方面可以保證queue的長度比較少，提高效率本缠。而如上文所將斥扛，Kafka并不刪除已消費的消息，為了實現(xiàn)傳統(tǒng)message queue消息只被消費一次的語義丹锹，Kafka保證保證同一個consumer group里只有一個consumer會消費一條消息稀颁。與傳統(tǒng)message queue不同的是，Kafka還允許不同consumer group同時消費同一條消息楣黍，這一特性可以為消息的多元化處理提供了支持匾灶。實際上，Kafka的設計理念之一就是同時提供離線處理和實時處理租漂。根據(jù)這一特性粘昨，可以使用Storm這種實時流處理系統(tǒng)對消息進行實時在線處理，同時使用Hadoop這種批處理系統(tǒng)進行離線處理窜锯，還可以同時將數(shù)據(jù)實時備份到另一個數(shù)據(jù)中心，只需要保證這三個操作所使用的consumer在不同的consumer group即可芭析。
- 創(chuàng)建一個topic (名為topic1)锚扎，創(chuàng)建一個屬于group1的consumer實例，并創(chuàng)建三個屬于group2的consumer實例馁启，然后通過producer向topic1發(fā)送key分別為1驾孔，2，3r的消息惯疙。結(jié)果發(fā)現(xiàn)屬于group1的consumer收到了所有的這三條消息翠勉，同時group2中的3個consumer分別收到了key為1，2霉颠，3的消息
八 Consumer Rebalance
- Kafka保證同一consumer group中只有一個consumer會消費某條消息对碌，實際上，Kafka保證的是穩(wěn)定狀態(tài)下每一個consumer實例只會消費某一個或多個特定partition的數(shù)據(jù)蒿偎，而某個partition的數(shù)據(jù)只會被某一個特定的consumer實例所消費朽们。這樣設計的劣勢是無法讓同一個consumer group里的consumer均勻消費數(shù)據(jù)怀读，優(yōu)勢是每個consumer不用都跟大量的broker通信，減少通信開銷骑脱，同時也降低了分配難度菜枷，實現(xiàn)也更簡單。另外叁丧，因為同一個partition里的數(shù)據(jù)是有序的啤誊，這種設計可以保證每個partition里的數(shù)據(jù)也是有序被消費。
- 如果某consumer group中consumer數(shù)量少于partition數(shù)量拥娄，則至少有一個consumer會消費多個partition的數(shù)據(jù)蚊锹，如果consumer的數(shù)量與partition數(shù)量相同，則正好一個consumer消費一個partition的數(shù)據(jù)条舔，而如果consumer的數(shù)量多于partition的數(shù)量時枫耳，會有部分consumer無法消費該topic下任何一條消息。
- 如果topic1有0孟抗，1迁杨，2共三個partition，當group1只有一個consumer(名為consumer1)時凄硼，該 consumer可消費這3個partition的所有數(shù)據(jù)铅协。
- 增加一個consumer(consumer2)后，其中一個consumer（consumer1）可消費2個partition的數(shù)據(jù)摊沉，另外一個consumer(consumer2)可消費另外一個partition的數(shù)據(jù)
- 再增加一個consumer(consumer3)后狐史，每個consumer可消費一個partition的數(shù)據(jù)。consumer1消費partition0说墨，consumer2消費partition1骏全，consumer3消費partition2
- 再增加一個consumer（consumer4）后，其中3個consumer可分別消費一個partition的數(shù)據(jù)尼斧，另外一個consumer（consumer4）不能消費topic1任何數(shù)據(jù)姜贡。
- 此時關(guān)閉consumer1，剩下的consumer可分別消費一個partition的數(shù)據(jù)棺棵。
- 接著關(guān)閉consumer2楼咳，剩下的consumer3可消費2個partition，consumer4可消費1個partition烛恤。
- 再關(guān)閉consumer3母怜，剩下的consumer4可同時消費topic1的3個partition。
- 目前consumer rebalance的控制策略是由每一個consumer通過Zookeeper完成的缚柏。在這種策略下苹熏，每一個consumer或者broker的增加或者減少都會觸發(fā)consumer rebalance。因為每個consumer只負責調(diào)整自己所消費的partition，為了保證整個consumer group的一致性柜裸，所以當一個consumer觸發(fā)了rebalance時缕陕，該consumer group內(nèi)的其它所有consumer也應該同時觸發(fā)rebalance。
