本文轉(zhuǎn)載自http://dataunion.org/?p=9307

背景介紹
Kafka簡介
Kafka是一種分布式的，基于發(fā)布/訂閱的消息系統(tǒng)您访。主要設(shè)計目標(biāo)如下：
以時間復(fù)雜度為O(1)的方式提供消息持久化能力，并保證即使對TB級以上數(shù)據(jù)也能保證常數(shù)時間的訪問性能
高吞吐率施禾。即使在非常廉價的商用機(jī)器上也能做到單機(jī)支持每秒100K條消息的傳輸
支持Kafka Server間的消息分區(qū)足画，及分布式消息消費(fèi)脐区，同時保證每個partition內(nèi)的消息順序傳輸
同時支持離線數(shù)據(jù)處理和實(shí)時數(shù)據(jù)處理

為什么要用Message Queue
解耦 在項目啟動之初來預(yù)測將來項目會碰到什么需求，是極其困難的登澜。消息隊列在處理過程中間插入了一個隱含的阔挠、基于數(shù)據(jù)的接口層，兩邊的處理過程都要實(shí)現(xiàn)這一接口脑蠕。這允許你獨(dú)立的擴(kuò)展或修改兩邊的處理過程购撼，只要確保它們遵守同樣的接口約束
冗余 有時在處理數(shù)據(jù)的時候處理過程會失敗。除非數(shù)據(jù)被持久化谴仙，否則將永遠(yuǎn)丟失迂求。消息隊列把數(shù)據(jù)進(jìn)行持久化直到它們已經(jīng)被完全處理，通過這一方式規(guī)避了數(shù)據(jù)丟失風(fēng)險晃跺。在被許多消息隊列所采用的”插入-獲取-刪除”范式中揩局，在把一個消息從隊列中刪除之前，需要你的處理過程明確的指出該消息已經(jīng)被處理完畢哼审，確保你的數(shù)據(jù)被安全的保存直到你使用完畢谐腰。
擴(kuò)展性 因?yàn)橄㈥犃薪怦盍四愕奶幚磉^程孕豹，所以增大消息入隊和處理的頻率是很容易的；只要另外增加處理過程即可十气。不需要改變代碼励背、不需要調(diào)節(jié)參數(shù)。擴(kuò)展就像調(diào)大電力按鈕一樣簡單砸西。
靈活性 & 峰值處理能力 在訪問量劇增的情況下叶眉，應(yīng)用仍然需要繼續(xù)發(fā)揮作用，但是這樣的突發(fā)流量并不常見芹枷；如果為以能處理這類峰值訪問為標(biāo)準(zhǔn)來投入資源隨時待命無疑是巨大的浪費(fèi)衅疙。使用消息隊列能夠使關(guān)鍵組件頂住增長的訪問壓力，而不是因?yàn)槌鲐?fù)荷的請求而完全崩潰鸳慈。
可恢復(fù)性 當(dāng)體系的一部分組件失效饱溢，不會影響到整個系統(tǒng)。消息隊列降低了進(jìn)程間的耦合度走芋，所以即使一個處理消息的進(jìn)程掛掉绩郎，加入隊列中的消息仍然可以在系統(tǒng)恢復(fù)后被處理。而這種允許重試或者延后處理請求的能力通常是造就一個略感不便的用戶和一個沮喪透頂?shù)挠脩糁g的區(qū)別翁逞。
送達(dá)保證 消息隊列提供的冗余機(jī)制保證了消息能被實(shí)際的處理肋杖，只要一個進(jìn)程讀取了該隊列即可。在此基礎(chǔ)上挖函，IronMQ提供了一個”只送達(dá)一次”保證状植。無論有多少進(jìn)程在從隊列中領(lǐng)取數(shù)據(jù)，每一個消息只能被處理一次怨喘。這之所以成為可能津畸，是因?yàn)楂@取一個消息只是”預(yù)定”了這個消息，暫時把它移出了隊列哲思。除非客戶端明確的表示已經(jīng)處理完了這個消息洼畅，否則這個消息會被放回隊列中去，在一段可配置的時間之后可再次被處理棚赔。

順序保證 在許多情況下帝簇，數(shù)據(jù)處理的順序都很重要。消息隊列本來就是排序的靠益，并且能保證數(shù)據(jù)會按照特定的順序來處理丧肴。IronMO保證消息漿糊通過FIFO（先進(jìn)先出）的順序來處理，因此消息在隊列中的位置就是從隊列中檢索他們的位置胧后。
緩沖 在任何重要的系統(tǒng)中芋浮，都會有需要不同的處理時間的元素。例如,加載一張圖片比應(yīng)用過濾器花費(fèi)更少的時間壳快。消息隊列通過一個緩沖層來幫助任務(wù)最高效率的執(zhí)行—寫入隊列的處理會盡可能的快速纸巷，而不受從隊列讀的預(yù)備處理的約束镇草。該緩沖有助于控制和優(yōu)化數(shù)據(jù)流經(jīng)過系統(tǒng)的速度。
理解數(shù)據(jù)流 在一個分布式系統(tǒng)里瘤旨，要得到一個關(guān)于用戶操作會用多長時間及其原因的總體印象梯啤，是個巨大的挑戰(zhàn)。消息系列通過消息被處理的頻率存哲，來方便的輔助確定那些表現(xiàn)不佳的處理過程或領(lǐng)域因宇，這些地方的數(shù)據(jù)流都不夠優(yōu)化。
異步通信 很多時候祟偷，你不想也不需要立即處理消息察滑。消息隊列提供了異步處理機(jī)制，允許你把一個消息放入隊列修肠，但并不立即處理它贺辰。你想向隊列中放入多少消息就放多少，然后在你樂意的時候再去處理它們嵌施。

常用Message Queue對比
RabbitMQ RabbitMQ是使用Erlang編寫的一個開源的消息隊列魂爪，本身支持很多的協(xié)議：AMQP，XMPP, SMTP, STOMP艰管，也正因如此，它非常重量級蒋川，更適合于企業(yè)級的開發(fā)牲芋。同時實(shí)現(xiàn)了Broker構(gòu)架，這意味著消息在發(fā)送給客戶端時先在中心隊列排隊捺球。對路由缸浦，負(fù)載均衡或者數(shù)據(jù)持久化都有很好的支持。
Redis Redis是一個基于Key-Value對的NoSQL數(shù)據(jù)庫氮兵，開發(fā)維護(hù)很活躍裂逐。雖然它是一個Key-Value數(shù)據(jù)庫存儲系統(tǒng)，但它本身支持MQ功能泣栈，所以完全可以當(dāng)做一個輕量級的隊列服務(wù)來使用卜高。對于RabbitMQ和Redis的入隊和出隊操作，各執(zhí)行100萬次南片，每10萬次記錄一次執(zhí)行時間掺涛。測試數(shù)據(jù)分為128Bytes、512Bytes疼进、1K和10K四個不同大小的數(shù)據(jù)薪缆。實(shí)驗(yàn)表明：入隊時，當(dāng)數(shù)據(jù)比較小時Redis的性能要高于RabbitMQ伞广，而如果數(shù)據(jù)大小超過了10K拣帽，Redis則慢的無法忍受疼电；出隊時，無論數(shù)據(jù)大小减拭，Redis都表現(xiàn)出非常好的性能蔽豺，而RabbitMQ的出隊性能則遠(yuǎn)低于Redis。
ZeroMQ ZeroMQ號稱最快的消息隊列系統(tǒng)峡谊，尤其針對大吞吐量的需求場景茫虽。ZMQ能夠?qū)崿F(xiàn)RabbitMQ不擅長的高級/復(fù)雜的隊列，但是開發(fā)人員需要自己組合多種技術(shù)框架既们，技術(shù)上的復(fù)雜度是對這MQ能夠應(yīng)用成功的挑戰(zhàn)濒析。ZeroMQ具有一個獨(dú)特的非中間件的模式，你不需要安裝和運(yùn)行一個消息服務(wù)器或中間件啥纸，因?yàn)槟愕膽?yīng)用程序?qū)缪萘诉@個服務(wù)角色号杏。你只需要簡單的引用ZeroMQ程序庫，可以使用NuGet安裝斯棒，然后你就可以愉快的在應(yīng)用程序之間發(fā)送消息了盾致。但是ZeroMQ僅提供非持久性的隊列，也就是說如果down機(jī)荣暮，數(shù)據(jù)將會丟失庭惜。其中，Twitter的Storm中默認(rèn)使用ZeroMQ作為數(shù)據(jù)流的傳輸穗酥。
ActiveMQ ActiveMQ是Apache下的一個子項目护赊。 類似于ZeroMQ，它能夠以代理人和點(diǎn)對點(diǎn)的技術(shù)實(shí)現(xiàn)隊列砾跃。同時類似于RabbitMQ骏啰，它少量代碼就可以高效地實(shí)現(xiàn)高級應(yīng)用場景。
Kafka/Jafka Kafka是Apache下的一個子項目抽高，是一個高性能跨語言分布式Publish/Subscribe消息隊列系統(tǒng)判耕，而Jafka是在Kafka之上孵化而來的，即Kafka的一個升級版翘骂。具有以下特性：快速持久化壁熄，可以在O(1)的系統(tǒng)開銷下進(jìn)行消息持久化；高吞吐雏胃，在一臺普通的服務(wù)器上既可以達(dá)到10W/s的吞吐速率请毛；完全的分布式系統(tǒng)，Broker瞭亮、Producer方仿、Consumer都原生自動支持分布式，自動實(shí)現(xiàn)復(fù)雜均衡；支持Hadoop數(shù)據(jù)并行加載仙蚜，對于像Hadoop的一樣的日志數(shù)據(jù)和離線分析系統(tǒng)此洲，但又要求實(shí)時處理的限制，這是一個可行的解決方案委粉。Kafka通過Hadoop的并行加載機(jī)制來統(tǒng)一了在線和離線的消息處理呜师，這一點(diǎn)也是本課題所研究系統(tǒng)所看重的。Apache Kafka相對于ActiveMQ是一個非常輕量級的消息系統(tǒng)，除了性能非常好之外，還是一個工作良好的分布式系統(tǒng)有梆。

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

Kafka架構(gòu)

Topic & Partition
Topic在邏輯上可以被認(rèn)為是一個在的queue，每條消費(fèi)都必須指定它的topic较木，可以簡單理解為必須指明把這條消息放進(jìn)哪個queue里红符。為了使得Kafka的吞吐率可以水平擴(kuò)展，物理上把topic分成一個或多個partition伐债，每個partition在物理上對應(yīng)一個文件夾预侯，該文件夾下存儲這個partition的所有消息和索引文件。

每個日志文件都是“l(fā)og entries”序列峰锁，每一個 log entry
包含一個4字節(jié)整型數(shù)（值為N）萎馅，其后跟N個字節(jié)的消息體。每條消息都有一個當(dāng)前partition下唯一的64字節(jié)的offset虹蒋，它指明了這條消息的起始位置糜芳。磁盤上存儲的消費(fèi)格式如下：
message length ： 4 bytes (value: 1+4+n)
“magic” value ： 1 byte
crc ： 4 bytes
payload ： n bytes
這個“l(fā)og entries”并非由一個文件構(gòu)成，而是分成多個segment魄衅，每個segment名為該segment第一條消息的offset和“.kafka”組成峭竣。另外會有一個索引文件，它標(biāo)明了每個segment下包含的 log entry
的offset范圍晃虫，如下圖所示皆撩。

中指定這個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ā)送一條消息時由境，可以指定這條消息的key，producer根據(jù)這個key和partition機(jī)制來判斷將這條消息發(fā)送到哪個parition蓖议。paritition機(jī)制可以通過指定producer的paritition. class這一參數(shù)來指定虏杰，該class必須實(shí)現(xiàn) kafka.producer.Partitioner
接口。本例中如果key可以被解析為整數(shù)則將對應(yīng)的整數(shù)與partition總數(shù)取余勒虾，該消息會被發(fā)送到該數(shù)對應(yīng)的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); } }}

如果將上例中的class作為partition.class，并通過如下代碼發(fā)送20條消息（key分別為0修然，1笛钝，2，3）至topic2（包含4個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相同的消息會被發(fā)送并存儲到同一個partition里玻靡，而且key的序號正好和partition序號相同。（partition序號從0開始中贝，本例中的key也正好從0開始）囤捻。如下圖所示。

對于傳統(tǒng)的message queue而言邻寿，一般會刪除已經(jīng)被消費(fèi)的消息蝎土，而Kafka集群會保留所有的消息，無論其被消費(fèi)與否绣否。當(dāng)然誊涯，因?yàn)榇疟P限制，不可能永久保留所有數(shù)據(jù)（實(shí)際上也沒必要）蒜撮，因此Kafka提供兩種策略去刪除舊數(shù)據(jù)醋拧。一是基于時間，二是基于partition文件大小淀弹。例如可以通過配置 $KAFKA_HOME/config/server.properties
，讓Kafka刪除一周前的數(shù)據(jù)庆械，也可通過配置讓Kafka在partition文件超過1GB時刪除舊數(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 deletionlog.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 policieslog.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讀取特定消息的時間復(fù)雜度為O(1)沐序，即與文件大小無關(guān)琉用，所以這里刪除文件與Kafka性能無關(guān)，選擇怎樣的刪除策略只與磁盤以及具體的需求有關(guān)策幼。另外邑时，Kafka會為每一個consumer group保留一些metadata信息—當(dāng)前消費(fèi)的消息的position，也即offset特姐。這個offset由consumer控制晶丘。正常情況下consumer會在消費(fèi)完一條消息后線性增加這個offset。當(dāng)然唐含，consumer也可將offset設(shè)成一個較小的值浅浮，重新消費(fèi)一些消息。因?yàn)閛ffet由consumer控制捷枯，所以Kafka broker是無狀態(tài)的滚秩，它不需要標(biāo)記哪些消息被哪些consumer過，不需要通過broker去保證同一個consumer group只有一個consumer能消費(fèi)某一條消息淮捆，因此也就不需要鎖機(jī)制郁油，這也為Kafka的高吞吐率提供了有力保障。
Replication & Leader election
Kafka從0.8開始提供partition級別的replication攀痊，replication的數(shù)量可在 $KAFKA_HOME/config/server.properties
中配置桐腌。
default.replication.factor = 1

該 Replication與leader election配合提供了自動的failover機(jī)制。replication對Kafka的吞吐率是有一定影響的蚕苇，但極大的增強(qiáng)了可用性哩掺。默認(rèn)情況下，Kafka的replication數(shù)量為1涩笤。 每個partition都有一個唯一的leader嚼吞，所有的讀寫操作都在leader上完成，leader批量從leader上pull數(shù)據(jù)蹬碧。一般情況下partition的數(shù)量大于等于broker的數(shù)量舱禽，并且所有partition的leader均勻分布在broker上。follower上的日志和其leader上的完全一樣恩沽。
和大部分分布式系統(tǒng)一樣誊稚，Kakfa處理失敗需要明確定義一個broker是否alive。對于Kafka而言罗心，Kafka存活包含兩個條件里伯，一是它必須維護(hù)與Zookeeper的session(這個通過Zookeeper的heartbeat機(jī)制來實(shí)現(xiàn))。二是follower必須能夠及時將leader的writing復(fù)制過來渤闷，不能“落后太多”疾瓮。
leader會track“in sync”的node list。如果一個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 deadreplica.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 deadreplica.lag.time.max.ms=10000

需要說明的是肩碟，Kafka只解決”fail/recover”强窖，不處理“Byzantine”（“拜占庭”）問題。
一條消息只有被“in sync” list里的所有follower都從leader復(fù)制過去才會被認(rèn)為已提交削祈。這樣就避免了部分?jǐn)?shù)據(jù)被寫進(jìn)了leader翅溺，還沒來得及被任何follower復(fù)制就宕機(jī)了，而造成數(shù)據(jù)丟失（consumer無法消費(fèi)這些數(shù)據(jù)）岩瘦。而對于producer而言未巫，它可以選擇是否等待消息commit，這可以通過 request.required.acks
來設(shè)置启昧。這種機(jī)制確保了只要“in sync” list有一個或以上的flollower叙凡，一條被commit的消息就不會丟失。
這里的復(fù)制機(jī)制即不是同步復(fù)制密末，也不是單純的異步復(fù)制握爷。事實(shí)上，同步復(fù)制要求“活著的”follower都復(fù)制完严里，這條消息才會被認(rèn)為commit新啼，這種復(fù)制方式極大的影響了吞吐率（高吞吐率是Kafka非常重要的一個特性）。而異步復(fù)制方式下刹碾，follower異步的從leader復(fù)制數(shù)據(jù)燥撞，數(shù)據(jù)只要被leader寫入log就被認(rèn)為已經(jīng)commit，這種情況下如果follwer都落后于leader迷帜，而leader突然宕機(jī)物舒，則會丟失數(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的奈搜，另外一個很重要的問題是當(dāng)leader宕機(jī)了悉盆，怎樣在follower中選舉出新的leader。因?yàn)閒ollower可能落后許多或者crash了馋吗，所以必須確保選擇“最新”的follower作為新的leader焕盟。一個基本的原則就是，如果leader不在了耗美，新的leader必須擁有原來的leader commit的所有消息京髓。這就需要作一個折衷，如果leader在標(biāo)明一條消息被commit前等待更多的follower確認(rèn)商架，那在它die之后就有更多的follower可以作為新的leader堰怨，但這也會造成吞吐率的下降。
一種非常常用的選舉leader的方式是“majority 靈秀”（“少數(shù)服從多數(shù)”）蛇摸，但Kafka并未采用這種方式备图。這種模式下，如果我們有2f+1個replica（包含leader和follower）赶袄，那在commit之前必須保證有f+1個replica復(fù)制完消息揽涮，為了保證正確選出新的leader，fail的replica不能超過f個饿肺。因?yàn)樵谑Ｏ碌娜我鈌+1個replica里蒋困，至少有一個replica包含有最新的所有消息。這種方式有個很大的優(yōu)勢敬辣，系統(tǒng)的latency只取決于最快的幾臺server雪标，也就是說，如果replication factor是3溉跃，那latency就取決于最快的那個follower而非最慢那個村刨。majority vote也有一些劣勢，為了保證leader election的正常進(jìn)行撰茎，它所能容忍的fail的follower個數(shù)比較少嵌牺。如果要容忍1個follower掛掉，必須要有3個以上的replica龄糊，如果要容忍2個follower掛掉逆粹，必須要有5個以上的replica。也就是說绎签，在生產(chǎn)環(huán)境下為了保證較高的容錯程度枯饿，必須要有大量的replica，而大量的replica又會在大數(shù)據(jù)量下導(dǎo)致性能的急劇下降诡必。這就是這種算法更多用在 Zookeeper 這種共享集群配置的系統(tǒng)中而很少在需要存儲大量數(shù)據(jù)的系統(tǒng)中使用的原因奢方。例如HDFS的HA feature是基于 majority-vote-based journal ，但是它的數(shù)據(jù)存儲并沒有使用這種expensive的方式爸舒。
實(shí)際上蟋字，leader election算法非常多，比如Zookeper的 Zab , Raft 和 Viewstamped Replication扭勉。而Kafka所使用的leader election算法更像微軟的 PacificA 算法鹊奖。
Kafka在Zookeeper中動態(tài)維護(hù)了一個ISR（in-sync replicas） set，這個set里的所有replica都跟上了leader涂炎，只有ISR里的成員才有被選為leader的可能忠聚。在這種模式下设哗，對于f+1個replica，一個Kafka topic能在保證不丟失已經(jīng)ommit的消息的前提下容忍f個replica的失敗两蟀。在大多數(shù)使用場景中网梢，這種模式是非常有利的。事實(shí)上赂毯，為了容忍f個replica的失敗战虏，majority vote和ISR在commit前需要等待的replica數(shù)量是一樣的，但是ISR需要的總的replica的個數(shù)幾乎是majority vote的一半党涕。
雖然majority vote與ISR相比有不需等待最慢的server這一優(yōu)勢烦感，但是Kafka作者認(rèn)為Kafka可以通過producer選擇是否被commit阻塞來改善這一問題，并且節(jié)省下來的replica和磁盤使得ISR模式仍然值得膛堤。

上文提到手趣，在ISR中至少有一個follower時，Kafka可以確保已經(jīng)commit的數(shù)據(jù)不丟失骑祟，但如果某一個partition的所有replica都掛了回懦，就無法保證數(shù)據(jù)不丟失了。這種情況下有兩種可行的方案：
等待ISR中的任一個replica“活”過來次企，并且選它作為leader
選擇第一個“活”過來的replica（不一定是ISR中的）作為leader

這就需要在可用性和一致性當(dāng)中作出一個簡單的平衡怯晕。如果一定要等待ISR中的replica“活”過來，那不可用的時間就可能會相對較長缸棵。而且如果ISR中的所有replica都無法“活”過來了舟茶，或者數(shù)據(jù)都丟失了，這個partition將永遠(yuǎn)不可用堵第。選擇第一個“活”過來的replica作為leader吧凉，而這個replica不是ISR中的replica，那即使它并不保證已經(jīng)包含了所有已commit的消息踏志，它也會成為leader而作為consumer的數(shù)據(jù)源（前文有說明阀捅，所有讀寫都由leader完成）。Kafka0.8.*使用了第二種方式针余。根據(jù)Kafka的文檔饲鄙，在以后的版本中，Kafka支持用戶通過配置選擇這兩種方式中的一種圆雁，從而根據(jù)不同的使用場景選擇高可用性還是強(qiáng)一致性忍级。
上文說明了一個parition的replication過程，然爾Kafka集群需要管理成百上千個partition伪朽，Kafka通過round-robin的方式來平衡partition從而避免大量partition集中在了少數(shù)幾個節(jié)點(diǎn)上轴咱。同時Kafka也需要平衡leader的分布，盡可能的讓所有partition的leader均勻分布在不同broker上。另一方面朴肺，優(yōu)化leadership election的過程也是很重要的窖剑，畢竟這段時間相應(yīng)的partition處于不可用狀態(tài)。一種簡單的實(shí)現(xiàn)是暫停宕機(jī)的broker上的所有partition戈稿，并為之選舉leader苛吱。實(shí)際上，Kafka選舉一個broker作為controller器瘪，這個controller通過watch Zookeeper檢測所有的broker failure，并負(fù)責(zé)為所有受影響的parition選舉leader绘雁，再將相應(yīng)的leader調(diào)整命令發(fā)送至受影響的broker橡疼，過程如下圖所示。

目前consumer rebalance的控制策略是由每一個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.
在這種策略下，每一個consumer或者broker的增加或者減少都會觸發(fā)consumer rebalance讨永。因?yàn)槊總€consumer只負(fù)責(zé)調(diào)整自己所消費(fèi)的partition滔驶，為了保證整個consumer group的一致性，所以當(dāng)一個consumer觸發(fā)了rebalance時卿闹，該consumer group內(nèi)的其它所有consumer也應(yīng)該同時觸發(fā)rebalance揭糕。

目前（2015-01-19）最新版（0.8.2）Kafka采用的是上述方式徘钥。但該方式有不利的方面：
Herd effect 任何broker或者consumer的增減都會觸發(fā)所有的consumer的rebalance
Split Brain 每個consumer分別單獨(dú)通過Zookeeper判斷哪些partition down了蝗蛙，那么不同consumer從Zookeeper“看”到的view就可能不一樣，這就會造成錯誤的reblance嘗試技俐。而且有可能所有的consumer都認(rèn)為rebalance已經(jīng)完成了旋恼，但實(shí)際上可能并非如此吏口。

根據(jù)Kafka官方文檔，Kafka作者正在考慮在還未發(fā)布的 0.9.x版本中使用中心協(xié)調(diào)器(coordinator) 。大體思想是選舉出一個broker作為coordinator锨侯，由它watch Zookeeper嫩海，從而判斷是否有partition或者consumer的增減，然后生成rebalance命令囚痴，并檢查是否這些rebalance在所有相關(guān)的consumer中被執(zhí)行成功叁怪，如果不成功則重試，若成功則認(rèn)為此次rebalance成功（這個過程跟replication controller非常類似深滚，所以我很奇怪為什么當(dāng)初設(shè)計replication controller時沒有使用類似方式來解決consumer rebalance的問題）奕谭。流程如下：
[圖片上傳中。痴荐。血柳。（18）]
消息Deliver guarantee
通過上文介紹，想必讀者已經(jīng)明天了producer和consumer是如何工作的生兆，以及Kafka是如何做replication的难捌，接下來要討論的是Kafka如何確保消息在producer和consumer之間傳輸。有這么幾種可能的delivery guarantee：
At most once
消息可能會丟鸦难，但絕不會重復(fù)傳輸
At least one
消息絕不會丟根吁，但可能會重復(fù)傳輸
Exactly once
每條消息肯定會被傳輸一次且僅傳輸一次，很多時候這是用戶所想要的合蔽。Kafka的delivery guarantee semantic非常直接击敌。當(dāng)producer向broker發(fā)送消息時，一旦這條消息被commit拴事，因數(shù)replication的存在沃斤，它就不會丟。但是如果producer發(fā)送數(shù)據(jù)給broker后刃宵，遇到的網(wǎng)絡(luò)問題而造成通信中斷衡瓶，那producer就無法判斷該條消息是否已經(jīng)commit。這一點(diǎn)有點(diǎn)像向一個自動生成primary key的數(shù)據(jù)庫表中插入數(shù)據(jù)牲证。雖然Kafka無法確定網(wǎng)絡(luò)故障期間發(fā)生了什么鞍陨，但是producer可以生成一種類似于primary key的東西，發(fā)生故障時冪等性的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缝其。（僅針對Kafka consumer high level API）。consumer在從broker讀取消息后徘六，可以選擇commit内边，該操作會在Zookeeper中存下該consumer在該partition下讀取的消息的offset。該consumer下一次再讀該partition時會從下一條開始讀取待锈。如未commit漠其，下一次讀取的開始位置會跟上一次commit之后的開始位置相同。當(dāng)然可以將consumer設(shè)置為autocommit竿音，即consumer一旦讀到數(shù)據(jù)立即自動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了萄涯，下次重新開始工作后就無法讀到剛剛已提交而未處理的消息绪氛，這就對應(yīng)于 At most once

讀完消息先處理再commit。這種模式下窃判，如果處理完了消息在commit之前consumer crash了，下次重新開始工作時還會處理剛剛未commit的消息喇闸，實(shí)際上該消息已經(jīng)被處理過了袄琳。這就對應(yīng)于 At least once
。在很多情況使用場景下燃乍，消息都有一個primary key唆樊，所以消息的處理往往具有冪等性，即多次處理這一條消息跟只處理一次是等效的刻蟹，那就可以認(rèn)為是 Exactly once
逗旁。（人個感覺這種說法有些牽強(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和操作輸入存在同一個地方，會更簡潔和通用只嚣。這種方式可能更好沮稚，因?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)存儲系統(tǒng)協(xié)作祸穷，幸運(yùn)的是Kafka提供的offset可以使用這種方式非常直接非常容易。

Benchmark
紙上得來終覺淺勺三，絕知些事要躬行雷滚。筆者希望能親自測一下Kafka的性能，而非從網(wǎng)上找一些測試數(shù)據(jù)吗坚。所以筆者曾在0.8發(fā)布前兩個月做過詳細(xì)的Kafka0.8性能測試祈远，不過很可惜測試報告不慎丟失。所幸在網(wǎng)上找到了Kafka的創(chuàng)始人之一的 Jay Kreps的bechmark 商源。以下描述皆基于該benchmark车份。（該benchmark基于Kafka0.8.1）
測試環(huán)境
該benchmark用到了六臺機(jī)器，機(jī)器配置如下
Intel Xeon 2.5 GHz processor with six cores
Six 7200 RPM SATA drives
32GB of RAM
1Gb Ethernet
這6臺機(jī)器其中3臺用來搭建Kafka broker集群牡彻，另外3臺用來安裝Zookeeper及生成測試數(shù)據(jù)扫沼。6個drive都直接以非RAID方式掛載。實(shí)際上kafka對機(jī)器的需求與Hadoop的類似庄吼。

producer吞吐率
該項測試只測producer的吞吐率缎除，也就是數(shù)據(jù)只被持久化，沒有consumer讀數(shù)據(jù)总寻。
1個producer線程伴找，無replication
在這一測試中，創(chuàng)建了一個包含6個partition且沒有replication的topic废菱。然后通過一個線程盡可能快的生成50 million條比較短（payload100字節(jié)長）的消息技矮。測試結(jié)果是 821,557 records/second（ **78.3MB/second **）抖誉。
之所以使用短消息，是因?yàn)閷τ谙⑾到y(tǒng)來說這種使用場景更難衰倦。因?yàn)槿绻褂肕B/second來表征吞吐率袒炉，那發(fā)送長消息無疑能使得測試結(jié)果更好。
整個測試中樊零，都是用每秒鐘delivery的消息的數(shù)量乘以payload的長度來計算MB/second的我磁，沒有把消息的元信息算在內(nèi)，所以實(shí)際的網(wǎng)絡(luò)使用量會比這個大驻襟。對于本測試來說夺艰，每次還需傳輸額外的22個字節(jié)，包括一個可選的key沉衣，消息長度描述郁副，CRC等。另外豌习，還包含一些請求相關(guān)的overhead存谎，比如topic，partition肥隆，acknowledgement等既荚。這就導(dǎo)致我們比較難判斷是否已經(jīng)達(dá)到網(wǎng)卡極限，但是把這些overhead都算在吞吐率里面應(yīng)該更合理一些栋艳。因此恰聘，我們已經(jīng)基本達(dá)到了網(wǎng)卡的極限。
初步觀察此結(jié)果會認(rèn)為它比人們所預(yù)期的要高很多吸占，尤其當(dāng)考慮到Kafka要把數(shù)據(jù)持久化到磁盤當(dāng)中晴叨。實(shí)際上，如果使用隨機(jī)訪問數(shù)據(jù)系統(tǒng)旬昭，比如RDBMS篙螟，或者key-velue store菌湃，可預(yù)期的最高訪問頻率大概是5000到50000個請求每秒问拘，這和一個好的RPC層所能接受的遠(yuǎn)程請求量差不多。而該測試中遠(yuǎn)超于此的原因有兩個惧所。

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

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

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

3個producer,3個異步replication
該測試相當(dāng)于把上文中的1個producer,復(fù)制到了3臺不同的機(jī)器上（在1臺機(jī)器上跑多個實(shí)例對吞吐率的增加不會有太大幫忙，因?yàn)榫W(wǎng)卡已經(jīng)基本飽和了）顾瞻，這3個producer同時發(fā)送數(shù)據(jù)泼疑。整個集群的吞吐率為 **2,024,032 records/second **（ **193,0MB/second **）。
Producer Throughput Vs. Stored Data
消息系統(tǒng)的一個潛在的危險是當(dāng)數(shù)據(jù)能都存于內(nèi)存時性能很好荷荤，但當(dāng)數(shù)據(jù)量太大無法完全存于內(nèi)存中時（然后很多消息系統(tǒng)都會刪除已經(jīng)被消費(fèi)的數(shù)據(jù)退渗，但當(dāng)消費(fèi)速度比生產(chǎn)速度慢時移稳，仍會造成數(shù)據(jù)的堆積），數(shù)據(jù)會被轉(zhuǎn)移到磁盤会油，從而使得吞吐率下降秒裕，這又反過來造成系統(tǒng)無法及時接收數(shù)據(jù)。這樣就非常糟糕钞啸，而實(shí)際上很多情景下使用queue的目的就是解決數(shù)據(jù)消費(fèi)速度和生產(chǎn)速度不一致的問題几蜻。
但Kafka不存在這一問題，因?yàn)镵afka始終以O(shè)（1）的時間復(fù)雜度將數(shù)據(jù)持久化到磁盤体斩，所以其吞吐率不受磁盤上所存儲的數(shù)據(jù)量的影響梭稚。為了驗(yàn)證這一特性，做了一個長時間的大數(shù)據(jù)量的測試絮吵，下圖是吞吐率與數(shù)據(jù)量大小的關(guān)系圖弧烤。
[圖片上傳中。蹬敲。暇昂。（19）]上圖中有一些variance的存在，并可以明顯看到伴嗡，吞吐率并不受磁盤上所存數(shù)據(jù)量大小的影響急波。實(shí)際上從上圖可以看到，當(dāng)磁盤數(shù)據(jù)量達(dá)到1TB時瘪校，吞吐率和磁盤數(shù)據(jù)只有幾百M(fèi)B時沒有明顯區(qū)別澄暮。
這個variance是由Linux I/O管理造成的，它會把數(shù)據(jù)緩存起來再批量flush阱扬。上圖的測試結(jié)果是在生產(chǎn)環(huán)境中對Kafka集群做了些tuning后得到的泣懊，這些tuning方法可參考 這里 。

consumer吞吐率
需要注意的是麻惶，replication factor并不會影響consumer的吞吐率測試馍刮，因?yàn)閏onsumer只會從每個partition的leader讀數(shù)據(jù)，而與replicaiton factor無關(guān)窃蹋。同樣卡啰，consumer吞吐率也與同步復(fù)制還是異步復(fù)制無關(guān)。
1個consumer
該測試從有6個partition脐彩，3個replication的topic消費(fèi)50 million的消息碎乃。測試結(jié)果為 **940,521 records/second **（ **89.7MB/second **）姊扔。
可以看到惠奸，Kafkar的consumer是非常高效的。它直接從broker的文件系統(tǒng)里讀取文件塊恰梢。Kafka使用 sendfile API 來直接通過操作系統(tǒng)直接傳輸佛南，而不用把數(shù)據(jù)拷貝到用戶空間梗掰。該項測試實(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的吞吐率會比該項測試中的要高娃豹。

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

Producer and Consumer
上面的測試只是把producer和consumer分開測試躯畴，而該項測試同時運(yùn)行producer和consumer民鼓，這更接近使用場景。實(shí)際上目前的replication系統(tǒng)中follower就相當(dāng)于consumer在工作蓬抄。
該項測試丰嘉，在具有6個partition和3個replica的topic上同時使用1個producer和1個consumer，并且使用異步復(fù)制嚷缭。測試結(jié)果為 **795,064 records/second **（ **75.8MB/second **）供嚎。
可以看到，該項測試結(jié)果與單獨(dú)測試1個producer時的結(jié)果幾乎一致峭状。所以說consumer非常輕量級克滴。

消息長度對吞吐率的影響
上面的所有測試都基于短消息（payload 100字節(jié)），而正如上文所說优床，短消息對Kafka來說是更難處理的使用方式劝赔，可以預(yù)期，隨著消息長度的增大胆敞，records/second會減小着帽，但MB/second會有所提高。下圖是records/second與消息長度的關(guān)系圖移层。
[圖片上傳中仍翰。。观话。（20）]正如我們所預(yù)期的那樣予借，隨著消息長度的增加，每秒鐘所能發(fā)送的消息的數(shù)量逐漸減小。但是如果看每秒鐘發(fā)送的消息的總大小灵迫，它會隨著消息長度的增加而增加秦叛，如下圖所示。
[圖片上傳中瀑粥。挣跋。。（21）]從上圖可以看出狞换，當(dāng)消息長度為10字節(jié)時避咆，因?yàn)橐l繁入隊，花了太多時間獲取鎖修噪，CPU成了瓶頸牌借，并不能充分利用帶寬。但從100字節(jié)開始割按，我們可以看到帶寬的使用逐漸趨于飽和（雖然MB/second還是會隨著消息長度的增加而增加膨报，但增加的幅度也越來越小）适荣。

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

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

ProducerSetupbin/kafka-topics.sh --zookeeper esv4-hcl197.grid.linkedin.com:2181 --create --topic test-rep-one --partitions 6 --replication-factor 1bin/kafka-topics.sh --zookeeper esv4-hcl197.grid.linkedin.com:2181 --create --topic test --partitions 6 --replication-factor 3Single thread, no replicationbin/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=8196Single-thread, async 3x replicationbin/kafktopics.sh --zookeeper esv4-hcl197.grid.linkedin.com:2181 --create --topic test --partitions 6 --replication-factor 3bin/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=8196Single-thread, sync 3x replicationbin/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=64000Three Producers, 3x async replicationbin/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=8196Throughput Versus Stored Databin/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=8196Effect of message sizefor i in 10 100 1000 10000 100000;doecho ""echo $ibin/kafka-run-class.sh org.apache.kafka.clients.tools.ProducerPerformance test $((100010241024/$i)) $i -1 acks=1 bootstrap.servers=esv4-hcl198.grid.linkedin.com:9092 buffer.memory=67108864 batch.size=128000done;ConsumerConsumer throughputbin/kafka-consumer-perf-test.sh --zookeeper esv4-hcl197.grid.linkedin.com:2181 --messages 50000000 --topic test --threads 13 ConsumersOn three servers, run:bin/kafka-consumer-perf-test.sh --zookeeper esv4-hcl197.grid.linkedin.com:2181 --messages 50000000 --topic test --threads 1End-to-end Latencybin/kafka-run-class.sh kafka.tools.TestEndToEndLatency esv4-hcl198.grid.linkedin.com:9092 esv4-hcl197.grid.linkedin.com:2181 test 5000Producer and consumerbin/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=8196bin/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 onport=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 requestsnum.network.threads=4# The number of threads doing disk I/Onum.io.threads=8# The send buffer (SO_SNDBUF) used by the socket serversocket.send.buffer.bytes=1048576# The receive buffer (SO_RCVBUF) used by the socket serversocket.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 fileslog.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 deletionlog.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 policieslog.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 zookeeperzookeeper.connection.timeout.ms=1000000# metrics reporter propertieskafka.metrics.polling.interval.secs=5kafka.metrics.reporters=kafka.metrics.KafkaCSVMetricsReporterkafka.csv.metrics.dir=/tmp/kafka_metrics# Disable csv reporting by default.kafka.csv.metrics.reporter.enabled=falsereplica.lag.max.messages=10000000

讀者也可參考另外一份 Kafka性能測試報告
參考