摘自Jason’s Blog，原文鏈接　http://www.jasongj.com/2015/01/02/Kafka深度解析

一 kafka的簡(jiǎn)單理解

Kafka是Apache下的一個(gè)子項(xiàng)目，是一個(gè)高性能跨語(yǔ)言分布式發(fā)布/訂閱消息隊(duì)列系統(tǒng)，而Jafka是在Kafka之上孵化而來(lái)的，即Kafka的一個(gè)升級(jí)版。具有以下特性：快速持久化，可以在O(1)的系統(tǒng)開銷下進(jìn)行消息持久化；高吞吐，在一臺(tái)普通的服務(wù)器上既可以達(dá)到10W/s的吞吐速率；完全的分布式系統(tǒng)，Broker、Producer、Consumer都原生自動(dòng)支持分布式，自動(dòng)實(shí)現(xiàn)負(fù)載均衡；支持Hadoop數(shù)據(jù)并行加載，對(duì)于像Hadoop的一樣的日志數(shù)據(jù)和離線分析系統(tǒng)，但又要求實(shí)時(shí)處理的限制，這是一個(gè)可行的解決方案。Kafka通過(guò)Hadoop的并行加載機(jī)制來(lái)統(tǒng)一了在線和離線的消息處理。Apache Kafka相對(duì)于ActiveMQ是一個(gè)非常輕量級(jí)的消息系統(tǒng)，除了性能非常好之外，還是一個(gè)工作良好的分布式系統(tǒng)。

二 Terminology

1. Broker：Kafka集群包含一個(gè)或多個(gè)服務(wù)器，這種服務(wù)器被稱為broker
2. Topic：每條發(fā)布到Kafka集群的消息都有一個(gè)類別，這個(gè)類別被稱為topic。（物理上不同topic的消息分開存儲(chǔ)，邏輯上一個(gè)topic的消息雖然保存于一個(gè)或多個(gè)broker上但用戶只需指定消息的topic即可生產(chǎn)或消費(fèi)數(shù)據(jù)而不必關(guān)心數(shù)據(jù)存于何處）
3. Partition：parition是物理上的概念，每個(gè)topic包含一個(gè)或多個(gè)partition，創(chuàng)建topic時(shí)可指定parition數(shù)量。每個(gè)partition對(duì)應(yīng)于一個(gè)文件夾，該文件夾下存儲(chǔ)該partition的數(shù)據(jù)和索引文件
4. Producer：負(fù)責(zé)發(fā)布消息到Kafka broker
5. Consumer：消費(fèi)消息。每個(gè)consumer屬于一個(gè)特定的consumer group（可為每個(gè)consumer指定group name，若不指定group name則屬于默認(rèn)的group）。使用consumer high level API時(shí)，同一topic的一條消息只能被同一個(gè)consumer group內(nèi)的一個(gè)consumer消費(fèi)，但多個(gè)consumer group可同時(shí)消費(fèi)這一消息。

三 kafka架構(gòu)

一個(gè)典型的kafka集群中包含若干producer（可以是web前端產(chǎn)生的page view，或者是服務(wù)器日志，系統(tǒng)CPU、memory等），若干broker（Kafka支持水平擴(kuò)展，一般broker數(shù)量越多，集群吞吐率越高），若干consumer group，以及一個(gè)Zookeeper集群。Kafka通過(guò)Zookeeper管理集群配置，選舉leader，以及在consumer group發(fā)生變化時(shí)進(jìn)行rebalance。producer使用push模式將消息發(fā)布到broker，consumer使用pull模式從broker訂閱并消費(fèi)消息。

四 Push vs. Pull

作為一個(gè)messaging system，Kafka遵循了傳統(tǒng)的方式，選擇由producer向broker push消息并由consumer從broker pull消息。一些logging-centric system，比如Facebook的Scribe和Cloudera的Flume,采用非常不同的push模式。事實(shí)上，push模式和pull模式各有優(yōu)劣。
　　push模式很難適應(yīng)消費(fèi)速率不同的消費(fèi)者，因?yàn)橄l(fā)送速率是由broker決定的。push模式的目標(biāo)是盡可能以最快速度傳遞消息，但是這樣很容易造成consumer來(lái)不及處理消息，典型的表現(xiàn)就是拒絕服務(wù)以及網(wǎng)絡(luò)擁塞。而pull模式則可以根據(jù)consumer的消費(fèi)能力以適當(dāng)?shù)乃俾氏M(fèi)消息。

五 Topic & Partition

1. Topic在邏輯上可以被認(rèn)為是一個(gè)queue。每條消費(fèi)都必須指定它的topic，可以簡(jiǎn)單理解為必須指明把這條消息放進(jìn)哪個(gè)queue里。為了使得Kafka的吞吐率可以水平擴(kuò)展，物理上把topic分成一個(gè)或多個(gè)partition，每個(gè)partition在物理上對(duì)應(yīng)一個(gè)文件夾，該文件夾下存儲(chǔ)這個(gè)partition的所有消息和索引文件。
2. 因?yàn)槊織l消息都被append到該partition中，是順序?qū)懘疟P，因此效率非常高（經(jīng)驗(yàn)證，順序?qū)懘疟P效率比隨機(jī)寫內(nèi)存還要高，這是Kafka高吞吐率的一個(gè)很重要的保證）。
3. 每一條消息被發(fā)送到broker時(shí)，會(huì)根據(jù)paritition規(guī)則選擇被存儲(chǔ)到哪一個(gè)partition。如果partition規(guī)則設(shè)置的合理，所有消息可以均勻分布到不同的partition里，這樣就實(shí)現(xiàn)了水平擴(kuò)展。（如果一個(gè)topic對(duì)應(yīng)一個(gè)文件，那這個(gè)文件所在的機(jī)器I/O將會(huì)成為這個(gè)topic的性能瓶頸，而partition解決了這個(gè)問題）。在創(chuàng)建topic時(shí)可以在$KAFKA_HOME/config/server.properties中指定這個(gè)partition的數(shù)量(如下所示)，當(dāng)然也可以在topic創(chuàng)建之后去修改parition數(shù)量。
   num.partitions=3
4. 在發(fā)送一條消息時(shí)，可以指定這條消息的key，producer根據(jù)這個(gè)key和partition機(jī)制來(lái)判斷將這條消息發(fā)送到哪個(gè)parition。paritition機(jī)制可以通過(guò)指定producer的paritition. class這一參數(shù)來(lái)指定，該class必須實(shí)現(xiàn)kafka.producer.Partitioner接口。本例中如果key可以被解析為整數(shù)則將對(duì)應(yīng)的整數(shù)與partition總數(shù)取余，該消息會(huì)被發(fā)送到該數(shù)對(duì)應(yīng)的partition。（每個(gè)parition都會(huì)有個(gè)序號(hào))

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);
        }
    }}

5. 對(duì)于傳統(tǒng)的message queue而言，一般會(huì)刪除已經(jīng)被消費(fèi)的消息，而Kafka集群會(huì)保留所有的消息，無(wú)論其被消費(fèi)與否。當(dāng)然，因?yàn)榇疟P限制，不可能永久保留所有數(shù)據(jù)（實(shí)際上也沒必要），因此Kafka提供兩種策略去刪除舊數(shù)據(jù)。一是基于時(shí)間，二是基于partition文件大小。例如可以通過(guò)配置$KAFKA_HOME/config/server.properties，讓Kafka刪除一周前的數(shù)據(jù)，也可通過(guò)配置讓Kafka在partition文件超過(guò)1GB時(shí)刪除舊數(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

6. 這里要注意，因?yàn)镵afka讀取特定消息的時(shí)間復(fù)雜度為O(1)，即與文件大小無(wú)關(guān)，所以這里刪除文件與Kafka性能無(wú)關(guān)，選擇怎樣的刪除策略只與磁盤以及具體的需求有關(guān)。另外，Kafka會(huì)為每一個(gè)consumer group保留一些metadata信息–當(dāng)前消費(fèi)的消息的position，也即offset。這個(gè)offset由consumer控制。正常情況下consumer會(huì)在消費(fèi)完一條消息后線性增加這個(gè)offset。當(dāng)然，consumer也可將offset設(shè)成一個(gè)較小的值，重新消費(fèi)一些消息。因?yàn)閛ffet由consumer控制，所以Kafka broker是無(wú)狀態(tài)的，它不需要標(biāo)記哪些消息被哪些consumer過(guò)，不需要通過(guò)broker去保證同一個(gè)consumer group只有一個(gè)consumer能消費(fèi)某一條消息，因此也就不需要鎖機(jī)制，這也為Kafka的高吞吐率提供了有力保障。

六 Replication & Leader election

1. Kafka從0.8開始提供partition級(jí)別的replication，replication的數(shù)量可在$KAFKA_HOME/config/server.properties中配置

default.replication.factor = 1

該 Replication與leader election配合提供了自動(dòng)的failover機(jī)制。replication對(duì)Kafka的吞吐率是有一定影響的，但極大的增強(qiáng)了可用性。默認(rèn)情況下，Kafka的replication數(shù)量為1。

2. 每個(gè)partition都有一個(gè)唯一的leader，所有的讀寫操作都在leader上完成，follower批量從leader上pull數(shù)據(jù)。一般情況下partition的數(shù)量大于等于broker的數(shù)量，并且所有partition的leader均勻分布在broker上。follower上的日志和其leader上的完全一樣。
3. 和大部分分布式系統(tǒng)一樣，Kakfa處理失敗需要明確定義一個(gè)broker是否alive。對(duì)于Kafka而言，Kafka存活包含兩個(gè)條件，一是它必須維護(hù)與Zookeeper的session(這個(gè)通過(guò)Zookeeper的heartbeat機(jī)制來(lái)實(shí)現(xiàn))。二是follower必須能夠及時(shí)將leader的writing復(fù)制過(guò)來(lái)，不能“落后太多”。
4. leader會(huì)track“in sync”的node list。如果一個(gè)follower宕機(jī)，或者落后太多，leader將把它從”in sync” list中移除。這里所描述的“落后太多”指follower復(fù)制的消息落后于leader后的條數(shù)超過(guò)預(yù)定值，該值可在$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

5. 需要說(shuō)明的是，Kafka只解決”fail/recover”，不處理“Byzantine”（“拜占庭”）問題。一條消息只有被“in sync” list里的所有follower都從leader復(fù)制過(guò)去才會(huì)被認(rèn)為已提交。這樣就避免了部分?jǐn)?shù)據(jù)被寫進(jìn)了leader，還沒來(lái)得及被任何follower復(fù)制就宕機(jī)了，而造成數(shù)據(jù)丟失（consumer無(wú)法消費(fèi)這些數(shù)據(jù)）。而對(duì)于producer而言，它可以選擇是否等待消息commit，這可以通過(guò)request.required.acks來(lái)設(shè)置。這種機(jī)制確保了只要“in sync” list有一個(gè)或以上的flollower，一條被commit的消息就不會(huì)丟失。
6. 這里的復(fù)制機(jī)制即不是同步復(fù)制，也不是單純的異步復(fù)制。事實(shí)上，同步復(fù)制要求“活著的”follower都復(fù)制完，這條消息才會(huì)被認(rèn)為commit，這種復(fù)制方式極大的影響了吞吐率（高吞吐率是Kafka非常重要的一個(gè)特性）。而異步復(fù)制方式下，follower異步的從leader復(fù)制數(shù)據(jù)，數(shù)據(jù)只要被leader寫入log就被認(rèn)為已經(jīng)commit，這種情況下如果follwer都落后于leader，而leader突然宕機(jī)，則會(huì)丟失數(shù)據(jù)。而Kafka的這種使用“in sync” list的方式則很好的均衡了確保數(shù)據(jù)不丟失以及吞吐率。follower可以批量的從leader復(fù)制數(shù)據(jù)，這樣極大的提高復(fù)制性能（批量寫磁盤），極大減少了follower與leader的差距（前文有說(shuō)到，只要follower落后leader不太遠(yuǎn)，則被認(rèn)為在“in sync” list里）。
7. 另外一個(gè)很重要的問題是當(dāng)leader宕機(jī)了，怎樣在follower中選舉出新的leader。因?yàn)閒ollower可能落后許多或者crash了，所以必須確保選擇“最新”的follower作為新的leader。一個(gè)基本的原則就是，如果leader不在了，新的leader必須擁有原來(lái)的leader commit的所有消息。這就需要作一個(gè)折衷，如果leader在標(biāo)明一條消息被commit前等待更多的follower確認(rèn)，那在它die之后就有更多的follower可以作為新的leader，但這也會(huì)造成吞吐率的下降。
8. 一種非常常用的選舉leader的方式是“majority vote”（“少數(shù)服從多數(shù)”），但Kafka并未采用這種方式。這種模式下，如果我們有2f+1個(gè)replica（包含leader和follower），那在commit之前必須保證有f+1個(gè)replica復(fù)制完消息，為了保證正確選出新的leader，fail的replica不能超過(guò)f個(gè)。因?yàn)樵谑Ｏ碌娜我鈌+1個(gè)replica里，至少有一個(gè)replica包含有最新的所有消息。這種方式有個(gè)很大的優(yōu)勢(shì)，系統(tǒng)的latency只取決于最快的幾臺(tái)server，也就是說(shuō)，如果replication factor是3，那latency就取決于最快的那個(gè)follower而非最慢那個(gè)。majority vote也有一些劣勢(shì)，為了保證leader election的正常進(jìn)行，它所能容忍的fail的follower個(gè)數(shù)比較少。如果要容忍1個(gè)follower掛掉，必須要有3個(gè)以上的replica，如果要容忍2個(gè)follower掛掉，必須要有5個(gè)以上的replica。也就是說(shuō)，在生產(chǎn)環(huán)境下為了保證較高的容錯(cuò)程度，必須要有大量的replica，而大量的replica又會(huì)在大數(shù)據(jù)量下導(dǎo)致性能的急劇下降。這就是這種算法更多用在Zookeeper這種共享集群配置的系統(tǒng)中而很少在需要存儲(chǔ)大量數(shù)據(jù)的系統(tǒng)中使用的原因。例如HDFS的HA feature是基于majority-vote-based journal，但是它的數(shù)據(jù)存儲(chǔ)并沒有使用這種expensive的方式。
9. Kafka在Zookeeper中動(dòng)態(tài)維護(hù)了一個(gè)ISR（in-sync replicas） set，這個(gè)set里的所有replica都跟上了leader，只有ISR里的成員才有被選為leader的可能。在這種模式下，對(duì)于f+1個(gè)replica，一個(gè)Kafka topic能在保證不丟失已經(jīng)ommit的消息的前提下容忍f個(gè)replica的失敗。在大多數(shù)使用場(chǎng)景中，這種模式是非常有利的。事實(shí)上，為了容忍f個(gè)replica的失敗，majority vote和ISR在commit前需要等待的replica數(shù)量是一樣的，但是ISR需要的總的replica的個(gè)數(shù)幾乎是majority vote的一半。
10. 雖然majority vote與ISR相比有不需等待最慢的server這一優(yōu)勢(shì)，但是Kafka作者認(rèn)為Kafka可以通過(guò)producer選擇是否被commit阻塞來(lái)改善這一問題，并且節(jié)省下來(lái)的replica和磁盤使得ISR模式仍然值得。
11. 上文提到，在ISR中至少有一個(gè)follower時(shí)，Kafka可以確保已經(jīng)commit的數(shù)據(jù)不丟失，但如果某一個(gè)partition的所有replica都掛了，就無(wú)法保證數(shù)據(jù)不丟失了。這種情況下有兩種可行的方案：

• 等待ISR中的任一個(gè)replica“活”過(guò)來(lái)，并且選它作為leader
• 選擇第一個(gè)“活”過(guò)來(lái)的replica（不一定是ISR中的）作為leader

12. 這就需要在可用性和一致性當(dāng)中作出一個(gè)簡(jiǎn)單的平衡。如果一定要等待ISR中的replica“活”過(guò)來(lái)，那不可用的時(shí)間就可能會(huì)相對(duì)較長(zhǎng)。而且如果ISR中的所有replica都無(wú)法“活”過(guò)來(lái)了，或者數(shù)據(jù)都丟失了，這個(gè)partition將永遠(yuǎn)不可用。選擇第一個(gè)“活”過(guò)來(lái)的replica作為leader，而這個(gè)replica不是ISR中的replica，那即使它并不保證已經(jīng)包含了所有已commit的消息，它也會(huì)成為leader而作為consumer的數(shù)據(jù)源（前文有說(shuō)明，所有讀寫都由leader完成）。Kafka0.8.*使用了第二種方式。根據(jù)Kafka的文檔，在以后的版本中，Kafka支持用戶通過(guò)配置選擇這兩種方式中的一種，從而根據(jù)不同的使用場(chǎng)景選擇高可用性還是強(qiáng)一致性。
13. Kafka集群需要管理成百上千個(gè)partition，Kafka通過(guò)round-robin的方式來(lái)平衡partition從而避免大量partition集中在了少數(shù)幾個(gè)節(jié)點(diǎn)上。同時(shí)Kafka也需要平衡leader的分布，盡可能的讓所有partition的leader均勻分布在不同broker上。另一方面，優(yōu)化leadership election的過(guò)程也是很重要的，畢竟這段時(shí)間相應(yīng)的partition處于不可用狀態(tài)。一種簡(jiǎn)單的實(shí)現(xiàn)是暫停宕機(jī)的broker上的所有partition，并為之選舉leader。實(shí)際上，Kafka選舉一個(gè)broker作為controller，這個(gè)controller通過(guò)watch Zookeeper檢測(cè)所有的broker failure，并負(fù)責(zé)為所有受影響的parition選舉leader，再將相應(yīng)的leader調(diào)整命令發(fā)送至受影響的broker

七 Consumer group

1. 每一個(gè)consumer實(shí)例都屬于一個(gè)consumer group，每一條消息只會(huì)被同一個(gè)consumer group里的一個(gè)consumer實(shí)例消費(fèi)。（不同consumer group可以同時(shí)消費(fèi)同一條消息）
2. 很多傳統(tǒng)的message queue都會(huì)在消息被消費(fèi)完后將消息刪除，一方面避免重復(fù)消費(fèi)，另一方面可以保證queue的長(zhǎng)度比較少，提高效率。而如上文所將，Kafka并不刪除已消費(fèi)的消息，為了實(shí)現(xiàn)傳統(tǒng)message queue消息只被消費(fèi)一次的語(yǔ)義，Kafka保證保證同一個(gè)consumer group里只有一個(gè)consumer會(huì)消費(fèi)一條消息。與傳統(tǒng)message queue不同的是，Kafka還允許不同consumer group同時(shí)消費(fèi)同一條消息，這一特性可以為消息的多元化處理提供了支持。實(shí)際上，Kafka的設(shè)計(jì)理念之一就是同時(shí)提供離線處理和實(shí)時(shí)處理。根據(jù)這一特性，可以使用Storm這種實(shí)時(shí)流處理系統(tǒng)對(duì)消息進(jìn)行實(shí)時(shí)在線處理，同時(shí)使用Hadoop這種批處理系統(tǒng)進(jìn)行離線處理，還可以同時(shí)將數(shù)據(jù)實(shí)時(shí)備份到另一個(gè)數(shù)據(jù)中心，只需要保證這三個(gè)操作所使用的consumer在不同的consumer group即可。
3. 創(chuàng)建一個(gè)topic (名為topic1)，創(chuàng)建一個(gè)屬于group1的consumer實(shí)例，并創(chuàng)建三個(gè)屬于group2的consumer實(shí)例，然后通過(guò)producer向topic1發(fā)送key分別為1，2，3r的消息。結(jié)果發(fā)現(xiàn)屬于group1的consumer收到了所有的這三條消息，同時(shí)group2中的3個(gè)consumer分別收到了key為1，2，3的消息

八 Consumer Rebalance

1. Kafka保證同一consumer group中只有一個(gè)consumer會(huì)消費(fèi)某條消息，實(shí)際上，Kafka保證的是穩(wěn)定狀態(tài)下每一個(gè)consumer實(shí)例只會(huì)消費(fèi)某一個(gè)或多個(gè)特定partition的數(shù)據(jù)，而某個(gè)partition的數(shù)據(jù)只會(huì)被某一個(gè)特定的consumer實(shí)例所消費(fèi)。這樣設(shè)計(jì)的劣勢(shì)是無(wú)法讓同一個(gè)consumer group里的consumer均勻消費(fèi)數(shù)據(jù)，優(yōu)勢(shì)是每個(gè)consumer不用都跟大量的broker通信，減少通信開銷，同時(shí)也降低了分配難度，實(shí)現(xiàn)也更簡(jiǎn)單。另外，因?yàn)橥粋€(gè)partition里的數(shù)據(jù)是有序的，這種設(shè)計(jì)可以保證每個(gè)partition里的數(shù)據(jù)也是有序被消費(fèi)。
2. 如果某consumer group中consumer數(shù)量少于partition數(shù)量，則至少有一個(gè)consumer會(huì)消費(fèi)多個(gè)partition的數(shù)據(jù)，如果consumer的數(shù)量與partition數(shù)量相同，則正好一個(gè)consumer消費(fèi)一個(gè)partition的數(shù)據(jù)，而如果consumer的數(shù)量多于partition的數(shù)量時(shí)，會(huì)有部分consumer無(wú)法消費(fèi)該topic下任何一條消息。

• 如果topic1有0，1，2共三個(gè)partition，當(dāng)group1只有一個(gè)consumer(名為consumer1)時(shí)，該 consumer可消費(fèi)這3個(gè)partition的所有數(shù)據(jù)。
• 增加一個(gè)consumer(consumer2)后，其中一個(gè)consumer（consumer1）可消費(fèi)2個(gè)partition的數(shù)據(jù)，另外一個(gè)consumer(consumer2)可消費(fèi)另外一個(gè)partition的數(shù)據(jù)
• 再增加一個(gè)consumer(consumer3)后，每個(gè)consumer可消費(fèi)一個(gè)partition的數(shù)據(jù)。consumer1消費(fèi)partition0，consumer2消費(fèi)partition1，consumer3消費(fèi)partition2
• 再增加一個(gè)consumer（consumer4）后，其中3個(gè)consumer可分別消費(fèi)一個(gè)partition的數(shù)據(jù)，另外一個(gè)consumer（consumer4）不能消費(fèi)topic1任何數(shù)據(jù)。
• 此時(shí)關(guān)閉consumer1，剩下的consumer可分別消費(fèi)一個(gè)partition的數(shù)據(jù)。
• 接著關(guān)閉consumer2，剩下的consumer3可消費(fèi)2個(gè)partition，consumer4可消費(fèi)1個(gè)partition。
• 再關(guān)閉consumer3，剩下的consumer4可同時(shí)消費(fèi)topic1的3個(gè)partition。

3. 目前consumer rebalance的控制策略是由每一個(gè)consumer通過(guò)Zookeeper完成的。在這種策略下，每一個(gè)consumer或者broker的增加或者減少都會(huì)觸發(fā)consumer rebalance。因?yàn)槊總€(gè)consumer只負(fù)責(zé)調(diào)整自己所消費(fèi)的partition，為了保證整個(gè)consumer group的一致性，所以當(dāng)一個(gè)consumer觸發(fā)了rebalance時(shí)，該consumer group內(nèi)的其它所有consumer也應(yīng)該同時(shí)觸發(fā)rebalance。