分布式消息系統(tǒng)kafka該怎么理解

今天就跟大家聊聊有關分布式消息系統(tǒng)kafka該怎么理解，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結了以下內容，希望大家根據(jù)這篇文章可以有所收獲。

kafka：一個分布式消息系統(tǒng)

1.背景

最近因為工作需要，調研了追求高吞吐的輕量級消息系統(tǒng)Kafka，打算替換掉線上運行的ActiveMQ，主要是因為明年的預算日流量有十億，而ActiveMQ的分布式實現(xiàn)的很奇怪，所以希望找一個適合分布式的消息系統(tǒng)。

以下是內容是調研過程中總結的一些知識和經(jīng)驗，歡迎拍磚。

2.基礎知識

2.1.什么是消息隊列

首先，我們來看看什么是消息隊列，維基百科里的解釋翻譯過來如下：

隊列提供了一種異步通信協(xié)議，這意味著消息的發(fā)送者和接收者不需要同時與消息保持聯(lián)系，發(fā)送者發(fā)送的消息會存儲在隊列中，直到接收者拿到它。

一般我們把消息的發(fā)送者稱為生產(chǎn)者，消息的接收者稱為消費者；注意定義中的那兩個字“異步”，通常生產(chǎn)者的生產(chǎn)速度和消費者的消費速度是不相等的；如果兩個程序始終保持同步溝通，那勢必會有一方存在空等時間；如果兩個程序一持續(xù)運行的話，消費者的平均速度一定要大于生產(chǎn)者，不然隊列囤積會越來越多；當然，如果消費者沒有時效性需求的話，也可以把消息囤積在隊列中，集中消費。

說到這里，我們再來談談隊列的分類，一般我們根據(jù)生產(chǎn)者和消費者的不同，可以把隊列分為三類：

第一類是在一個應用程序內部（進程之間或者線程之間），相信大家學多線程時都寫過“生產(chǎn)者消費者”程序，生產(chǎn)者負責生產(chǎn)，將生產(chǎn)的結果放到緩沖區(qū)（如共享數(shù)組），消費者從緩沖區(qū)取出消費，在這里，這個緩沖區(qū)就可以稱為“消息隊列”。

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

第三類是更為通用意義上的“消息隊列”，這類隊列主要作用于不同應用，特別是跨機器、平臺，這令數(shù)據(jù)的交換更加廣泛，一般一款獨立的隊列產(chǎn)品除了實現(xiàn)消息的傳遞外，還提供了相應的可靠性、事務、分布式等特性，將生產(chǎn)者、消費者從中解耦。常見的消費隊列產(chǎn)品根據(jù)開源與否又可分為兩類：

專有軟件：IBM WebSphere MQ，MSMQ…

開源軟件：ActiveMQ、RabbitMQ、Kafka…

2.2.JMS與AMQP

好了，對于上述第三類“消息隊列”，要在不同的機器中提供消息隊列的功能，那勢必要有統(tǒng)一的規(guī)范，這時候SUN就跳出來了，作為跨平臺的JAVA勢必也要支持跨平臺的消息傳遞，基于此，SUN提供了一套消息標準：Java Message Service，縮寫JMS，但是這套規(guī)范定義的是API層面的標準，在JAVA體系中可以很方便的交換，但對于其他平臺就需要，可能需要消息隊列產(chǎn)品本身支持多協(xié)議（如OpenWire、STMOP）。

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

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

一是消息傳遞模型；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數(shù)組。

2.3.ActiveMQ

ActiveMQ是基于JMS實現(xiàn)的Provider（可以理解為隊列），它支持多種協(xié)議，如OpenWire，Stomp，AMQP等，基于此，支持多平臺；支持事務，支持分發(fā)策略、還有上面的多種消息模型。這里我們不細談ActiveMQ的各特性，我們著重來看ActiveMQ的分布式模型。

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

在這種機制下，任意時刻一條消只會被一個broker持有，producer發(fā)送的消息，可能會經(jīng)過多個broker轉發(fā)最終才會到達consumer，可以想象，當broker越來越多時，幾乎每次消費都要經(jīng)過轉發(fā)，效率會明顯下降；并且在這種復雜邏輯下，任一broker的加入和移除都顯得十分復雜；這兩點是我不建議使用ActiveMQ分布式集群的根本原因。

1

3.Kafka

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

3.1.架構

2

我們先宏觀的看看Kafka的架構，Producer集群通過zookeeper（實際中寫的是broker list）獲取所寫topic對應的partition列表，然后順序發(fā)送消息（支持自己實現(xiàn)分發(fā)策略），broker集群負責消息的存儲和傳遞，支持Master Slaver模型，可分布式擴展；Consumer集群從zookeeper上獲取topic所在的partition列表，然后消費，一個partition只能被一個consumer消費。Name Server集群（一般是zookeeper）提供名稱服務等協(xié)調信息。至于什么是topic，什么是partition，我們接下來看。

3.2.Topic

Topic是生產(chǎn)者生產(chǎn)、消費者消費的隊列標識。一個Topic由一個或多個partition組成，每個partition可以單獨存在一個broker上，消費者可以往任一partition發(fā)送消息，以此實現(xiàn)生產(chǎn)的分布式，任一partition都可以被且只被一個消費者消息，以此實現(xiàn)消費的分布式；因此partition的設計提供了分布式的基礎。

3

同時，從上圖我們也能發(fā)現(xiàn)這種設計還有一個優(yōu)點，因為每個partition內的消息是有序的，而一個partition只能被一個消費者消費，因此Kafka能提供partition層面的消息有序，而傳統(tǒng)的隊列在多個consumer的情況下是完全無法保證有序的。

3.3.消息傳遞模型

傳統(tǒng)的消息隊列最少提供兩種消息模型，一種P2P，一種PUB/SUB，而Kafka并沒有這么做，巧妙的，它提供了一個消費者組的概念，一個消息可以被多個消費者組消費，但是只能被一個消費者組里的一個消費者消費，這樣當只有一個消費者組時就等同與P2P模型，當存在多個消費者組時就是PUB/SUB模型。

4

3.4.消息持久化

很多系統(tǒng)、組件為了提升效率一般恨不得把所有數(shù)據(jù)都扔到內存里，然后定期flush到磁盤上；可實際上，現(xiàn)代操作系統(tǒng)也是這樣，所有的現(xiàn)代操作系統(tǒng)都樂于將空閑內存轉作磁盤緩存（頁面緩存），想不用都難；對于這樣的系統(tǒng)，他的數(shù)據(jù)在內存中保存了一份，同時也在OS的頁面緩存中保存了一份，這樣不但多了一個步驟還讓內存的使用率下降了一半；因此，Kafka決定直接使用頁面緩存；但是隨機寫入的效率很慢，為了維護彼此的關系順序還需要額外的操作和存儲，而線性的寫入可以避免這些，實際上，線性寫入（linear write）的速度大約是300MB/秒，但隨即寫入?yún)s只有50k/秒，其中的差別接近10000倍。這樣，Kafka以頁面緩存為中間的設計在保證效率的同時還提供了消息的持久化，每個消費者自己維護當前讀取數(shù)據(jù)的offser（也可委托給zookeeper），以此可同時支持在線和離線的消費。

3.5.Push vs. Pull

對于消息的消費，ActiveMQ使用PUSH模型，而Kafka使用PULL模型，兩者各有利弊，對于PUSH，broker很難控制數(shù)據(jù)發(fā)送給不同消費者的速度，而PULL可以由消費者自己控制，但是PULL模型可能造成消費者在沒有消息的情況下盲等，這種情況下可以通過long polling機制緩解，而對于幾乎每時每刻都有消息傳遞的流式系統(tǒng)，這種影響可以忽略。

3.6.可靠性

剛剛說Kafka犧牲了一些可靠性來提升吞吐量，很多同學可能擔心消息的丟失，那么我們現(xiàn)在來看看各種情況下的可靠性。

5

對于如上的模型，我們分開來看，

先來看消息投遞可靠性，一個消息如何算投遞成功，Kafka提供了三種模式，第一種是啥都不管，發(fā)送出去就當作成功，這種情況當然不能保證消息成功投遞到broker；第二種是對于Master Slave模型，只有當Master和所有Slave都接收到消息時，才算投遞成功，這種模型提供了最高的投遞可靠性，但是損傷了性能；第三種模型，即只要Master確認收到消息就算投遞成功；實際使用時，根據(jù)應用特性選擇，絕大多數(shù)情況下都會中和可靠性和性能選擇第三種模型。

我們再來看消息在broker上的可靠性，因為消息會持久化到磁盤上，所以如果正常stop一個broker，其上的數(shù)據(jù)不會丟失；但是如果不正常stop，可能會使存在頁面緩存來不及寫入磁盤的消息丟失，這可以通過配置flush頁面緩存的周期、閾值緩解，但是同樣會頻繁的寫磁盤會影響性能，又是一個選擇題，根據(jù)實際情況配置。

接著，我們再看消息消費的可靠性，Kafka提供的是“At least once”模型，因為消息的讀取進度由offset提供，offset可以由消費者自己維護也可以維護在zookeeper里，但是當消息消費后consumer掛掉，offset沒有即時寫回，就有可能發(fā)生重復讀的情況，這種情況同樣可以通過調整commit offset周期、閾值緩解，甚至消費者自己把消費和commit offset做成一個事務解決，但是如果你的應用不在乎重復消費，那就干脆不要解決，以換取最大的性能。

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

3.7.性能

好了，說了那么多，我們實際來測試下Kafka在各種情況下的性能，為了對比我也測了下單機模式下ActiveMQ的性能，不過由于懶，沒有搭建ActiveMQ集群進行測試，但是基于其惡心的Forwarding Bridge模型，我也持悲觀態(tài)度。

首先，測試環(huán)境如下：

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

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

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

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

除了特殊說明，生產(chǎn)和消費同時進行。

然后，我使用如下字符表示各種測試條件：

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的情況下的測試，結果比我想象中的好，官方的給出的一個數(shù)據(jù)是1-2K的數(shù)據(jù)，每秒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模擬單機模式，結果和預想的一樣，在單機模型下，兩者差異不大；而官方給的數(shù)據(jù)說生產(chǎn)者能達到50MB/S，消費者能達到100MB/S，生產(chǎn)者符合官方數(shù)據(jù)，而消費者我始終沒有壓到那么高的速度。

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集群，我想同樣數(shù)量的消息會不會因為topic數(shù)目的增多而影響，測試結果如下，表明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數(shù)量越多，總的生產(chǎn)和消費速度越快；但是意外的是Only produce情況下生產(chǎn)效率沒有明顯提升反而略慢，這里懷疑和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，看起來還不太成熟。

看完上述內容，你們對分布式消息系統(tǒng)kafka該怎么理解有進一步的了解嗎？如果還想了解更多知識或者相關內容，請關注億速云行業(yè)資訊頻道，感謝大家的支持。