原文地址：http://www.infoq.com/cn/articles/kafka-analysis-part-1?utm_source=infoq&utm_campaign=user_page&utm_medium=link

Kafka是由LinkedIn開發(fā)的一個分布式的消息系統(tǒng)，使用Scala編寫，它以可水平擴展和高吞吐率而被廣泛使用。目前越來越多的開源分布式處理系統(tǒng)如Cloudera、Apache Storm、Spark都支持與Kafka集成。InfoQ一直在緊密關注Kafka的應用以及發(fā)展，“Kafka剖析”專欄將會從架構設計、實現(xiàn)、應用場景、性能等方面深度解析Kafka。

背景介紹

Kafka創(chuàng)建背景

Kafka是一個消息系統(tǒng)，原本開發(fā)自LinkedIn，用作LinkedIn的活動流（Activity Stream）和運營數(shù)據(jù)處理管道（Pipeline）的基礎。現(xiàn)在它已被多家不同類型的公司?作為多種類型的數(shù)據(jù)管道和消息系統(tǒng)使用。

活動流數(shù)據(jù)是幾乎所有站點在對其網(wǎng)站使用情況做報表時都要用到的數(shù)據(jù)中最常規(guī)的部分。活動數(shù)據(jù)包括頁面訪問量（Page View）、被查看內(nèi)容方面的信息以及搜索情況等內(nèi)容。這種數(shù)據(jù)通常的處理方式是先把各種活動以日志的形式寫入某種文件，然后周期性地對這些文件進行統(tǒng)計分析。運營數(shù)據(jù)指的是服務器的性能數(shù)據(jù)（CPU、IO使用率、請求時間、服務日志等等數(shù)據(jù))。運營數(shù)據(jù)的統(tǒng)計方法種類繁多。

Kafka簡介

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

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

為何使用消息系統(tǒng)

• 解耦

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

• 冗余

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

• 擴展性

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

• 靈活性 & 峰值處理能力

  在訪問量劇增的情況下，應用仍然需要繼續(xù)發(fā)揮作用，但是這樣的突發(fā)流量并不常見；如果為以能處理這類峰值訪問為標準來投入資源隨時待命無疑是巨大的浪費。使用消息隊列能夠使關鍵組件頂住突發(fā)的訪問壓力，而不會因為突發(fā)的超負荷的請求而完全崩潰。

• 可恢復性

  系統(tǒng)的一部分組件失效時，不會影響到整個系統(tǒng)。消息隊列降低了進程間的耦合度，所以即使一個處理消息的進程掛掉，加入隊列中的消息仍然可以在系統(tǒng)恢復后被處理。

• 順序保證

  在大多使用場景下，數(shù)據(jù)處理的順序都很重要。大部分消息隊列本來就是排序的，并且能保證數(shù)據(jù)會按照特定的順序來處理。Kafka保證一個Partition內(nèi)的消息的有序性。

• 緩沖

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

• 異步通信

  很多時候，用戶不想也不需要立即處理消息。消息隊列提供了異步處理機制，允許用戶把一個消息放入隊列，但并不立即處理它。想向隊列中放入多少消息就放多少，然后在需要的時候再去處理它們。

常用Message Queue對比

• RabbitMQ

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

• Redis

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

• ZeroMQ

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

• ActiveMQ

  ActiveMQ是Apache下的一個子項目。 類似于ZeroMQ，它能夠以代理人和點對點的技術實現(xiàn)隊列。同時類似于RabbitMQ，它少量代碼就可以高效地實現(xiàn)高級應用場景。

• Kafka/Jafka

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

Kafka架構

Terminology

• Broker

  Kafka集群包含一個或多個服務器，這種服務器被稱為broker

• Topic

  每條發(fā)布到Kafka集群的消息都有一個類別，這個類別被稱為Topic。（物理上不同Topic的消息分開存儲，邏輯上一個Topic的消息雖然保存于一個或多個broker上但用戶只需指定消息的Topic即可生產(chǎn)或消費數(shù)據(jù)而不必關心數(shù)據(jù)存于何處）

• Partition

  Parition是物理上的概念，每個Topic包含一個或多個Partition.

• Producer

  負責發(fā)布消息到Kafka broker

• Consumer

  消息消費者，向Kafka broker讀取消息的客戶端。

• Consumer Group

  每個Consumer屬于一個特定的Consumer Group（可為每個Consumer指定group name，若不指定group name則屬于默認的group）。

Kafka拓撲結構

如上圖所示，一個典型的Kafka集群中包含若干Producer（可以是web前端產(chǎn)生的Page View，或者是服務器日志，系統(tǒng)CPU、Memory等），若干broker（Kafka支持水平擴展，一般broker數(shù)量越多，集群吞吐率越高），若干Consumer Group，以及一個Zookeeper集群。Kafka通過Zookeeper管理集群配置，選舉leader，以及在Consumer Group發(fā)生變化時進行rebalance。Producer使用push模式將消息發(fā)布到broker，Consumer使用pull模式從broker訂閱并消費消息。

Topic & Partition

Topic在邏輯上可以被認為是一個queue，每條消費都必須指定它的Topic，可以簡單理解為必須指明把這條消息放進哪個queue里。為了使得Kafka的吞吐率可以線性提高，物理上把Topic分成一個或多個Partition，每個Partition在物理上對應一個文件夾，該文件夾下存儲這個Partition的所有消息和索引文件。若創(chuàng)建topic1和topic2兩個topic，且分別有13個和19個分區(qū)，則整個集群上會相應會生成共32個文件夾（本文所用集群共8個節(jié)點，此處topic1和topic2 replication-factor均為1），如下圖所示。

每個日志文件都是一個log entrie序列，每個log entrie包含一個4字節(jié)整型數(shù)值（值為N+5），1個字節(jié)的"magic value"，4個字節(jié)的CRC校驗碼，其后跟N個字節(jié)的消息體。每條消息都有一個當前Partition下唯一的64字節(jié)的offset，它指明了這條消息的起始位置。磁盤上存儲的消息格式如下：

message length ： 4 bytes (value: 1+4+n) "magic" value ： 1 byte crc ： 4 bytes payload ： n bytes

這個log entries并非由一個文件構成，而是分成多個segment，每個segment以該segment第一條消息的offset命名并以“.kafka”為后綴。另外會有一個索引文件，它標明了每個segment下包含的log entry的offset范圍，如下圖所示。

因為每條消息都被append到該Partition中，屬于順序寫磁盤，因此效率非常高（經(jīng)驗證，順序寫磁盤效率比隨機寫內(nèi)存還要高，這是Kafka高吞吐率的一個很重要的保證）。

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

這里要注意，因為Kafka讀取特定消息的時間復雜度為O(1)，即與文件大小無關，所以這里刪除過期文件與提高Kafka性能無關。選擇怎樣的刪除策略只與磁盤以及具體的需求有關。另外，Kafka會為每一個Consumer Group保留一些metadata信息——當前消費的消息的position，也即offset。這個offset由Consumer控制。正常情況下Consumer會在消費完一條消息后遞增該offset。當然，Consumer也可將offset設成一個較小的值，重新消費一些消息。因為offet由Consumer控制，所以Kafka broker是無狀態(tài)的，它不需要標記哪些消息被哪些消費過，也不需要通過broker去保證同一個Consumer Group只有一個Consumer能消費某一條消息，因此也就不需要鎖機制，這也為Kafka的高吞吐率提供了有力保障。

Producer消息路由

Producer發(fā)送消息到broker時，會根據(jù)Paritition機制選擇將其存儲到哪一個Partition。如果Partition機制設置合理，所有消息可以均勻分布到不同的Partition里，這樣就實現(xiàn)了負載均衡。如果一個Topic對應一個文件，那這個文件所在的機器I/O將會成為這個Topic的性能瓶頸，而有了Partition后，不同的消息可以并行寫入不同broker的不同Partition里，極大的提高了吞吐率。可以在$KAFKA_HOME/config/server.properties中通過配置項num.partitions來指定新建Topic的默認Partition數(shù)量，也可在創(chuàng)建Topic時通過參數(shù)指定，同時也可以在Topic創(chuàng)建之后通過Kafka提供的工具修改。

在發(fā)送一條消息時，可以指定這條消息的key，Producer根據(jù)這個key和Partition機制來判斷應該將這條消息發(fā)送到哪個Parition。Paritition機制可以通過指定Producer的paritition. class這一參數(shù)來指定，該class必須實現(xiàn)kafka.producer.Partitioner接口。本例中如果key可以被解析為整數(shù)則將對應的整數(shù)與Partition總數(shù)取余，該消息會被發(fā)送到該數(shù)對應的Partition。（每個Parition都會有個序號,序號從0開始）

import kafka.producer.Partitioner; import kafka.utils.VerifiableProperties;public class JasonPartitioner<T> implements Partitioner {public JasonPartitioner(VerifiableProperties verifiableProperties) {}@Overridepublic 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個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開始）。下圖所示是通過Java程序調(diào)用Consumer后打印出的消息列表。

Consumer Group

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

使用Consumer high level API時，同一Topic的一條消息只能被同一個Consumer Group內(nèi)的一個Consumer消費，但多個Consumer Group可同時消費這一消息。

這是Kafka用來實現(xiàn)一個Topic消息的廣播（發(fā)給所有的Consumer）和單播（發(fā)給某一個Consumer）的手段。一個Topic可以對應多個Consumer Group。如果需要實現(xiàn)廣播，只要每個Consumer有一個獨立的Group就可以了。要實現(xiàn)單播只要所有的Consumer在同一個Group里。用Consumer Group還可以將Consumer進行自由的分組而不需要多次發(fā)送消息到不同的Topic。

實際上，Kafka的設計理念之一就是同時提供離線處理和實時處理。根據(jù)這一特性，可以使用Storm這種實時流處理系統(tǒng)對消息進行實時在線處理，同時使用Hadoop這種批處理系統(tǒng)進行離線處理，還可以同時將數(shù)據(jù)實時備份到另一個數(shù)據(jù)中心，只需要保證這三個操作所使用的Consumer屬于不同的Consumer Group即可。下圖是Kafka在Linkedin的一種簡化部署示意圖。

下面這個例子更清晰地展示了Kafka Consumer Group的特性。首先創(chuàng)建一個Topic (名為topic1，包含3個Partition)，然后創(chuàng)建一個屬于group1的Consumer實例，并創(chuàng)建三個屬于group2的Consumer實例，最后通過Producer向topic1發(fā)送key分別為1，2，3的消息。結果發(fā)現(xiàn)屬于group1的Consumer收到了所有的這三條消息，同時group2中的3個Consumer分別收到了key為1，2，3的消息。如下圖所示。

Push vs. Pull

作為一個消息系統(tǒng)，Kafka遵循了傳統(tǒng)的方式，選擇由Producer向broker push消息并由Consumer從broker pull消息。一些logging-centric system，比如Facebook的Scribe和Cloudera的Flume，采用push模式。事實上，push模式和pull模式各有優(yōu)劣。

push模式很難適應消費速率不同的消費者，因為消息發(fā)送速率是由broker決定的。push模式的目標是盡可能以最快速度傳遞消息，但是這樣很容易造成Consumer來不及處理消息，典型的表現(xiàn)就是拒絕服務以及網(wǎng)絡擁塞。而pull模式則可以根據(jù)Consumer的消費能力以適當?shù)乃俾氏M消息。

對于Kafka而言，pull模式更合適。pull模式可簡化broker的設計，Consumer可自主控制消費消息的速率，同時Consumer可以自己控制消費方式——即可批量消費也可逐條消費，同時還能選擇不同的提交方式從而實現(xiàn)不同的傳輸語義。

Kafka delivery guarantee

有這么幾種可能的delivery guarantee：

• At most once 消息可能會丟，但絕不會重復傳輸
• At least one 消息絕不會丟，但可能會重復傳輸

• Exactly once 每條消息肯定會被傳輸一次且僅傳輸一次，很多時候這是用戶所想要的。

  當Producer向broker發(fā)送消息時，一旦這條消息被commit，因數(shù)replication的存在，它就不會丟。但是如果Producer發(fā)送數(shù)據(jù)給broker后，遇到網(wǎng)絡問題而造成通信中斷，那Producer就無法判斷該條消息是否已經(jīng)commit。雖然Kafka無法確定網(wǎng)絡故障期間發(fā)生了什么，但是Producer可以生成一種類似于主鍵的東西，發(fā)生故障時冪等性的重試多次，這樣就做到了Exactly once。截止到目前(Kafka 0.8.2版本，2015-03-04)，這一Feature還并未實現(xiàn)，有希望在Kafka未來的版本中實現(xiàn)。（所以目前默認情況下一條消息從Producer到broker是確保了At least once，可通過設置Producer異步發(fā)送實現(xiàn)At most once）。

  接下來討論的是消息從broker到Consumer的delivery guarantee語義。（僅針對Kafka consumer high level API）。Consumer在從broker讀取消息后，可以選擇commit，該操作會在Zookeeper中保存該Consumer在該Partition中讀取的消息的offset。該Consumer下一次再讀該Partition時會從下一條開始讀取。如未commit，下一次讀取的開始位置會跟上一次commit之后的開始位置相同。當然可以將Consumer設置為autocommit，即Consumer一旦讀到數(shù)據(jù)立即自動commit。如果只討論這一讀取消息的過程，那Kafka是確保了Exactly once。但實際使用中應用程序并非在Consumer讀取完數(shù)據(jù)就結束了，而是要進行進一步處理，而數(shù)據(jù)處理與commit的順序在很大程度上決定了消息從broker和consumer的delivery guarantee semantic。

• 讀完消息先commit再處理消息。這種模式下，如果Consumer在commit后還沒來得及處理消息就crash了，下次重新開始工作后就無法讀到剛剛已提交而未處理的消息，這就對應于At most once

• 讀完消息先處理再commit。這種模式下，如果在處理完消息之后commit之前Consumer crash了，下次重新開始工作時還會處理剛剛未commit的消息，實際上該消息已經(jīng)被處理過了。這就對應于At least once。在很多使用場景下，消息都有一個主鍵，所以消息的處理往往具有冪等性，即多次處理這一條消息跟只處理一次是等效的，那就可以認為是Exactly once。（筆者認為這種說法比較牽強，畢竟它不是Kafka本身提供的機制，主鍵本身也并不能完全保證操作的冪等性。而且實際上我們說delivery guarantee 語義是討論被處理多少次，而非處理結果怎樣，因為處理方式多種多樣，我們不應該把處理過程的特性——如是否冪等性，當成Kafka本身的Feature）

• 如果一定要做到Exactly once，就需要協(xié)調(diào)offset和實際操作的輸出。精典的做法是引入兩階段提交。如果能讓offset和操作輸入存在同一個地方，會更簡潔和通用。這種方式可能更好，因為許多輸出系統(tǒng)可能不支持兩階段提交。比如，Consumer拿到數(shù)據(jù)后可能把數(shù)據(jù)放到HDFS，如果把最新的offset和數(shù)據(jù)本身一起寫到HDFS，那就可以保證數(shù)據(jù)的輸出和offset的更新要么都完成，要么都不完成，間接實現(xiàn)Exactly once。（目前就high level API而言，offset是存于Zookeeper中的，無法存于HDFS，而low level API的offset是由自己去維護的，可以將之存于HDFS中）

總之，Kafka默認保證At least once，并且允許通過設置Producer異步提交來實現(xiàn)At most once。而Exactly once要求與外部存儲系統(tǒng)協(xié)作，幸運的是Kafka提供的offset可以非常直接非常容易得使用這種方式。

作者簡介

郭俊（Jason），碩士，從事大數(shù)據(jù)平臺研發(fā)工作，精通Kafka等分布式消息系統(tǒng)及Storm等流式處理系統(tǒng)。

聯(lián)系方式：新浪微博： 郭俊_Jason 微信：habren 博客：http://www.jasongj.com

下篇預告

下一篇將深入講解Kafka是如何做Replication和Leader Election的。在Kafka0.8以前的版本中，如果某個broker宕機，或者磁盤出現(xiàn)問題，則該broker上所有partition的數(shù)據(jù)都會丟失。而Kafka0.8以后加入了Replication機制，可以將每個Partition的數(shù)據(jù)備份多份，即使某些broker宕機也能保證系統(tǒng)的可用性和數(shù)據(jù)的完整性。

轉載于:https://www.cnblogs.com/davidwang456/articles/7027520.html

總結

以上是生活随笔為你收集整理的Kafka剖析（一）：Kafka背景及架构介绍--转的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。