Kafka背景及架構(gòu)介紹
Kafka是由LinkedIn開(kāi)發(fā)的一個(gè)分布式的消息系統(tǒng)晋柱，使用Scala編寫，它以可水平擴(kuò)展和高吞吐率而被廣泛使用邢笙。目前越來(lái)越多的開(kāi)源分布式處理系統(tǒng)如Cloudera泞当、Apache Storm洪鸭、Spark都支持與Kafka集成挡毅。InfoQ一直在緊密關(guān)注Kafka的應(yīng)用以及發(fā)展蒜撮，“Kafka剖析”專欄將會(huì)從架構(gòu)設(shè)計(jì)、實(shí)現(xiàn)跪呈、應(yīng)用場(chǎng)景段磨、性能等方面深度解析Kafka。
背景介紹
Kafka創(chuàng)建背景
Kafka是一個(gè)消息系統(tǒng)耗绿，原本開(kāi)發(fā)自LinkedIn薇溃，用作LinkedIn的活動(dòng)流（Activity Stream）和運(yùn)營(yíng)數(shù)據(jù)處理管道（Pipeline）的基礎(chǔ)。現(xiàn)在它已被多家不同類型的公司 作為多種類型的數(shù)據(jù)管道和消息系統(tǒng)使用缭乘。
活動(dòng)流數(shù)據(jù)是幾乎所有站點(diǎn)在對(duì)其網(wǎng)站使用情況做報(bào)表時(shí)都要用到的數(shù)據(jù)中最常規(guī)的部分×鹩茫活動(dòng)數(shù)據(jù)包括頁(yè)面訪問(wèn)量（Page View）堕绩、被查看內(nèi)容方面的信息以及搜索情況等內(nèi)容。這種數(shù)據(jù)通常的處理方式是先把各種活動(dòng)以日志的形式寫入某種文件邑时，然后周期性地對(duì)這些文件進(jìn)行統(tǒng)計(jì)分析奴紧。運(yùn)營(yíng)數(shù)據(jù)指的是服務(wù)器的性能數(shù)據(jù)（CPU、IO使用率晶丘、請(qǐng)求時(shí)間黍氮、服務(wù)日志等等數(shù)據(jù))。運(yùn)營(yíng)數(shù)據(jù)的統(tǒng)計(jì)方法種類繁多浅浮。
近年來(lái)沫浆，活動(dòng)和運(yùn)營(yíng)數(shù)據(jù)處理已經(jīng)成為了網(wǎng)站軟件產(chǎn)品特性中一個(gè)至關(guān)重要的組成部分，這就需要一套稍微更加復(fù)雜的基礎(chǔ)設(shè)施對(duì)其提供支持滚秩。
Kafka簡(jiǎn)介
Kafka是一種分布式的专执，基于發(fā)布/訂閱的消息系統(tǒng)。主要設(shè)計(jì)目標(biāo)如下：
- 以時(shí)間復(fù)雜度為O(1)的方式提供消息持久化能力郁油，即使對(duì)TB級(jí)以上數(shù)據(jù)也能保證常數(shù)時(shí)間復(fù)雜度的訪問(wèn)性能本股。
- 高吞吐率。即使在非常廉價(jià)的商用機(jī)器上也能做到單機(jī)支持每秒100K條以上消息的傳輸桐腌。
- 支持Kafka Server間的消息分區(qū)拄显，及分布式消費(fèi)，同時(shí)保證每個(gè)Partition內(nèi)的消息順序傳輸案站。
- 同時(shí)支持離線數(shù)據(jù)處理和實(shí)時(shí)數(shù)據(jù)處理躬审。
- Scale out：支持在線水平擴(kuò)展。
為何使用消息系統(tǒng)
解耦
在項(xiàng)目啟動(dòng)之初來(lái)預(yù)測(cè)將來(lái)項(xiàng)目會(huì)碰到什么需求，是極其困難的盒件。消息系統(tǒng)在處理過(guò)程中間插入了一個(gè)隱含的蹬碧、基于數(shù)據(jù)的接口層，兩邊的處理過(guò)程都要實(shí)現(xiàn)這一接口炒刁。這允許你獨(dú)立的擴(kuò)展或修改兩邊的處理過(guò)程恩沽，只要確保它們遵守同樣的接口約束。
冗余
有些情況下翔始，處理數(shù)據(jù)的過(guò)程會(huì)失敗罗心。除非數(shù)據(jù)被持久化，否則將造成丟失城瞎。消息隊(duì)列把數(shù)據(jù)進(jìn)行持久化直到它們已經(jīng)被完全處理渤闷，通過(guò)這一方式規(guī)避了數(shù)據(jù)丟失風(fēng)險(xiǎn)。許多消息隊(duì)列所采用的"插入-獲取-刪除"范式中脖镀，在把一個(gè)消息從隊(duì)列中刪除之前飒箭，需要你的處理系統(tǒng)明確的指出該消息已經(jīng)被處理完畢，從而確保你的數(shù)據(jù)被安全的保存直到你使用完畢蜒灰。
擴(kuò)展性
因?yàn)橄㈥?duì)列解耦了你的處理過(guò)程弦蹂，所以增大消息入隊(duì)和處理的頻率是很容易的，只要另外增加處理過(guò)程即可强窖。不需要改變代碼凸椿、不需要調(diào)節(jié)參數(shù)。擴(kuò)展就像調(diào)大電力按鈕一樣簡(jiǎn)單翅溺。
靈活性 & 峰值處理能力
在訪問(wèn)量劇增的情況下脑漫，應(yīng)用仍然需要繼續(xù)發(fā)揮作用，但是這樣的突發(fā)流量并不常見(jiàn)咙崎；如果為以能處理這類峰值訪問(wèn)為標(biāo)準(zhǔn)來(lái)投入資源隨時(shí)待命無(wú)疑是巨大的浪費(fèi)优幸。使用消息隊(duì)列能夠使關(guān)鍵組件頂住突發(fā)的訪問(wèn)壓力，而不會(huì)因?yàn)橥话l(fā)的超負(fù)荷的請(qǐng)求而完全崩潰叙凡。
可恢復(fù)性
系統(tǒng)的一部分組件失效時(shí)劈伴，不會(huì)影響到整個(gè)系統(tǒng)。消息隊(duì)列降低了進(jìn)程間的耦合度握爷，所以即使一個(gè)處理消息的進(jìn)程掛掉跛璧，加入隊(duì)列中的消息仍然可以在系統(tǒng)恢復(fù)后被處理。
順序保證
在大多使用場(chǎng)景下新啼，數(shù)據(jù)處理的順序都很重要追城。大部分消息隊(duì)列本來(lái)就是排序的，并且能保證數(shù)據(jù)會(huì)按照特定的順序來(lái)處理燥撞。Kafka保證一個(gè)Partition內(nèi)的消息的有序性座柱。
緩沖
在任何重要的系統(tǒng)中迷帜，都會(huì)有需要不同的處理時(shí)間的元素。例如色洞，加載一張圖片比應(yīng)用過(guò)濾器花費(fèi)更少的時(shí)間戏锹。消息隊(duì)列通過(guò)一個(gè)緩沖層來(lái)幫助任務(wù)最高效率的執(zhí)行———寫入隊(duì)列的處理會(huì)盡可能的快速。該緩沖有助于控制和優(yōu)化數(shù)據(jù)流經(jīng)過(guò)系統(tǒng)的速度火诸。
異步通信
很多時(shí)候锦针，用戶不想也不需要立即處理消息。消息隊(duì)列提供了異步處理機(jī)制置蜀，允許用戶把一個(gè)消息放入隊(duì)列奈搜，但并不立即處理它。想向隊(duì)列中放入多少消息就放多少盯荤，然后在需要的時(shí)候再去處理它們馋吗。
Kafka架構(gòu)
術(shù)語(yǔ)
Broker
Kafka集群包含一個(gè)或多個(gè)服務(wù)器，這種服務(wù)器被稱為broker
Topic
每條發(fā)布到Kafka集群的消息都有一個(gè)類別秋秤，這個(gè)類別被稱為Topic宏粤。(物理上不同Topic的消息分開(kāi)存儲(chǔ)，邏輯上一個(gè)Topic的消息雖然保存于一個(gè)或多個(gè)broker上灼卢，但用戶只需指定消息的Topic即可生產(chǎn)或者消費(fèi)數(shù)據(jù)而不需要關(guān)心數(shù)據(jù)存儲(chǔ)于何處)
Partition
Partition是物理上的概念商架，每個(gè)Topic包含一個(gè)或多個(gè)Partition
Producer
負(fù)責(zé)發(fā)布消息到Kafka broker(生產(chǎn)消息)
Consumer
消息消費(fèi)者，向Kafka broker讀取消息的客戶端芥玉。
Consumer Group
每個(gè)Consumer屬于一個(gè)特定的Consumer Group(可為每個(gè)Consumer制定group name，若不指定group name則屬于默認(rèn)的group备图，并且每個(gè)Consumer自成一組)灿巧。
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通過(guò)Zookeeper管理集群配置村刨，選舉leader告抄，以及在Consumer Group發(fā)生變化時(shí)進(jìn)行rebalance。Producer使用push模式將消息發(fā)布到broker嵌牺，Consumer使用pull模式從broker訂閱并消費(fèi)消息打洼。
Topic & Partition
Topic在邏輯上可以被認(rèn)為是一個(gè)queue龄糊，每條消費(fèi)都必須制定它的Topic，可以簡(jiǎn)單理解為必須指明把這條消息放進(jìn)那個(gè)queue里募疮。為了使Kafka的吞吐量可以線性提高炫惩，物理上把Topic分成一個(gè)或者多個(gè)Partiion，每個(gè)Partition在物理上對(duì)應(yīng)一個(gè)文件夾阿浓，該文件夾下存儲(chǔ)這個(gè)Partition的所有消息和索引文件他嚷。
每個(gè)日志文件都是一個(gè)log entry序列，每個(gè)log entry包含一個(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 entry并非由一個(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ì)保留所有的消息烦感，無(wú)論其被消費(fèi)與否巡社。當(dāng)然，因?yàn)榇疟P限制手趣，不可能永久保留所有數(shù)據(jù)（實(shí)際上也沒(méi)必要）晌该，因此Kafka提供兩種策略刪除舊數(shù)據(jù)。一是基于時(shí)間绿渣，二是基于Partition文件大小朝群。例如可以通過(guò)配置$KAFKA_HOME/config/server.properties，讓Kafka刪除一周前的數(shù)據(jù)中符，也可在Partition文件超過(guò)1GB時(shí)刪除舊數(shù)據(jù)潜圃，配置如下所示。
# 日志保存時(shí)間
log.retention.hours=168
# 日志段文件的最大值舟茶，當(dāng)超過(guò)這個(gè)值時(shí)將創(chuàng)建一個(gè)新的日志段
log.segment.bytes=1073741824
# 檢查日志段的時(shí)間間隔谭期，看看是否可以根據(jù)保留策略刪除它們
log.retention.check.interval.ms=300000
# 啟用清理日志
log.cleaner.enable=false
這里要注意堵第，因?yàn)镵afka讀取特定消息的時(shí)間復(fù)雜度為O(1)，即與文件大小無(wú)關(guān)隧出，所以這里刪除過(guò)期文件與提高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)完一條消息后遞增該offset圆雁。當(dāng)然，Consumer也可將offset設(shè)成一個(gè)較小的值帆谍，重新消費(fèi)一些消息伪朽。因?yàn)閛ffet由Consumer控制，所以Kafka broker是無(wú)狀態(tài)的汛蝙，它不需要標(biāo)記哪些消息被哪些消費(fèi)過(guò)烈涮，也不需要通過(guò)broker去保證同一個(gè)Consumer Group只有一個(gè)Consumer能消費(fèi)某一條消息，因此也就不需要鎖機(jī)制窖剑，這也為Kafka的高吞吐率提供了有力保障坚洽。
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中通過(guò)配置項(xiàng)num.partitions來(lái)指定新建Topic的默認(rèn)Partition數(shù)量，也可在創(chuàng)建Topic時(shí)通過(guò)參數(shù)指定历帚，同時(shí)也可以在Topic創(chuàng)建之后通過(guò)Kafka提供的工具修改滔岳。
在發(fā)送一條消息時(shí)，可以指定這條消息的key挽牢，Producer根據(jù)這個(gè)Key和Partiton機(jī)制來(lái)判斷應(yīng)該將這條消息發(fā)送到那個(gè)Partition谱煤。Partition機(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),序號(hào)從0開(kāi)始）
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);
}
}
}
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)這一消息。
這是Kafka用來(lái)實(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。
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來(lái)不及處理消息，典型的表現(xiàn)就是拒絕服務(wù)以及網(wǎng)絡(luò)擁塞佩迟。而pull模式則可以根據(jù)Consumer的消費(fèi)能力以適當(dāng)?shù)乃俾氏M(fèi)消息团滥。
對(duì)于Kafka而言，pull模式更合適报强。pull模式可簡(jiǎn)化broker的設(shè)計(jì)灸姊，Consumer可自主控制消費(fèi)消息的速率，同時(shí)Consumer可以自己控制消費(fèi)方式——即可批量消費(fèi)也可逐條消費(fèi)秉溉，同時(shí)還能選擇不同的提交方式從而實(shí)現(xiàn)不同的傳輸語(yǔ)義力惯。
本文轉(zhuǎn)發(fā)自技術(shù)世界，原文鏈接 http://www.jasongj.com/2015/03/10/KafkaColumn1/
