kafka：一個分佈式消息系統

1.背景

最近由於工做須要，調研了追求高吞吐的輕量級消息系統Kafka，打算替換掉線上運行的ActiveMQ，主要是由於明年的預算日流量有十億，而ActiveMQ的分佈式實現的很奇怪，因此但願找一個適合分佈式的消息系統。

如下是內容是調研過程當中總結的一些知識和經驗，歡迎拍磚。

2.基礎知識

2.1.什麼是消息隊列

首先，咱們來看看什麼是消息隊列，維基百科裏的解釋翻譯過來以下：

 

隊列提供了一種異步通訊協議，這意味着消息的發送者和接收者不須要同時與消息保持聯繫，發送者發送的消息會存儲在隊列中，直到接收者拿到它。

 

通常咱們把消息的發送者稱爲生產者，消息的接收者稱爲消費者；注意定義中的那兩個字「異步」，一般生產者的生產速度和消費者的消費速度是不相等的；若是兩個程序始終保持同步溝通，那勢必會有一方存在空等時間；若是兩個程序一持續運行的話，消費者的平均速度必定要大於生產者，否則隊列囤積會愈來愈多；固然，若是消費者沒有時效性需求的話，也能夠把消息囤積在隊列中，集中消費。

說到這裏，咱們再來談談隊列的分類，通常咱們根據生產者和消費者的不一樣，能夠把隊列分爲三類：

第一類是在一個應用程序內部（進程之間或者線程之間），相信你們學多線程時都寫過「生產者消費者」程序，生產者負責生產，將生產的結果放到緩衝區（如共享數組），消費者從緩衝區取出消費，在這裏，這個緩衝區就能夠稱爲「消息隊列」。

第二類其實也算在第一類的特例，就像咱們喜歡把操做系統和應用程序區別對待來看，操做系統要處理無數繁雜的事物，各進程、線程之間的數據交換少不了消息隊列的支持。

第三類是更爲通用意義上的「消息隊列」，這類隊列主要做用於不一樣應用，特別是跨機器、平臺，這令數據的交換更加普遍，通常一款獨立的隊列產品除了實現消息的傳遞外，還提供了相應的可靠性、事務、分佈式等特性，將生產者、消費者從中解耦。常見的消費隊列產品根據開源與否又可分爲兩類：

專有軟件：IBM WebSphere MQ，MSMQ…

開源軟件：ActiveMQ、RabbitMQ、Kafka…

2.2.JMS與AMQP

好了，對於上述第三類「消息隊列」，要在不一樣的機器中提供消息隊列的功能，那勢必要有統一的規範，這時候SUN就跳出來了，做爲跨平臺的JAVA勢必也要支持跨平臺的消息傳遞，基於此，SUN提供了一套消息標準：Java Message Service，縮寫JMS，可是這套規範定義的是API層面的標準，在JAVA體系中能夠很方便的交換，但對於其餘平臺就須要，可能須要消息隊列產品自己支持多協議（如OpenWire、STMOP）。

而AMQP定義的比JMS更加底層，從名字就能看出來（Advanced Message Queuing Protocol），它定義的是Wire-level的協議，自然具備跨平臺、跨語言的特性，基於此實現的消息隊列能夠與任何支持該協議的平臺交互。

一種是JAVA層面的API，一種是Wire-level協議，這是JMS和AMQP最本質的區別；同時兩種標準還有兩個比較明顯的差別：

一是消息傳遞模型；JMS比較簡單，支持兩種最通用的Peer-2-Peer、publisher/subscriber；通俗點就是點對點和廣播模式；而AMQP定義的更爲複雜，其定義了一種exchange&binding機制，由此支持五種模型：direct exchange、fanout exchange、topic exchange、headers exchange、system exchange，本質上與P2P、PUB/SUB同樣，可是更加細緻些。

二是支持的消息類型，JMS支持多種消息模型：TextMessage、MapMessage、BytesMessage、StreamMessage、ObjectMessage、Message等；而AMQP只有byte數組。

2.3.ActiveMQ

ActiveMQ是基於JMS實現的Provider（能夠理解爲隊列），它支持多種協議，如OpenWire，Stomp，AMQP等，基於此，支持多平臺；支持事務，支持分發策略、還有上面的多種消息模型。這裏咱們不細談ActiveMQ的各特性，咱們着重來看ActiveMQ的分佈式模型。

ActiveMQ支持分佈式，它支持Master-Slave提供高可用，也支持Broker-Cluster提供負載均衡，可是它的負載基於一種Forwarding Bridge機制。

 

在這種機制下，任意時刻一條消只會被一個broker持有，producer發送的消息，可能會通過多個broker轉發最終纔會到達consumer，能夠想象，當broker愈來愈多時，幾乎每次消費都要通過轉發，效率會明顯降低；而且在這種複雜邏輯下，任一broker的加入和移除都顯得十分複雜；這兩點是我不建議使用ActiveMQ分佈式集羣的根本緣由。

3.Kafka

好，咱們最後來談今天的主角Kafka，這個奇特的名字我始終沒有找到典故，也許是開發者暗戀女孩（基友）的名字吧^_^，Kafka由linkin開發，最初的目的是爲了應對linkin龐大的活動流數據（登陸、瀏覽、點擊、分享、喜歡等），這部分數據容量龐大，可是可靠性要求不高，故而經過犧牲一部分可靠性（這並非說咱們的數據會按百分比丟，咱們後面再談）來提高吞吐量；它砍掉了不少複雜的特性，如事務、分發策略、多種消息模型等；經過自身獨特的設計將消息持久化到磁盤上，以此同時支持在線和離線消費；而且其天生爲分佈式而設計，壓根就沒有單機模式（或者說單機模式是分佈式的特例），可以很好的擴展。實際應用中，Kafka能夠用來作消息隊列、流式處理（通常結合storm）、日誌聚合等。

3.1.架構

咱們先宏觀的看看Kafka的架構，Producer集羣經過zookeeper（實際中寫的是broker list）獲取所寫topic對應的partition列表，而後順序發送消息（支持本身實現分發策略），broker集羣負責消息的存儲和傳遞，支持Master Slaver模型，可分佈式擴展；Consumer集羣從zookeeper上獲取topic所在的partition列表，而後消費，一個partition只能被一個consumer消費。Name Server集羣（通常是zookeeper）提供名稱服務等協調信息。至於什麼是topic，什麼是partition，咱們接下來看。

3.2.Topic

Topic是生產者生產、消費者消費的隊列標識。一個Topic由一個或多個partition組成，每一個partition能夠單獨存在一個broker上，消費者能夠往任一partition發送消息，以此實現生產的分佈式，任一partition均可以被且只被一個消費者消息，以此實現消費的分佈式；所以partition的設計提供了分佈式的基礎。

同時，從上圖咱們也能發現這種設計還有一個優勢，由於每一個partition內的消息是有序的，而一個partition只能被一個消費者消費，所以Kafka能提供partition層面的消息有序，而傳統的隊列在多個consumer的狀況下是徹底沒法保證有序的。

3.3.消息傳遞模型

傳統的消息隊列最少提供兩種消息模型，一種P2P，一種PUB/SUB，而Kafka並無這麼作，巧妙的，它提供了一個消費者組的概念，一個消息能夠被多個消費者組消費，可是隻能被一個消費者組裏的一個消費者消費，這樣當只有一個消費者組時就等同與P2P模型，當存在多個消費者組時就是PUB/SUB模型。

3.4.消息持久化

不少系統、組件爲了提高效率通常巴不得把全部數據都扔到內存裏，而後按期flush到磁盤上；可實際上，現代操做系統也是這樣，全部的現代操做系統都樂於將空閒內存轉做磁盤緩存（頁面緩存），想不用都難；對於這樣的系統，他的數據在內存中保存了一份，同時也在OS的頁面緩存中保存了一份，這樣不但多了一個步驟還讓內存的使用率降低了一半；所以，Kafka決定直接使用頁面緩存；可是隨機寫入的效率很慢，爲了維護彼此的關係順序還須要額外的操做和存儲，而線性的寫入能夠避免這些，實際上，線性寫入（linear write）的速度大約是300MB/秒，但隨即寫入卻只有50k/秒，其中的差異接近10000倍。這樣，Kafka以頁面緩存爲中間的設計在保證效率的同時還提供了消息的持久化，每一個消費者本身維護當前讀取數據的offser（也可委託給zookeeper），以此可同時支持在線和離線的消費。

3.5.Push vs. Pull

對於消息的消費，ActiveMQ使用PUSH模型，而Kafka使用PULL模型，二者各有利弊，對於PUSH，broker很難控制數據發送給不一樣消費者的速度，而PULL能夠由消費者本身控制，可是PULL模型可能形成消費者在沒有消息的狀況下盲等，這種狀況下能夠經過long polling機制緩解，而對於幾乎每時每刻都有消息傳遞的流式系統，這種影響能夠忽略。

3.6.可靠性

剛剛說Kafka犧牲了一些可靠性來提高吞吐量，不少同窗可能擔憂消息的丟失，那麼咱們如今來看看各類狀況下的可靠性。

對於如上的模型，咱們分開來看，

先來看消息投遞可靠性，一個消息如何算投遞成功，Kafka提供了三種模式，第一種是啥都無論，發送出去就看成成功，這種狀況固然不能保證消息成功投遞到broker；第二種是對於Master Slave模型，只有當Master和全部Slave都接收到消息時，纔算投遞成功，這種模型提供了最高的投遞可靠性，可是損傷了性能；第三種模型，即只要Master確認收到消息就算投遞成功；實際使用時，根據應用特性選擇，絕大多數狀況下都會中和可靠性和性能選擇第三種模型。

咱們再來看消息在broker上的可靠性，由於消息會持久化到磁盤上，因此若是正常stop一個broker，其上的數據不會丟失；可是若是不正常stop，可能會使存在頁面緩存來不及寫入磁盤的消息丟失，這能夠經過配置flush頁面緩存的週期、閾值緩解，可是一樣會頻繁的寫磁盤會影響性能，又是一個選擇題，根據實際狀況配置。

接着，咱們再看消息消費的可靠性，Kafka提供的是「At least once」模型，由於消息的讀取進度由offset提供，offset能夠由消費者本身維護也能夠維護在zookeeper裏，可是當消息消費後consumer掛掉，offset沒有即時寫回，就有可能發生重複讀的狀況，這種狀況一樣能夠經過調整commit offset週期、閾值緩解，甚至消費者本身把消費和commit offset作成一個事務解決，可是若是你的應用不在意重複消費，那就乾脆不要解決，以換取最大的性能。

最後，咱們再來看zookeeper的可靠性，很明顯，他要掛了，一切都完了，地球就毀滅了，人類就滅絕了，星級穿越也挽救不了了……因此加強可靠性的方式就是把zookeeper也部署成集羣。

3.7.性能

好了，說了那麼多，咱們實際來測試下Kafka在各類狀況下的性能，爲了對比我也測了下單機模式下ActiveMQ的性能，不過因爲懶，沒有搭建ActiveMQ集羣進行測試，可是基於其噁心的Forwarding Bridge模型，我也持悲觀態度。

首先，測試環境以下：

Kafka：3 broker；8核/32G；默認配置

ActiveMQ：1 broker；8核/32G；默認配置

Producer: 一臺機器經過多線程模擬多producer；8核/32G；默認配置，異步發送

Consumer: 一臺機器經過多線程模擬多consumer；8核/32G；默認配置

除了特殊說明，生產和消費同時進行。

 

而後，我使用以下字符表示各類測試條件：

1T-1P3C-1P1C-1KW-1K：

1T:1個toipc

1P3C：1個partition 3個replication

1P1C：1個producer 1個consumer

1KW：1千萬條消息

1K：每一個消息1K

 

我先對ActiveMQ在單機多Producer、多consumer的狀況下的測試，結果比我想象中的好，官方的給出的一個數據是1-2K的數據，每秒10-20K個，這樣算下來大概30-40MB/S，而測試的結果在多線程的狀況下會更好些。

ActiveMQ-thread	Produce	Consume
1T-XXX-1P1C-1KW-1K	28.925MB/S	28.829MB/S
1T-XXX-3P3C-1KW-1K	43.711MB/S	41.791MB/S
1T-XXX-8P8C-1KW-1K	52.426MB/S	52.383MB/S

 

而後我又對Kafka進行了相應的測試，用一個partition模擬單機模式，結果和預想的同樣，在單機模型下，二者差別不大；而官方給的數聽說生產者能達到50MB/S，消費者能達到100MB/S，生產者符合官方數據，而消費者我始終沒有壓到那麼高的速度。

Kafka- thread	Produce	Consume
1T-1P1C-1P1C-1KW-1K	29.214MB/S	29.117MB/S
1T-1P1C-3P3C-1KW-1K	46.168MB/S	43.018MB/S
1T-1P1C-8P8C-1KW-1K	52.140MB/S	51.975MB/S

 

接下來的對於Kafka集羣，我想一樣數量的消息會不會由於topic數目的增多而影響，測試結果以下，代表topic越多，速度會有所降低，也符合預期。

Kafka-topic	Produce	Consume
1T-3P3C-3P3C-1.2KW-1K	49.255MB/S	49.204MB/S
3T-3P3C-3P3C-0.4KW*3-1K	46.239MB/S	45.774MB/S

 

而後爲了測試partition對性能的影響，進行了以下測試，能夠看到partition數量越多，總的生產和消費速度越快；可是意外的是Only produce狀況下生產效率沒有明顯提高反而略慢，這裏懷疑和page cache有關，沒有深刻研究。

Kafka-partition	Produce	Consume	Only Produce	Only Consume
1T-1P3C-1P1C-1KW-1K	29.213MB/S	29.117MB/S	28.941MB/S	34.360MB/S
1T-3P3C-3P3C-1KW-1K	47.103MB/S	46.966MB/S	46.540MB/S	66.219MB/S
1T-8P3C-8P8C-1KW-1K	61.522MB/S	61.412MB/S	60.703MB/S	72.701MB/S

 

綜上，咱們能夠看到Kafka的性能和吞吐是能夠擴展的。

3.8.風險點

對於咱們來講，Kafka主要有兩個風險點，第一，要深刻使用必需要熟讀源碼，而kafka源碼是用scala寫的，咱們並無相應的技術儲備，須要學習；第二，kafka技術較新，目前的版本是0.8.1.1，看起來還不太成熟。

4.KG應用

這一塊是在公司內部系統的應用，不適合對外，因此這裏刪去。

5.參考資料

Kafka-DOC：http://kafka.apache.org/documentation.html

ActiveMQ-DOC：http://activemq.apache.org

Understading the differences between AMQP & JMS：http://www.wmrichards.com/amqp.pdf

WIKI-MQ：http://en.wikipedia.org/wiki/Message_queue

WIKI-JMS：http://en.wikipedia.org/wiki/Java_Message_Service

WIKI-AMQP：http://en.wikipedia.org/wiki/Advanced_Message_Queuing_Protocol