簡介
Kafka是一種分布式的，基于發(fā)布/訂閱的消息系統(tǒng)非洲。使用Scala編寫，它以可水平擴(kuò)展和高吞吐率而被廣泛使用。
Kafka架構(gòu)
Terminoliogy
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
分割咆蒿、分區(qū)
Producer：負(fù)責(zé)發(fā)布消息到Kafka broker
Consumer：消息消費(fèi)者东抹，向Kafka broker讀取消息的客戶端蚂子。
-
Consumer Group：每個(gè)Consumer屬于一個(gè)特定的Consumer Group
可為每個(gè)Consumer指定group name，若不指定group name則屬于默認(rèn)的group
Kafka拓?fù)浣Y(jié)構(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)消息。
Topic & Partition
Topic =Queue
Topic在邏輯上可以被認(rèn)為是一個(gè)queue箩朴，每條消費(fèi)都必須指定它的Topic岗喉，可以簡單理解為必須指明把這條消息放進(jìn)哪個(gè)queue里。為了使得Kafka的吞吐率可以線性提高炸庞，物理上把Topic分成一個(gè)或多個(gè) Partition钱床，每個(gè)Partition在物理上對(duì)應(yīng)一個(gè)文件夾，該文件夾下存儲(chǔ)這個(gè)Partition的所有消息和索引文件埠居。若創(chuàng)建topic1和topic2兩個(gè)topic查牌，且分別有13個(gè)和19個(gè)分區(qū)，則整個(gè)集群上會(huì)相應(yīng)會(huì)生成共32個(gè)文件夾（本文所用集群共8個(gè)節(jié)點(diǎn)滥壕，此處topic1和topic2 replication-factor均為1）纸颜，如下圖所示。
每個(gè)日志文件都是一個(gè)log entrie序列绎橘，每個(gè)log entrie包含一個(gè)4字節(jié)整型數(shù)值（值為N+5）胁孙，1個(gè)字節(jié)的"magic value"，4個(gè)字節(jié)的CRC校驗(yàn)碼称鳞，其后跟N個(gè)字節(jié)的消息體涮较。每條消息都有一個(gè)當(dāng)前Partition下唯一的64字節(jié)的offset，它指明了這條消息的起始位置冈止。磁盤上存儲(chǔ)的消息格式如下：
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到該P(yáng)artition中屋剑，屬于順序?qū)懘疟P润匙，因此效率非常高（經(jīng)驗(yàn)證，順序?qū)懘疟P效率比隨機(jī)寫內(nèi)存還要高唉匾，這是Kafka高吞吐率的一個(gè)很重要的保證）孕讳。
對(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ù)谒撼，也可在Partition文件超過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
這里要注意廓潜，因?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)完一條消息后遞增該offset樱蛤。當(dāng)然，Consumer也可將offset設(shè)成一個(gè)較小的值剑鞍，重新消費(fèi)一些消息。因?yàn)閛ffet由Consumer控制爽醋，所以Kafka broker是無狀態(tài)的蚁署，它不需要標(biāo)記哪些消息被哪些消費(fèi)過，也不需要通過broker去保證同一個(gè)Consumer Group只有一個(gè)Consumer能消費(fèi)某一條消息蚂四，因此也就不需要鎖機(jī)制晌杰，這也為Kafka的高吞吐率提供了有力保障。
備注：
Kafka讀取特定消息的時(shí)間復(fù)雜度為O(1)肋演，按順序讀取，即不受文件大小約束
Kafka broker是無狀態(tài)的梗夸，broker沒有鎖機(jī)制，提高了高吞吐率
Producer消息路由
Producer發(fā)送消息到broker時(shí)，會(huì)根據(jù)Paritition機(jī)制選擇將其存儲(chǔ)到哪一個(gè)Partition该酗。如果Partition機(jī)制設(shè)置合理莱衩，所有消息可以均勻分布到不同的Partition里睹晒，這樣就實(shí)現(xiàn)了負(fù)載均衡。如果一個(gè)Topic對(duì)應(yīng)一個(gè)文件，那這個(gè)文件所在的機(jī)器I/O將會(huì)成為這個(gè)Topic的性能瓶頸呆盖，而有了Partition后，不同的消息可以并行寫入不同broker的不同Partition里尤筐，極大的提高了吞吐率奖磁★跽可以在$KAFKA_HOME/config/server.properties中通過配置項(xiàng)num.partitions來指定新建Topic的默認(rèn)Partition數(shù)量吞彤，也可在創(chuàng)建Topic時(shí)通過參數(shù)指定，同時(shí)也可以在Topic創(chuàng)建之后通過Kafka提供的工具修改破加。
備注:不同的消息可以并行寫入不同broker的不同Partition里
在發(fā)送一條消息時(shí)锭环，可以指定這條消息的key吱七，Producer根據(jù)這個(gè)key和Partition機(jī)制來判斷應(yīng)該將這條消息發(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),序號(hào)從0開始）
//發(fā)送消息時(shí)失乾，按照此機(jī)制將消息發(fā)送到對(duì)應(yīng)的Parition上 (key---partition)
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);
}
}
}
如果將上例中的類作為partition.class碱茁，并通過如下代碼發(fā)送20條消息（key分別為0聋袋，1，2干毅，3）至topic3（包含4個(gè)Partition）。
//將20條信息發(fā)送到topic3對(duì)應(yīng)的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++）//4個(gè)partition
{
//第i個(gè)消息給第j個(gè)paitition
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開始）。下圖所示是通過Java程序調(diào)用Consumer后打印出的消息列表徽缚。
Consumer Group
（本節(jié)所有描述都是基于Consumer hight level API而非low level API）憨奸。
使用Consumer high level API時(shí)，同一Topic的一條消息只能被同一個(gè)Consumer Group內(nèi)的一個(gè)Consumer消費(fèi)凿试，但多個(gè)Consumer Group可同時(shí)消費(fèi)這一消息排宰。
廣播：不同的Consumer Group
單播：在同一個(gè)Consumer Group
離線處理：Hadoop
實(shí)時(shí)處理：Storm
這是Kafka用來實(shí)現(xiàn)一個(gè)Topic消息的廣播（發(fā)給所有的Consumer）和單播（發(fā)給某一個(gè)Consumer）的手段。一個(gè)Topic可以對(duì)應(yīng)多個(gè)Consumer Group那婉。如果需要實(shí)現(xiàn)廣播板甘，只要每個(gè)Consumer有一個(gè)獨(dú)立的Group就可以了。要實(shí)現(xiàn)單播只要所有的Consumer在同一個(gè)Group里详炬。用Consumer Group還可以將Consumer進(jìn)行自由的分組而不需要多次發(fā)送消息到不同的Topic盐类。
實(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的一種簡化部署示意圖礼仗。
下面這個(gè)例子更清晰地展示了Kafka Consumer Group的特性吐咳。首先創(chuàng)建一個(gè)Topic (名為topic1，包含3個(gè)Partition)元践，然后創(chuàng)建一個(gè)屬于group1的Consumer實(shí)例韭脊，并創(chuàng)建三個(gè)屬于group2的Consumer實(shí)例，最后通過Producer向topic1發(fā)送key分別為1单旁，2沪羔，3的消息。結(jié)果發(fā)現(xiàn)屬于group1的Consumer收到了所有的這三條消息象浑，同時(shí)group2中的3個(gè)Consumer分別收到了key為1蔫饰，2，3的消息愉豺。如下圖所示篓吁。
Push vs. Pull
作為一個(gè)消息系統(tǒng)，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)消息城须。
對(duì)于Kafka而言蚤认，pull模式更合適。pull模式可簡化broker的設(shè)計(jì)糕伐，Consumer可自主控制消費(fèi)消息的速率砰琢，同時(shí)Consumer可以自己控制消費(fèi)方式——即可批量消費(fèi)也可逐條消費(fèi)，同時(shí)還能選擇不同的提交方式從而實(shí)現(xiàn)不同的傳輸語義良瞧。
Kafka delivery guarantee
有這么幾種可能的delivery guarantee：
At most once 消息可能會(huì)丟陪汽，但絕不會(huì)重復(fù)傳輸
At least one 消息絕不會(huì)丟，但可能會(huì)重復(fù)傳輸
-
Exactly once 每條消息肯定會(huì)被傳輸一次且僅傳輸一次褥蚯，很多時(shí)候這是用戶所想要的挚冤。
當(dāng)Producer向broker發(fā)送消息時(shí)，一旦這條消息被commit赞庶，因數(shù)replication的存在训挡，它就不會(huì)丟。但是如果Producer發(fā)送數(shù)據(jù)給broker后歧强，遇到網(wǎng)絡(luò)問題而造成通信中斷澜薄，那Producer就無法判斷該條消息是否已經(jīng)commit。雖然Kafka無法確定網(wǎng)絡(luò)故障期間發(fā)生了什么摊册，但是Producer可以生成一種類似于主鍵的東西肤京，發(fā)生故障時(shí)冪等性的重試多次，這樣就做到了Exactly once茅特。截止到目前(Kafka 0.8.2版本忘分，2015-03-04)，這一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語義舟山。（僅針對(duì)Kafka consumer high level API）。Consumer在從broker讀取消息后，可以選擇commit累盗，該操作會(huì)在Zookeeper中保存該Consumer在該P(yáng)artition中讀取的消息的offset寒矿。該Consumer下一次再讀該P(yáng)artition時(shí)會(huì)從下一條開始讀取。如未commit若债，下一次讀取的開始位置會(huì)跟上一次commit之后的開始位置相同符相。當(dāng)然可以將Consumer設(shè)置為autocommit，即Consumer一旦讀到數(shù)據(jù)立即自動(dòng)commit蠢琳。如果只討論這一讀取消息的過程啊终，那Kafka是確保了Exactly once。但實(shí)際使用中應(yīng)用程序并非在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永乌。在很多使用場景下惑申，消息都有一個(gè)主鍵，所以消息的處理往往具有冪等性翅雏，即多次處理這一條消息跟只處理一次是等效的圈驼，那就可以認(rèn)為是Exactly once。（筆者認(rèn)為這種說法比較牽強(qiáng)望几，畢竟它不是Kafka本身提供的機(jī)制绩脆，主鍵本身也并不能完全保證操作的冪等性。而且實(shí)際上我們說delivery guarantee 語義是討論被處理多少次橄抹，而非處理結(jié)果怎樣靴迫，因?yàn)樘幚矸绞蕉喾N多樣，我們不應(yīng)該把處理過程的特性——如是否冪等性楼誓，當(dāng)成Kafka本身的Feature）
如果一定要做到Exactly once玉锌，就需要協(xié)調(diào)offset和實(shí)際操作的輸出。精典的做法是引入兩階段提交疟羹。如果能讓offset和操作輸入存在同一個(gè)地方主守，會(huì)更簡潔和通用禀倔。這種方式可能更好，因?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要求與外部存儲(chǔ)系統(tǒng)協(xié)作臂寝，幸運(yùn)的是Kafka提供的offset可以非常直接非常容易得使用這種方式章鲤。
