背景介紹

Kafka簡(jiǎn)介

Kafka是一種分布式的，基于發(fā)布/訂閱的消息系統(tǒng)。主要設(shè)計(jì)目標(biāo)如下：

• 以時(shí)間復(fù)雜度為O(1)的方式提供消息持久化能力，即使對(duì)TB級(jí)以上數(shù)據(jù)也能保證常數(shù)時(shí)間的訪問性能
• 高吞吐率。即使在非常廉價(jià)的商用機(jī)器上也能做到單機(jī)支持每秒100K條消息的傳輸
• 支持Kafka Server間的消息分區(qū)，及分布式消費(fèi)，同時(shí)保證每個(gè)partition內(nèi)的消息順序傳輸
• 同時(shí)支持離線數(shù)據(jù)處理和實(shí)時(shí)數(shù)據(jù)處理

為什么要用Message Queue

• 解耦
  在項(xiàng)目啟動(dòng)之初來預(yù)測(cè)將來項(xiàng)目會(huì)碰到什么需求，是極其困難的。消息隊(duì)列在處理過程中間插入了一個(gè)隱含的、基于數(shù)據(jù)的接口層，兩邊的處理過程都要實(shí)現(xiàn)這一接口。這允許你獨(dú)立的擴(kuò)展或修改兩邊的處理過程，只要確保它們遵守同樣的接口約束

• 冗余
  有些情況下，處理數(shù)據(jù)的過程會(huì)失敗。除非數(shù)據(jù)被持久化，否則將造成丟失。消息隊(duì)列把數(shù)據(jù)進(jìn)行持久化直到它們已經(jīng)被完全處理，通過這一方式規(guī)避了數(shù)據(jù)丟失風(fēng)險(xiǎn)。在被許多消息隊(duì)列所采用的”插入-獲取-刪除”范式中，在把一個(gè)消息從隊(duì)列中刪除之前，需要你的處理過程明確的指出該消息已經(jīng)被處理完畢，確保你的數(shù)據(jù)被安全的保存直到你使用完畢。

• 擴(kuò)展性
  因?yàn)橄㈥?duì)列解耦了你的處理過程，所以增大消息入隊(duì)和處理的頻率是很容易的；只要另外增加處理過程即可。不需要改變代碼、不需要調(diào)節(jié)參數(shù)。擴(kuò)展就像調(diào)大電力按鈕一樣簡(jiǎn)單。

• 靈活性 & 峰值處理能力
  在訪問量劇增的情況下，應(yīng)用仍然需要繼續(xù)發(fā)揮作用，但是這樣的突發(fā)流量并不常見；如果為以能處理這類峰值訪問為標(biāo)準(zhǔn)來投入資源隨時(shí)待命無疑是巨大的浪費(fèi)。使用消息隊(duì)列能夠使關(guān)鍵組件頂住突發(fā)的訪問壓力，而不會(huì)因?yàn)橥话l(fā)的超負(fù)荷的請(qǐng)求而完全崩潰。

• 可恢復(fù)性
  當(dāng)體系的一部分組件失效，不會(huì)影響到整個(gè)系統(tǒng)。消息隊(duì)列降低了進(jìn)程間的耦合度，所以即使一個(gè)處理消息的進(jìn)程掛掉，加入隊(duì)列中的消息仍然可以在系統(tǒng)恢復(fù)后被處理。而這種允許重試或者延后處理請(qǐng)求的能力通常是造就一個(gè)略感不便的用戶和一個(gè)沮喪透頂?shù)挠脩糁g的區(qū)別。

• 送達(dá)保證
  消息隊(duì)列提供的冗余機(jī)制保證了消息能被實(shí)際的處理，只要一個(gè)進(jìn)程讀取了該隊(duì)列即可。在此基礎(chǔ)上，IronMQ提供了一個(gè)”只送達(dá)一次”保證。無論有多少進(jìn)程在從隊(duì)列中領(lǐng)取數(shù)據(jù)，每一個(gè)消息只能被處理一次。這之所以成為可能，是因?yàn)楂@取一個(gè)消息只是”預(yù)定”了這個(gè)消息，暫時(shí)把它移出了隊(duì)列。除非客戶端明確的表示已經(jīng)處理完了這個(gè)消息，否則這個(gè)消息會(huì)被放回隊(duì)列中去，在一段可配置的時(shí)間之后可再次被處理。

• 順序保證
  在大多使用場(chǎng)景下，數(shù)據(jù)處理的順序都很重要。消息隊(duì)列本來就是排序的，并且能保證數(shù)據(jù)會(huì)按照特定的順序來處理。IronMO保證消息通過FIFO（先進(jìn)先出）的順序來處理，因此消息在隊(duì)列中的位置就是從隊(duì)列中檢索他們的位置。

• 緩沖
  在任何重要的系統(tǒng)中，都會(huì)有需要不同的處理時(shí)間的元素。例如,加載一張圖片比應(yīng)用過濾器花費(fèi)更少的時(shí)間。消息隊(duì)列通過一個(gè)緩沖層來幫助任務(wù)***效率的執(zhí)行—寫入隊(duì)列的處理會(huì)盡可能的快速，而不受從隊(duì)列讀的預(yù)備處理的約束。該緩沖有助于控制和優(yōu)化數(shù)據(jù)流經(jīng)過系統(tǒng)的速度。

• 理解數(shù)據(jù)流
  在一個(gè)分布式系統(tǒng)里，要得到一個(gè)關(guān)于用戶操作會(huì)用多長(zhǎng)時(shí)間及其原因的總體印象，是個(gè)巨大的挑戰(zhàn)。消息系列通過消息被處理的頻率，來方便的輔助確定那些表現(xiàn)不佳的處理過程或領(lǐng)域，這些地方的數(shù)據(jù)流都不夠優(yōu)化。

• 異步通信
  很多時(shí)候，你不想也不需要立即處理消息。消息隊(duì)列提供了異步處理機(jī)制，允許你把一個(gè)消息放入隊(duì)列，但并不立即處理它。你想向隊(duì)列中放入多少消息就放多少，然后在你樂意的時(shí)候再去處理它們。

常用Message Queue對(duì)比

• RabbitMQ
  RabbitMQ是使用Erlang編寫的一個(gè)開源的消息隊(duì)列，本身支持很多的協(xié)議：AMQP，XMPP, SMTP, STOMP，也正因如此，它非常重量級(jí)，更適合于企業(yè)級(jí)的開發(fā)。同時(shí)實(shí)現(xiàn)了Broker構(gòu)架，這意味著消息在發(fā)送給客戶端時(shí)先在中心隊(duì)列排隊(duì)。對(duì)路由，負(fù)載均衡或者數(shù)據(jù)持久化都有很好的支持。

• Redis
  Redis是一個(gè)基于Key-Value對(duì)的NoSQL數(shù)據(jù)庫，開發(fā)維護(hù)很活躍。雖然它是一個(gè)Key-Value數(shù)據(jù)庫存儲(chǔ)系統(tǒng)，但它本身支持MQ功能，所以完全可以當(dāng)做一個(gè)輕量級(jí)的隊(duì)列服務(wù)來使用。對(duì)于RabbitMQ和Redis的入隊(duì)和出隊(duì)操作，各執(zhí)行100萬次，每10萬次記錄一次執(zhí)行時(shí)間。測(cè)試數(shù)據(jù)分為128Bytes、512Bytes、1K和10K四個(gè)不同大小的數(shù)據(jù)。實(shí)驗(yàn)表明：入隊(duì)時(shí)，當(dāng)數(shù)據(jù)比較小時(shí)Redis的性能要高于RabbitMQ，而如果數(shù)據(jù)大小超過了10K，Redis則慢的無法忍受；出隊(duì)時(shí)，無論數(shù)據(jù)大小，Redis都表現(xiàn)出非常好的性能，而RabbitMQ的出隊(duì)性能則遠(yuǎn)低于Redis。

• ZeroMQ
  ZeroMQ號(hào)稱最快的消息隊(duì)列系統(tǒng)，尤其針對(duì)大吞吐量的需求場(chǎng)景。ZMQ能夠?qū)崿F(xiàn)RabbitMQ不擅長(zhǎng)的高級(jí)/復(fù)雜的隊(duì)列，但是開發(fā)人員需要自己組合多種技術(shù)框架，技術(shù)上的復(fù)雜度是對(duì)這MQ能夠應(yīng)用成功的挑戰(zhàn)。ZeroMQ具有一個(gè)獨(dú)特的非中間件的模式，你不需要安裝和運(yùn)行一個(gè)消息服務(wù)器或中間件，因?yàn)槟愕膽?yīng)用程序?qū)缪萘诉@個(gè)服務(wù)角色。你只需要簡(jiǎn)單的引用ZeroMQ程序庫，可以使用NuGet安裝，然后你就可以愉快的在應(yīng)用程序之間發(fā)送消息了。但是ZeroMQ僅提供非持久性的隊(duì)列，也就是說如果宕機(jī)，數(shù)據(jù)將會(huì)丟失。其中，Twitter的Storm 0.9.0以前的版本中默認(rèn)使用ZeroMQ作為數(shù)據(jù)流的傳輸（Storm從0.9版本開始同時(shí)支持ZeroMQ和Netty作為傳輸模塊）。

• ActiveMQ
  ActiveMQ是Apache下的一個(gè)子項(xiàng)目。 類似于ZeroMQ，它能夠以代理人和點(diǎn)對(duì)點(diǎn)的技術(shù)實(shí)現(xiàn)隊(duì)列。同時(shí)類似于RabbitMQ，它少量代碼就可以高效地實(shí)現(xiàn)高級(jí)應(yīng)用場(chǎng)景。

• Kafka/Jafka
  Kafka是Apache下的一個(gè)子項(xiàng)目，是一個(gè)高性能跨語言分布式發(fā)布/訂閱消息隊(duì)列系統(tǒng)，而Jafka是在Kafka之上孵化而來的，即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通過Hadoop的并行加載機(jī)制來統(tǒng)一了在線和離線的消息處理。Apache Kafka相對(duì)于ActiveMQ是一個(gè)非常輕量級(jí)的消息系統(tǒng)，除了性能非常好之外，還是一個(gè)工作良好的分布式系統(tǒng)。

Kafka解析

Terminology

• Broker
  Kafka集群包含一個(gè)或多個(gè)服務(wù)器，這種服務(wù)器被稱為broker
• Topic
  每條發(fā)布到Kafka集群的消息都有一個(gè)類別，這個(gè)類別被稱為topic。（物理上不同topic的消息分開存儲(chǔ)，邏輯上一個(gè)topic的消息雖然保存于一個(gè)或多個(gè)broker上但用戶只需指定消息的topic即可生產(chǎn)或消費(fèi)數(shù)據(jù)而不必關(guān)心數(shù)據(jù)存于何處）
• Partition
  parition是物理上的概念，每個(gè)topic包含一個(gè)或多個(gè)partition，創(chuàng)建topic時(shí)可指定parition數(shù)量。每個(gè)partition對(duì)應(yīng)于一個(gè)文件夾，該文件夾下存儲(chǔ)該partition的數(shù)據(jù)和索引文件
• Producer
  負(fù)責(zé)發(fā)布消息到Kafka broker
• Consumer
  消費(fèi)消息。每個(gè)consumer屬于一個(gè)特定的consuer 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通過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來不及處理消息，典型的表現(xiàn)就是拒絕服務(wù)以及網(wǎng)絡(luò)擁塞。而pull模式則可以根據(jù)consumer的消費(fèi)能力以適當(dāng)?shù)乃俾氏M(fèi)消息。

Topic & Partition

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的所有消息和索引文件。

每個(gè)日志文件都是“log entries”序列，每一個(gè)log entry包含一個(gè)4字節(jié)整型數(shù)（值為N），其后跟N個(gè)字節(jié)的消息體。每條消息都有一個(gè)當(dāng)前partition下唯一的64字節(jié)的offset，它指明了這條消息的起始位置。磁盤上存儲(chǔ)的消費(fèi)格式如下：

message length ： 4 bytes (value: 1+4+n)

“magic” value ： 1 byte

crc ： 4 bytes

payload ： n bytes

這個(gè)“log entries”并非由一個(gè)文件構(gòu)成，而是分成多個(gè)segment，每個(gè)segment名為該segment***條消息的offset和“.kafka”組成。另外會(huì)有一個(gè)索引文件，它標(biāo)明了每個(gè)segment下包含的log entry的offset范圍，如下圖所示。

因?yàn)槊織l消息都被append到該partition中，是順序?qū)懘疟P，因此效率非常高（經(jīng)驗(yàn)證，順序?qū)懘疟P效率比隨機(jī)寫內(nèi)存還要高，這是Kafka高吞吐率的一個(gè)很重要的保證）。

每一條消息被發(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ù)量。

# The default number of log partitions per topic. More partitions allow greater  
# parallelism for consumption, but this will also result in more files across  
# the brokers.  
num.partitions=3 

在發(fā)送一條消息時(shí)，可以指定這條消息的key，producer根據(jù)這個(gè)key和partition機(jī)制來判斷將這條消息發(fā)送到哪個(gè)parition。paritition機(jī)制可以通過指定producer的paritition. class這一參數(shù)來指定，該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);  
        }  
    }  
}  

如果將上例中的class作為partition.class，并通過如下代碼發(fā)送20條消息（key分別為0，1，2，3）至topic2（包含4個(gè)partition）。

public void sendMessage() throws InterruptedException{  
for(int i = 1; i <= 5; i++){  
      List messageList = new ArrayList<KeyedMessage<String, String>>();  
      for(int j = 0; j < 4; j++）{  
          messageList.add(new KeyedMessage<String, String>("topic2", j+"", "The " + i + " message for key " + j));  
      }  
      producer.send(messageList);  
    }  
producer.close();  
} 

則key相同的消息會(huì)被發(fā)送并存儲(chǔ)到同一個(gè)partition里，而且key的序號(hào)正好和partition序號(hào)相同。（partition序號(hào)從0開始，本例中的key也正好從0開始）。如下圖所示。

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

############################# Log Retention Policy #############################  
 
# The following configurations control the disposal of log segments. The policy can  
# be set to delete segments after a period of time, or after a given size has accumulated.  
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens  
# from the end of the log.  
 
# The minimum age of a log file to be eligible for deletion  
log.retention.hours=168 
 
# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining  
# segments don't drop below log.retention.bytes.  
#log.retention.bytes=1073741824 
 
# 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 
 
# By default the log cleaner is disabled and the log retention policy will default to   
#just delete segments after their retention expires.  
# 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 

這里要注意，因?yàn)镵afka讀取特定消息的時(shí)間復(fù)雜度為O(1)，即與文件大小無關(guān)，所以這里刪除文件與Kafka性能無關(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是無狀態(tài)的，它不需要標(biāo)記哪些消息被哪些consumer過，不需要通過broker去保證同一個(gè)consumer group只有一個(gè)consumer能消費(fèi)某一條消息，因此也就不需要鎖機(jī)制，這也為Kafka的高吞吐率提供了有力保障。

#p#

Replication & Leader election

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。每個(gè)partition都有一個(gè)唯一的leader，所有的讀寫操作都在leader上完成，leader批量從leader上pull數(shù)據(jù)。一般情況下partition的數(shù)量大于等于broker的數(shù)量，并且所有partition的leader均勻分布在broker上。follower上的日志和其leader上的完全一樣。

和大部分分布式系統(tǒng)一樣，Kakfa處理失敗需要明確定義一個(gè)broker是否alive。對(duì)于Kafka而言，Kafka存活包含兩個(gè)條件，一是它必須維護(hù)與Zookeeper的session(這個(gè)通過Zookeeper的heartbeat機(jī)制來實(shí)現(xiàn))。二是follower必須能夠及時(shí)將leader的writing復(fù)制過來，不能“落后太多”。

leader會(huì)track“in sync”的node list。如果一個(gè)follower宕機(jī)，或者落后太多，leader將把它從”in sync” list中移除。這里所描述的“落后太多”指follower復(fù)制的消息落后于leader后的條數(shù)超過預(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    

需要說明的是，Kafka只解決”fail/recover”，不處理“Byzantine”（“拜占庭”）問題。

一條消息只有被“in sync” list里的所有follower都從leader復(fù)制過去才會(huì)被認(rèn)為已提交。這樣就避免了部分?jǐn)?shù)據(jù)被寫進(jìn)了leader，還沒來得及被任何follower復(fù)制就宕機(jī)了，而造成數(shù)據(jù)丟失（consumer無法消費(fèi)這些數(shù)據(jù)）。而對(duì)于producer而言，它可以選擇是否等待消息commit，這可以通過request.required.acks來設(shè)置。這種機(jī)制確保了只要“in sync” list有一個(gè)或以上的flollower，一條被commit的消息就不會(huì)丟失。

這里的復(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的差距（前文有說到，只要follower落后leader不太遠(yuǎn)，則被認(rèn)為在“in sync” list里）。

上文說明了Kafka是如何做replication的，另外一個(gè)很重要的問題是當(dāng)leader宕機(jī)了，怎樣在follower中選舉出新的leader。因?yàn)閒ollower可能落后許多或者crash了，所以必須確保選擇“***”的follower作為新的leader。一個(gè)基本的原則就是，如果leader不在了，新的leader必須擁有原來的leader commit的所有消息。這就需要作一個(gè)折衷，如果leader在標(biāo)明一條消息被commit前等待更多的follower確認(rèn)，那在它die之后就有更多的follower可以作為新的leader，但這也會(huì)造成吞吐率的下降。

一種非常常用的選舉leader的方式是“majority 靈秀”（“少數(shù)服從多數(shù)”），但Kafka并未采用這種方式。這種模式下，如果我們有2f+1個(gè)replica（包含leader和follower），那在commit之前必須保證有f+1個(gè)replica復(fù)制完消息，為了保證正確選出新的leader，fail的replica不能超過f個(gè)。因?yàn)樵谑Ｏ碌娜我鈌+1個(gè)replica里，至少有一個(gè)replica包含有***的所有消息。這種方式有個(gè)很大的優(yōu)勢(shì)，系統(tǒng)的latency只取決于最快的幾臺(tái)server，也就是說，如果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。也就是說，在生產(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的方式。

實(shí)際上，leader election算法非常多，比如Zookeper的Zab, Raft和Viewstamped Replication。而Kafka所使用的leader election算法更像微軟的PacificA算法。

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的一半。

雖然majority vote與ISR相比有不需等待最慢的server這一優(yōu)勢(shì)，但是Kafka作者認(rèn)為Kafka可以通過producer選擇是否被commit阻塞來改善這一問題，并且節(jié)省下來的replica和磁盤使得ISR模式仍然值得。

上文提到，在ISR中至少有一個(gè)follower時(shí)，Kafka可以確保已經(jīng)commit的數(shù)據(jù)不丟失，但如果某一個(gè)partition的所有replica都掛了，就無法保證數(shù)據(jù)不丟失了。這種情況下有兩種可行的方案：

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

這就需要在可用性和一致性當(dāng)中作出一個(gè)簡(jiǎn)單的平衡。如果一定要等待ISR中的replica“活”過來，那不可用的時(shí)間就可能會(huì)相對(duì)較長(zhǎng)。而且如果ISR中的所有replica都無法“活”過來了，或者數(shù)據(jù)都丟失了，這個(gè)partition將永遠(yuǎn)不可用。選擇***個(gè)“活”過來的replica作為leader，而這個(gè)replica不是ISR中的replica，那即使它并不保證已經(jīng)包含了所有已commit的消息，它也會(huì)成為leader而作為consumer的數(shù)據(jù)源（前文有說明，所有讀寫都由leader完成）。Kafka0.8.*使用了第二種方式。根據(jù)Kafka的文檔，在以后的版本中，Kafka支持用戶通過配置選擇這兩種方式中的一種，從而根據(jù)不同的使用場(chǎng)景選擇高可用性還是強(qiáng)一致性。

上文說明了一個(gè)parition的replication過程，然爾Kafka集群需要管理成百上千個(gè)partition，Kafka通過round-robin的方式來平衡partition從而避免大量partition集中在了少數(shù)幾個(gè)節(jié)點(diǎn)上。同時(shí)Kafka也需要平衡leader的分布，盡可能的讓所有partition的leader均勻分布在不同broker上。另一方面，優(yōu)化leadership election的過程也是很重要的，畢竟這段時(shí)間相應(yīng)的partition處于不可用狀態(tài)。一種簡(jiǎn)單的實(shí)現(xiàn)是暫停宕機(jī)的broker上的所有partition，并為之選舉leader。實(shí)際上，Kafka選舉一個(gè)broker作為controller，這個(gè)controller通過watch Zookeeper檢測(cè)所有的broker failure，并負(fù)責(zé)為所有受影響的parition選舉leader，再將相應(yīng)的leader調(diào)整命令發(fā)送至受影響的broker，過程如下圖所示。

這樣做的好處是，可以批量的通知leadership的變化，從而使得選舉過程成本更低，尤其對(duì)大量的partition而言。如果controller失敗了，幸存的所有broker都會(huì)嘗試在Zookeeper中創(chuàng)建/controller->{this broker id}，如果創(chuàng)建成功（只可能有一個(gè)創(chuàng)建成功），則該broker會(huì)成為controller，若創(chuàng)建不成功，則該broker會(huì)等待新controller的命令。

Consumer group

（本節(jié)所有描述都是基于consumer hight level API而非low level API）。

每一個(gè)consumer實(shí)例都屬于一個(gè)consumer group，每一條消息只會(huì)被同一個(gè)consumer group里的一個(gè)consumer實(shí)例消費(fèi)。（不同consumer group可以同時(shí)消費(fèi)同一條消息）

很多傳統(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)一次的語義，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即可。下圖展示了Kafka在Linkedin的一種簡(jiǎn)化部署。

為了更清晰展示Kafka consumer group的特性，筆者作了一項(xiàng)測(cè)試。創(chuàng)建一個(gè)topic (名為topic1)，創(chuàng)建一個(gè)屬于group1的consumer實(shí)例，并創(chuàng)建三個(gè)屬于group2的consumer實(shí)例，然后通過producer向topic1發(fā)送key分別為1，2，3r的消息。結(jié)果發(fā)現(xiàn)屬于group1的consumer收到了所有的這三條消息，同時(shí)group2中的3個(gè)consumer分別收到了key為1，2，3的消息。如下圖所示。

#p#

Consumer Rebalance

（本節(jié)所講述內(nèi)容均基于Kafka consumer high level API）

Kafka保證同一consumer group中只有一個(gè)consumer會(huì)消息某條消息，實(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ì)是無法讓同一個(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)。

如果某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無法消費(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。
 

consumer rebalance算法如下：

• Sort PT (all partitions in topic T)
• Sort CG(all consumers in consumer group G)
• Let i be the index position of Ci in CG and let N=size(PT)/size(CG)
• Remove current entries owned by Ci from the partition owner registry
• Assign partitions from iN to (i+1)N-1 to consumer Ci
• Add newly assigned partitions to the partition owner registry

目前consumer rebalance的控制策略是由每一個(gè)consumer通過Zookeeper完成的。具體的控制方式如下：

• Register itself in the consumer id registry under its group.
• Register a watch on changes under the consumer id registry.
• Register a watch on changes under the broker id registry.
• If the consumer creates a message stream using a topic filter, it also registers a watch on changes under the broker topic registry.
• Force itself to rebalance within in its consumer group.

在這種策略下，每一個(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。

目前（2015-01-19）***版（0.8.2）Kafka采用的是上述方式。但該方式有不利的方面：

• Herd effect
  任何broker或者consumer的增減都會(huì)觸發(fā)所有的consumer的rebalance
• Split Brain
  每個(gè)consumer分別單獨(dú)通過Zookeeper判斷哪些partition down了，那么不同consumer從Zookeeper“看”到的view就可能不一樣，這就會(huì)造成錯(cuò)誤的reblance嘗試。而且有可能所有的consumer都認(rèn)為rebalance已經(jīng)完成了，但實(shí)際上可能并非如此。

根據(jù)Kafka官方文檔，Kafka作者正在考慮在還未發(fā)布的0.9.x版本中使用中心協(xié)調(diào)器(coordinator)。大體思想是選舉出一個(gè)broker作為coordinator，由它watch Zookeeper，從而判斷是否有partition或者consumer的增減，然后生成rebalance命令，并檢查是否這些rebalance在所有相關(guān)的consumer中被執(zhí)行成功，如果不成功則重試，若成功則認(rèn)為此次rebalance成功（這個(gè)過程跟replication controller非常類似，所以我很奇怪為什么當(dāng)初設(shè)計(jì)replication controller時(shí)沒有使用類似方式來解決consumer rebalance的問題）。流程如下：

消息Deliver guarantee

通過上文介紹，想必讀者已經(jīng)明天了producer和consumer是如何工作的，以及Kafka是如何做replication的，接下來要討論的是Kafka如何確保消息在producer和consumer之間傳輸。有這么幾種可能的delivery guarantee：

• At most once 消息可能會(huì)丟，但絕不會(huì)重復(fù)傳輸
• At least one 消息絕不會(huì)丟，但可能會(huì)重復(fù)傳輸
• Exactly once 每條消息肯定會(huì)被傳輸一次且僅傳輸一次，很多時(shí)候這是用戶所想要的。
  Kafka的delivery guarantee semantic非常直接。當(dāng)producer向broker發(fā)送消息時(shí)，一旦這條消息被commit，因數(shù)replication的存在，它就不會(huì)丟。但是如果producer發(fā)送數(shù)據(jù)給broker后，遇到的網(wǎng)絡(luò)問題而造成通信中斷，那producer就無法判斷該條消息是否已經(jīng)commit。這一點(diǎn)有點(diǎn)像向一個(gè)自動(dòng)生成primary key的數(shù)據(jù)庫表中插入數(shù)據(jù)。雖然Kafka無法確定網(wǎng)絡(luò)故障期間發(fā)生了什么，但是producer可以生成一種類似于primary key的東西，發(fā)生故障時(shí)冪等性的retry多次，這樣就做到了Exactly one。截止到目前(Kafka 0.8.2版本，2015-01-25)，這一feature還并未實(shí)現(xiàn)，有希望在Kafka未來的版本中實(shí)現(xiàn)。（所以目前默認(rèn)情況下一條消息從producer和broker是確保了At least once，但可通過設(shè)置producer異步發(fā)送實(shí)現(xiàn)At most once）。
  接下來討論的是消息從broker到consumer的delivery guarantee semantic。（僅針對(duì)Kafka consumer high level API）。consumer在從broker讀取消息后，可以選擇commit，該操作會(huì)在Zookeeper中存下該consumer在該partition下讀取的消息的offset。該consumer下一次再讀該partition時(shí)會(huì)從下一條開始讀取。如未commit，下一次讀取的開始位置會(huì)跟上一次commit之后的開始位置相同。當(dāng)然可以將consumer設(shè)置為autocommit，即consumer一旦讀到數(shù)據(jù)立即自動(dòng)commit。如果只討論這一讀取消息的過程，那Kafka是確保了Exactly once。但實(shí)際上實(shí)際使用中consumer并非讀取完數(shù)據(jù)就結(jié)束了，而是要進(jìn)行進(jìn)一步處理，而數(shù)據(jù)處理與commit的順序在很大程度上決定了消息從broker和consumer的delivery guarantee semantic。
• 讀完消息先commit再處理消息。這種模式下，如果consumer在commit后還沒來得及處理消息就crash了，下次重新開始工作后就無法讀到剛剛已提交而未處理的消息，這就對(duì)應(yīng)于At most once
• 讀完消息先處理再commit。這種模式下，如果處理完了消息在commit之前consumer crash了，下次重新開始工作時(shí)還會(huì)處理剛剛未commit的消息，實(shí)際上該消息已經(jīng)被處理過了。這就對(duì)應(yīng)于At least once。在很多情況使用場(chǎng)景下，消息都有一個(gè)primary key，所以消息的處理往往具有冪等性，即多次處理這一條消息跟只處理一次是等效的，那就可以認(rèn)為是Exactly once。（人個(gè)感覺這種說法有些牽強(qiáng)，畢竟它不是Kafka本身提供的機(jī)制，而且primary key本身不保證操作的冪等性。而且實(shí)際上我們說delivery guarantee semantic是討論被處理多少次，而非處理結(jié)果怎樣，因?yàn)樘幚矸绞蕉喾N多樣，我們的系統(tǒng)不應(yīng)該把處理過程的特性—如是否冪等性，當(dāng)成Kafka本身的feature）
• 如果一定要做到Exactly once，就需要協(xié)調(diào)offset和實(shí)際操作的輸出。精典的做法是引入兩階段提交。如果能讓offset和操作輸入存在同一個(gè)地方，會(huì)更簡(jiǎn)潔和通用。這種方式可能更好，因?yàn)樵S多輸出系統(tǒng)可能不支持兩階段提交。比如，consumer拿到數(shù)據(jù)后可能把數(shù)據(jù)放到HDFS，如果把***的offset和數(shù)據(jù)本身一起寫到HDFS，那就可以保證數(shù)據(jù)的輸出和offset的更新要么都完成，要么都不完成，間接實(shí)現(xiàn)Exactly once。（目前就high level API而言，offset是存于Zookeeper中的，無法存于HDFS，而low level API的offset是由自己去維護(hù)的，可以將之存于HDFS中）
  總之，Kafka默認(rèn)保證At least once，并且允許通過設(shè)置producer異步提交來實(shí)現(xiàn)At most once。而Exactly once要求與目標(biāo)存儲(chǔ)系統(tǒng)協(xié)作，幸運(yùn)的是Kafka提供的offset可以使用這種方式非常直接非常容易。

#p#

Benchmark

紙上得來終覺淺，絕知些事要躬行。筆者希望能親自測(cè)一下Kafka的性能，而非從網(wǎng)上找一些測(cè)試數(shù)據(jù)。所以筆者曾在0.8發(fā)布前兩個(gè)月做過詳細(xì)的Kafka0.8性能測(cè)試，不過很可惜測(cè)試報(bào)告不慎丟失。所幸在網(wǎng)上找到了Kafka的創(chuàng)始人之一的Jay Kreps的bechmark。以下描述皆基于該benchmark。（該benchmark基于Kafka0.8.1）

測(cè)試環(huán)境

該benchmark用到了六臺(tái)機(jī)器，機(jī)器配置如下

• Intel Xeon 2.5 GHz processor with six cores
• Six 7200 RPM SATA drives
• 32GB of RAM
• 1Gb Ethernet
  
  這6臺(tái)機(jī)器其中3臺(tái)用來搭建Kafka broker集群，另外3臺(tái)用來安裝Zookeeper及生成測(cè)試數(shù)據(jù)。6個(gè)drive都直接以非RAID方式掛載。實(shí)際上kafka對(duì)機(jī)器的需求與Hadoop的類似。

producer吞吐率

該項(xiàng)測(cè)試只測(cè)producer的吞吐率，也就是數(shù)據(jù)只被持久化，沒有consumer讀數(shù)據(jù)。

1個(gè)producer線程，無replication

在這一測(cè)試中，創(chuàng)建了一個(gè)包含6個(gè)partition且沒有replication的topic。然后通過一個(gè)線程盡可能快的生成50 million條比較短（payload100字節(jié)長(zhǎng)）的消息。測(cè)試結(jié)果是821,557 records/second（78.3MB/second）。
之所以使用短消息，是因?yàn)閷?duì)于消息系統(tǒng)來說這種使用場(chǎng)景更難。因?yàn)槿绻褂肕B/second來表征吞吐率，那發(fā)送長(zhǎng)消息無疑能使得測(cè)試結(jié)果更好。
整個(gè)測(cè)試中，都是用每秒鐘delivery的消息的數(shù)量乘以payload的長(zhǎng)度來計(jì)算MB/second的，沒有把消息的元信息算在內(nèi)，所以實(shí)際的網(wǎng)絡(luò)使用量會(huì)比這個(gè)大。對(duì)于本測(cè)試來說，每次還需傳輸額外的22個(gè)字節(jié)，包括一個(gè)可選的key，消息長(zhǎng)度描述，CRC等。另外，還包含一些請(qǐng)求相關(guān)的overhead，比如topic，partition，acknowledgement等。這就導(dǎo)致我們比較難判斷是否已經(jīng)達(dá)到網(wǎng)卡極限，但是把這些overhead都算在吞吐率里面應(yīng)該更合理一些。因此，我們已經(jīng)基本達(dá)到了網(wǎng)卡的極限。
初步觀察此結(jié)果會(huì)認(rèn)為它比人們所預(yù)期的要高很多，尤其當(dāng)考慮到Kafka要把數(shù)據(jù)持久化到磁盤當(dāng)中。實(shí)際上，如果使用隨機(jī)訪問數(shù)據(jù)系統(tǒng)，比如RDBMS，或者key-velue store，可預(yù)期的***訪問頻率大概是5000到50000個(gè)請(qǐng)求每秒，這和一個(gè)好的RPC層所能接受的遠(yuǎn)程請(qǐng)求量差不多。而該測(cè)試中遠(yuǎn)超于此的原因有兩個(gè)。

• Kafka確保寫磁盤的過程是線性磁盤I/O，測(cè)試中使用的6塊廉價(jià)磁盤線性I/O的***吞吐量是822MB/second，這已經(jīng)遠(yuǎn)大于1Gb網(wǎng)卡所能帶來的吞吐量了。許多消息系統(tǒng)把數(shù)據(jù)持久化到磁盤當(dāng)成是一個(gè)開銷很大的事情，這是因?yàn)樗麄儗?duì)磁盤的操作都不是線性I/O。
• 在每一個(gè)階段，Kafka都盡量使用批量處理。如果想了解批處理在I/O操作中的重要性，可以參考David Patterson的”Latency Lags Bandwidth“

1個(gè)producer線程，3個(gè)異步replication

該項(xiàng)測(cè)試與上一測(cè)試基本一樣，唯一的區(qū)別是每個(gè)partition有3個(gè)replica（所以網(wǎng)絡(luò)傳輸?shù)暮蛯懭氪疟P的總的數(shù)據(jù)量增加了3倍）。每一個(gè)broker即要寫作為leader的partition，也要讀（從leader讀數(shù)據(jù)）寫（將數(shù)據(jù)寫到磁盤）作為follower的partition。測(cè)試結(jié)果為786,980 records/second（75.1MB/second）。
該項(xiàng)測(cè)試中replication是異步的，也就是說broker收到數(shù)據(jù)并寫入本地磁盤后就acknowledge producer，而不必等所有replica都完成replication。也就是說，如果leader crash了，可能會(huì)丟掉一些***的還未備份的數(shù)據(jù)。但這也會(huì)讓message acknowledgement延遲更少，實(shí)時(shí)性更好。
這項(xiàng)測(cè)試說明，replication可以很快。整個(gè)集群的寫能力可能會(huì)由于3倍的replication而只有原來的三分之一，但是對(duì)于每一個(gè)producer來說吞吐率依然足夠好。

1個(gè)producer線程，3個(gè)同步replication

該項(xiàng)測(cè)試與上一測(cè)試的唯一區(qū)別是replication是同步的，每條消息只有在被in sync集合里的所有replica都復(fù)制過去后才會(huì)被置為committed（此時(shí)broker會(huì)向producer發(fā)送acknowledgement）。在這種模式下，Kafka可以保證即使leader crash了，也不會(huì)有數(shù)據(jù)丟失。測(cè)試結(jié)果為421,823 records/second（40.2MB/second）。
Kafka同步復(fù)制與異步復(fù)制并沒有本質(zhì)的不同。leader會(huì)始終track follower replica從而監(jiān)控它們是否還alive，只有所有in sync集合里的replica都acknowledge的消息才可能被consumer所消費(fèi)。而對(duì)follower的等待影響了吞吐率。可以通過增大batch size來改善這種情況，但為了避免特定的優(yōu)化而影響測(cè)試結(jié)果的可比性，本次測(cè)試并沒有做這種調(diào)整。

3個(gè)producer,3個(gè)異步replication

該測(cè)試相當(dāng)于把上文中的1個(gè)producer,復(fù)制到了3臺(tái)不同的機(jī)器上（在1臺(tái)機(jī)器上跑多個(gè)實(shí)例對(duì)吞吐率的增加不會(huì)有太大幫忙，因?yàn)榫W(wǎng)卡已經(jīng)基本飽和了），這3個(gè)producer同時(shí)發(fā)送數(shù)據(jù)。整個(gè)集群的吞吐率為2,024,032 records/second（193,0MB/second）。

Producer Throughput Vs. Stored Data

消息系統(tǒng)的一個(gè)潛在的危險(xiǎn)是當(dāng)數(shù)據(jù)能都存于內(nèi)存時(shí)性能很好，但當(dāng)數(shù)據(jù)量太大無法完全存于內(nèi)存中時(shí)（然后很多消息系統(tǒng)都會(huì)刪除已經(jīng)被消費(fèi)的數(shù)據(jù)，但當(dāng)消費(fèi)速度比生產(chǎn)速度慢時(shí)，仍會(huì)造成數(shù)據(jù)的堆積），數(shù)據(jù)會(huì)被轉(zhuǎn)移到磁盤，從而使得吞吐率下降，這又反過來造成系統(tǒng)無法及時(shí)接收數(shù)據(jù)。這樣就非常糟糕，而實(shí)際上很多情景下使用queue的目的就是解決數(shù)據(jù)消費(fèi)速度和生產(chǎn)速度不一致的問題。

但Kafka不存在這一問題，因?yàn)镵afka始終以O(shè)（1）的時(shí)間復(fù)雜度將數(shù)據(jù)持久化到磁盤，所以其吞吐率不受磁盤上所存儲(chǔ)的數(shù)據(jù)量的影響。為了驗(yàn)證這一特性，做了一個(gè)長(zhǎng)時(shí)間的大數(shù)據(jù)量的測(cè)試，下圖是吞吐率與數(shù)據(jù)量大小的關(guān)系圖。

上圖中有一些variance的存在，并可以明顯看到，吞吐率并不受磁盤上所存數(shù)據(jù)量大小的影響。實(shí)際上從上圖可以看到，當(dāng)磁盤數(shù)據(jù)量達(dá)到1TB時(shí)，吞吐率和磁盤數(shù)據(jù)只有幾百M(fèi)B時(shí)沒有明顯區(qū)別。

這個(gè)variance是由Linux I/O管理造成的，它會(huì)把數(shù)據(jù)緩存起來再批量flush。上圖的測(cè)試結(jié)果是在生產(chǎn)環(huán)境中對(duì)Kafka集群做了些tuning后得到的，這些tuning方法可參考這里。

consumer吞吐率

需要注意的是，replication factor并不會(huì)影響consumer的吞吐率測(cè)試，因?yàn)閏onsumer只會(huì)從每個(gè)partition的leader讀數(shù)據(jù)，而與replicaiton factor無關(guān)。同樣，consumer吞吐率也與同步復(fù)制還是異步復(fù)制無關(guān)。

1個(gè)consumer

該測(cè)試從有6個(gè)partition，3個(gè)replication的topic消費(fèi)50 million的消息。測(cè)試結(jié)果為940,521 records/second（89.7MB/second）。
可以看到，Kafkar的consumer是非常高效的。它直接從broker的文件系統(tǒng)里讀取文件塊。Kafka使用sendfile API來直接通過操作系統(tǒng)直接傳輸，而不用把數(shù)據(jù)拷貝到用戶空間。該項(xiàng)測(cè)試實(shí)際上從log的起始處開始讀數(shù)據(jù)，所以它做了真實(shí)的I/O。在生產(chǎn)環(huán)境下，consumer可以直接讀取producer剛剛寫下的數(shù)據(jù)（它可能還在緩存中）。實(shí)際上，如果在生產(chǎn)環(huán)境下跑I/O stat，你可以看到基本上沒有物理“讀”。也就是說生產(chǎn)環(huán)境下consumer的吞吐率會(huì)比該項(xiàng)測(cè)試中的要高。

3個(gè)consumer

將上面的consumer復(fù)制到3臺(tái)不同的機(jī)器上，并且并行運(yùn)行它們（從同一個(gè)topic上消費(fèi)數(shù)據(jù)）。測(cè)試結(jié)果為2,615,968 records/second（249.5MB/second）。
正如所預(yù)期的那樣，consumer的吞吐率幾乎線性增漲。

Producer and Consumer

上面的測(cè)試只是把producer和consumer分開測(cè)試，而該項(xiàng)測(cè)試同時(shí)運(yùn)行producer和consumer，這更接近使用場(chǎng)景。實(shí)際上目前的replication系統(tǒng)中follower就相當(dāng)于consumer在工作。
該項(xiàng)測(cè)試，在具有6個(gè)partition和3個(gè)replica的topic上同時(shí)使用1個(gè)producer和1個(gè)consumer，并且使用異步復(fù)制。測(cè)試結(jié)果為795,064 records/second（75.8MB/second）。
可以看到，該項(xiàng)測(cè)試結(jié)果與單獨(dú)測(cè)試1個(gè)producer時(shí)的結(jié)果幾乎一致。所以說consumer非常輕量級(jí)。

消息長(zhǎng)度對(duì)吞吐率的影響

上面的所有測(cè)試都基于短消息（payload 100字節(jié)），而正如上文所說，短消息對(duì)Kafka來說是更難處理的使用方式，可以預(yù)期，隨著消息長(zhǎng)度的增大，records/second會(huì)減小，但MB/second會(huì)有所提高。下圖是records/second與消息長(zhǎng)度的關(guān)系圖。

正如我們所預(yù)期的那樣，隨著消息長(zhǎng)度的增加，每秒鐘所能發(fā)送的消息的數(shù)量逐漸減小。但是如果看每秒鐘發(fā)送的消息的總大小，它會(huì)隨著消息長(zhǎng)度的增加而增加，如下圖所示。

從上圖可以看出，當(dāng)消息長(zhǎng)度為10字節(jié)時(shí)，因?yàn)橐l繁入隊(duì)，花了太多時(shí)間獲取鎖，CPU成了瓶頸，并不能充分利用帶寬。但從100字節(jié)開始，我們可以看到帶寬的使用逐漸趨于飽和（雖然MB/second還是會(huì)隨著消息長(zhǎng)度的增加而增加，但增加的幅度也越來越小）。

端到端的Latency

上文中討論了吞吐率，那消息傳輸?shù)膌atency如何呢？也就是說消息從producer到consumer需要多少時(shí)間呢？該項(xiàng)測(cè)試創(chuàng)建1個(gè)producer和1個(gè)consumer并反復(fù)計(jì)時(shí)。結(jié)果是，2 ms (median), 3ms (99th percentile, 14ms (99.9th percentile)。
（這里并沒有說明topic有多少個(gè)partition，也沒有說明有多少個(gè)replica，replication是同步還是異步。實(shí)際上這會(huì)極大影響producer發(fā)送的消息被commit的latency，而只有committed的消息才能被consumer所消費(fèi)，所以它會(huì)最終影響端到端的latency）

#p#

重現(xiàn)該benchmark

如果讀者想要在自己的機(jī)器上重現(xiàn)本次benchmark測(cè)試，可以參考本次測(cè)試的配置和所使用的命令。
實(shí)際上Kafka Distribution提供了producer性能測(cè)試工具，可通過bin/kafka-producer-perf-test.sh腳本來啟動(dòng)。所使用的命令如下

Producer  
Setup  
bin/kafka-topics.sh --zookeeper esv4-hcl197.grid.linkedin.com:2181 --create --topic test-rep-one --partitions 6 --replication-factor 1  
bin/kafka-topics.sh --zookeeper esv4-hcl197.grid.linkedin.com:2181 --create --topic test --partitions 6 --replication-factor 3  
 
Single thread, no replication  
 
bin/kafka-run-class.sh org.apache.kafka.clients.tools.ProducerPerformance test7 50000000 100 -1 acks=1 bootstrap.servers=esv4-hcl198.grid.linkedin.com:9092 buffer.memory=67108864 batch.size=8196 
 
Single-thread, async 3x replication  
 
bin/kafktopics.sh --zookeeper esv4-hcl197.grid.linkedin.com:2181 --create --topic test --partitions 6 --replication-factor 3  
bin/kafka-run-class.sh org.apache.kafka.clients.tools.ProducerPerformance test6 50000000 100 -1 acks=1 bootstrap.servers=esv4-hcl198.grid.linkedin.com:9092 buffer.memory=67108864 batch.size=8196 
 
Single-thread, sync 3x replication  
 
bin/kafka-run-class.sh org.apache.kafka.clients.tools.ProducerPerformance test 50000000 100 -1 acks=-1 bootstrap.servers=esv4-hcl198.grid.linkedin.com:9092 buffer.memory=67108864 batch.size=64000 
 
Three Producers, 3x async replication  
bin/kafka-run-class.sh org.apache.kafka.clients.tools.ProducerPerformance test 50000000 100 -1 acks=1 bootstrap.servers=esv4-hcl198.grid.linkedin.com:9092 buffer.memory=67108864 batch.size=8196 
 
Throughput Versus Stored Data  
 
bin/kafka-run-class.sh org.apache.kafka.clients.tools.ProducerPerformance test 50000000000 100 -1 acks=1 bootstrap.servers=esv4-hcl198.grid.linkedin.com:9092 buffer.memory=67108864 batch.size=8196 
 
Effect of message size  
 
for i in 10 100 1000 10000 100000;  
do  
echo ""  
echo $i  
bin/kafka-run-class.sh org.apache.kafka.clients.tools.ProducerPerformance test $((1000*1024*1024/$i)) $i -1 acks=1 bootstrap.servers=esv4-hcl198.grid.linkedin.com:9092 buffer.memory=67108864 batch.size=128000 
done;  
 
Consumer  
Consumer throughput  
 
bin/kafka-consumer-perf-test.sh --zookeeper esv4-hcl197.grid.linkedin.com:2181 --messages 50000000 --topic test --threads 1  
 
3 Consumers  
 
On three servers, run:  
bin/kafka-consumer-perf-test.sh --zookeeper esv4-hcl197.grid.linkedin.com:2181 --messages 50000000 --topic test --threads 1  
 
End-to-end Latency  
 
bin/kafka-run-class.sh kafka.tools.TestEndToEndLatency esv4-hcl198.grid.linkedin.com:9092 esv4-hcl197.grid.linkedin.com:2181 test 5000  
 
Producer and consumer  
 
bin/kafka-run-class.sh org.apache.kafka.clients.tools.ProducerPerformance test 50000000 100 -1 acks=1 bootstrap.servers=esv4-hcl198.grid.linkedin.com:9092 buffer.memory=67108864 batch.size=8196 
 
bin/kafka-consumer-perf-test.sh --zookeeper esv4-hcl197.grid.linkedin.com:2181 --messages 50000000 --topic test --threads 1  

broker配置如下

############################# Server Basics #############################  
 
# The id of the broker. This must be set to a unique integer for each broker.  
broker.id=0 
 
############################# Socket Server Settings #############################  
 
# The port the socket server listens on  
port=9092 
 
# Hostname the broker will bind to and advertise to producers and consumers.  
# If not set, the server will bind to all interfaces and advertise the value returned from  
# from java.net.InetAddress.getCanonicalHostName().  
#host.name=localhost 
 
# The number of threads handling network requests  
num.network.threads=4 
 
# The number of threads doing disk I/O  
num.io.threads=8 
 
# The send buffer (SO_SNDBUF) used by the socket server  
socket.send.buffer.bytes=1048576 
 
# The receive buffer (SO_RCVBUF) used by the socket server  
socket.receive.buffer.bytes=1048576 
 
# The maximum size of a request that the socket server will accept (protection against OOM)  
socket.request.max.bytes=104857600 
 
 
############################# Log Basics #############################  
 
# The directory under which to store log files  
log.dirs=/grid/a/dfs-data/kafka-logs,/grid/b/dfs-data/kafka-logs,/grid/c/dfs-data/kafka-logs,/grid/d/dfs-data/kafka-logs,/grid/e/dfs-data/kafka-logs,/grid/f/dfs-data/kafka-logs  
 
# The number of logical partitions per topic per server. More partitions allow greater parallelism  
# for consumption, but also mean more files.  
num.partitions=8 
 
############################# Log Flush Policy #############################  
 
# The following configurations control the flush of data to disk. This is the most  
# important performance knob in kafka.  
# There are a few important trade-offs here:  
#    1. Durability: Unflushed data is at greater risk of loss in the event of a crash.  
#    2. Latency: Data is not made available to consumers until it is flushed (which adds latency).  
#    3. Throughput: The flush is generally the most expensive operation.   
# The settings below allow one to configure the flush policy to flush data after a period of time or  
# every N messages (or both). This can be done globally and overridden on a per-topic basis.  
 
# Per-topic overrides for log.flush.interval.ms  
#log.flush.intervals.ms.per.topic=topic1:1000, topic2:3000  
 
############################# Log Retention Policy #############################  
 
# The following configurations control the disposal of log segments. The policy can  
# be set to delete segments after a period of time, or after a given size has accumulated.  
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens  
# from the end of the log.  
 
# The minimum age of a log file to be eligible for deletion  
log.retention.hours=168 
 
# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining  
# segments don't drop below log.retention.bytes.  
#log.retention.bytes=1073741824 
 
# The maximum size of a log segment file. When this size is reached a new log segment will be created.  
log.segment.bytes=536870912 
 
# The interval at which log segments are checked to see if they can be deleted according   
# to the retention policies  
log.cleanup.interval.mins=1 
 
############################# Zookeeper #############################  
 
# Zookeeper connection string (see zookeeper docs for details).  
# This is a comma separated host:port pairs, each corresponding to a zk  
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".  
# You can also append an optional chroot string to the urls to specify the  
# root directory for all kafka znodes.  
zookeeper.connect=esv4-hcl197.grid.linkedin.com:2181  
 
# Timeout in ms for connecting to zookeeper  
zookeeper.connection.timeout.ms=1000000 
 
# metrics reporter properties  
kafka.metrics.polling.interval.secs=5 
kafkakafka.metrics.reporters=kafka.metrics.KafkaCSVMetricsReporter  
kafka.csv.metrics.dir=/tmp/kafka_metrics  
# Disable csv reporting by default.  
kafka.csv.metrics.reporter.enabled=false 
 
replica.lag.max.messages=10000000 

讀者也可參考另外一份Kafka性能測(cè)試報(bào)告

參考

• 使用消息隊(duì)列的 10 個(gè)理由
• Apache Kafka
• Efficient data transfer through zero copy
• Benchmarking Apache Kafka: 2 Million Writes Per Second (On Three Cheap Machines)
• Kafka 0.8 Producer Performance