楊冬暉

摘要：現(xiàn)如今，互聯(lián)網(wǎng)已經(jīng)滲透到日常生活的方方面面。生活的各個領(lǐng)域都迎來大數(shù)據(jù)的影響，數(shù)據(jù)傳輸過程的可靠性，以及如何有效地使用數(shù)據(jù)尤為重要。分布式消息系統(tǒng)可以有效地解決大規(guī)模分布式系統(tǒng)中消息傳遞問題。因此需要一個高吞吐量，高性能以及具有一定可靠性的分布式消息系統(tǒng)。Kafka是一個處理海量數(shù)據(jù)的分布式消息系統(tǒng)。[1]Kafka具有高效的數(shù)據(jù)傳輸速率，相對于其他的消息隊列系統(tǒng)具有較高的性能，采用發(fā)布/訂閱模式。該文主要總結(jié)介紹了kafka系統(tǒng)的架構(gòu)以及特征，重點介紹了分布式集群下kafka如何通過副本模式保證其消息的可靠傳輸。并通過實驗驗證其副本模式的可靠性，以支持后續(xù)相關(guān)研究。

關(guān)鍵詞：分布式消息系統(tǒng)；分布式系統(tǒng)；卡夫卡；可靠性

中圖分類號：TP311 文獻標(biāo)識碼：A 文章編號：1009-3044（2015）21-0075-02

An Research on the Reliability of a Distributed Message Queue

YANG Dong-hui

（School of Software Engineering， Tongji University， Shanghai 201804， China）

Abstract：Nowadays， the Internet has penetrated into every aspect of our daily lives. Every field of social life have the impact of big data transmission， and how to effectively use the data is particularly important. Distributed information system can effectively solve the problem of transmission information in large scale distributed systems. So we need a high throughput， high performance and distributed information system with a certain reliability. Kafka is a distributed information system with a massive data processing. Kafka has efficient data transfer rate， and compared with other message queuing the system has higher performance. Using the publish / subscribe model. This paper mainly introduces the Kafka system structure and characteristics， introduced the Kafka how to ensure reliable transmission of messages with replica. And ensure reliability by experiments， in order to support the follow-up study.

Key words：distributed messaging system； distributed system； Kafka； reliability

在現(xiàn)如今生活中，網(wǎng)上購物已成為日常生活的一部分，電商的發(fā)展規(guī)模越來越龐大。大型互聯(lián)網(wǎng)電商公司的業(yè)務(wù)越來越復(fù)雜，不同業(yè)務(wù)需要不同的子系統(tǒng)進行支持，每個子系統(tǒng)之間的通信需要依靠消息系統(tǒng)支持，這就進一步加大了對消息系統(tǒng)的依賴。當(dāng)消息系統(tǒng)出現(xiàn)故障，消息不能按時傳遞、消息丟失等等都會造成不可估量的后果，公司可能造成重大損失。因此分布式系統(tǒng)中的消息通信系統(tǒng)的可靠性尤為重要。

分布式系統(tǒng)[2，11]是指分散的物理機通過互聯(lián)網(wǎng)連接建立起的一套軟件系統(tǒng)，分布式系統(tǒng)具有高度的內(nèi)聚性和透明性。分布式系統(tǒng)中數(shù)據(jù)的傳輸需要面臨比傳統(tǒng)系統(tǒng)更多的挑戰(zhàn)。如何保證分布式系統(tǒng)中大量，高速以及可靠的數(shù)據(jù)傳輸是分布式消息隊列需要解決的問題。消息隊列是操作系統(tǒng)的進程之間用于通信的一種機制，兩個或多個進程間通過訪問共同的消息隊列完成消息的交換。在分布式環(huán)境下分布式的消息隊列能在客戶端和服務(wù)端提供同步和異步的連接，實現(xiàn)應(yīng)用程序之間的協(xié)同。

1 KAFKA可靠性研究

1.1 相關(guān)背景

在分布式消息隊列[3]發(fā)展初期，消息傳遞通常采用點對點的傳輸結(jié)構(gòu)，發(fā)送方需要事先指明接收方的地址，雖然消息的接收方和發(fā)送方是松耦合連接的，不必保持相互之間通信的同步。但是由于消息在傳遞過程中需要綁定接收方的地址，系統(tǒng)不夠靈活難以擴展。由于以上的原因分布式消息系統(tǒng)向發(fā)布/訂閱[4]模式轉(zhuǎn)變，發(fā)布/訂閱模式具有異步松耦合和多對多通信的特點，此模式適應(yīng)目前大多數(shù)企業(yè)分布式計算環(huán)境的要求。發(fā)布/訂閱模式下發(fā)送消息的一方稱為發(fā)布者，接收消息的一方稱為訂閱者。消息不再被發(fā)送到指定的接收者中而是轉(zhuǎn)而發(fā)送給一個中間的消息代理服務(wù)器。訂閱者只需去服務(wù)器中接收自己感興趣的消息，發(fā)布者不需要知道何種訂閱者接收了發(fā)布的消息。這種發(fā)布/訂閱的模式更易于擴展。常見的發(fā)布/訂閱消息隊列包括Kafka[5]、RabbitMQ[6]、ActiveMQ[7]以及Microsoft MSMQ等。其中RabbitMQ是使用Erlang語言編寫的一個開源的消息隊列，是AMQP協(xié)議的一個實現(xiàn)。它實現(xiàn)了代理（Broker）架構(gòu)，使得消息在發(fā)送到客戶端之前可以在中央節(jié)點上排隊。

1.2 Kafka結(jié)構(gòu)

Kafka由Linkedin公司開發(fā)，并使用Scala語言編寫。Kafka是一個分布式的、分區(qū)的、多副本的、多訂閱者的日志提交系統(tǒng)[9]。Kafka系統(tǒng)包括消息的生產(chǎn)者（producer），消息的消費者（consumer），消息代理者（broker）和管理者（zookeeper） [8]四個部分。生產(chǎn)者生產(chǎn)的消息被放在topic中。一個topic中可以設(shè)置多個partition，每一個partition可以對應(yīng)一個消費者被消費。當(dāng)設(shè)置了多個partition，生產(chǎn)者在生產(chǎn)消息的時，需指定對應(yīng)的partition，如果沒有指定，則會使用默認(rèn)值。每個partition是一個有序的、不可變的消息序列，這個序列可以被連續(xù)地追加。每條消息會有一個序列號，在文件中的位置被稱為offset（偏移量），offset為一個long型數(shù)字，唯一標(biāo)識一條消息。生產(chǎn)者生產(chǎn)消息，產(chǎn)生的消息發(fā)送到broker，等待消費者的接收。消息在broker中根據(jù)topic和partition來區(qū)分不同。消費者根據(jù)topic和partition的值從broker中訂閱消息。消費者在想要訂閱消息時向broker端發(fā)送請求，告知其topic和partition值來獲取消息。消費者可以決定從partition何處開始消費，并通過重置offset的值來重新消費已經(jīng)消費的消息。在Kafka中消費者在和broker建立連接后，主動拉取消息。生產(chǎn)者和消費者之間消息的傳遞可以用圖1所示[9]。

管理者zookeeper在其中進行協(xié)調(diào)控制，管理broker和consumer的動態(tài)加入和離開，維護了生產(chǎn)者和消費者之間的關(guān)系和topic中的信息，并提供一定的負(fù)載均衡支持。Kafka server為生產(chǎn)者消費者提供服務(wù)，當(dāng)生產(chǎn)者和消費者運行時，都需要配置相應(yīng)的zookeeper信息。Zookeeper服務(wù)器主要功能為與kafka服務(wù)器進行交互，zookeeper服務(wù)器對kafka服務(wù)器進行簡單的管理。每個broker啟動后會在zookeeper上臨時注冊，注冊信息包含broker的ip地址和端口號，broker上存在的topic和此topic的partition。在kafka中producer可以將生產(chǎn)的消息指定發(fā)送到某一個具體的partition。相對應(yīng)的對于consumer來說，每一個consumer屬于一個消費者集群，每個group中的consumer消費消息相互獨立。

1.3 Kafka可靠性研究

Kafka消息隊列為分布式系統(tǒng)提供一定的可靠性保證，針對生產(chǎn)者生產(chǎn)的消息，在集群模式下，kafka可以將每一個partition中的數(shù)據(jù)復(fù)制到多個kafka server中。當(dāng)復(fù)制為多個副本時，每個partition會由zookeeper指定一個leader節(jié)點和多個follower節(jié)點（當(dāng)副本數(shù)多于2個）。備份的個數(shù)可以通過修改broker配置文件或者手動配置更改。Follower節(jié)點需要和leader節(jié)點保持同步。其中l(wèi)eader節(jié)點負(fù)責(zé)處理讀寫請求，leader節(jié)點還需要監(jiān)控它所控制的所有follower節(jié)點的運行狀態(tài)，當(dāng)follower節(jié)點與leader節(jié)點嚴(yán)重不一致或者節(jié)點失效，leader節(jié)點會刪除此follower節(jié)點。當(dāng)leader節(jié)點失效時，通過選舉將在存活的follower節(jié)點中重新選舉出新的leader節(jié)點，新的leader節(jié)點替代失效leader節(jié)點繼續(xù)負(fù)責(zé)監(jiān)控整個集群運行。因此當(dāng)分布式集群中某一個業(yè)務(wù)節(jié)點失效時，只要還有一個節(jié)點存活，即還有一個partition的副本。此消息都可以進行正常發(fā)送和接收[10]。

Kafka集群的可靠性通過partition的多副本方式得到提高。在kafka集群的運行過程中，每一個kafka server上可以運行多個生產(chǎn)者或多個消費者，kafka server啟動時會在zookeeper上注冊相應(yīng)的信息。對相應(yīng)的partition保存了多個副本時，當(dāng)其中某個kafka server發(fā)生故障，zookeeper會將失效的kafka server部署到集群中處于存活狀態(tài)的kafka server中去完成此操作后，分布式集群中消息通信會通過備用的kafka server進行傳遞，由此保證消息通信服務(wù)不受影響。

Kafka集群的建立使得partition可以復(fù)制為多個副本，當(dāng)正在使用的kafka server因故障而停止工作時，可以使用副本繼續(xù)進行生產(chǎn)者消費者的消息之間的傳遞，不需要重啟kafka server，保證了進程的正常運行。如圖2所示，當(dāng)kafka服務(wù)開始時，生產(chǎn)者與消費者進程正常運行，經(jīng)過broker進行消息的傳遞，生產(chǎn)者將消息推送給已訂閱的消費者。Zookeeper server對kafka server進行管理，本圖包含兩個kafka server所以topic最多保存了兩個副本。如圖中所示，正在提供服務(wù)的kafka server出現(xiàn)故障，zookeeper server選舉新的leader節(jié)點并調(diào)用其副本繼續(xù)提供服務(wù)。生產(chǎn)者和消費者正常的消息傳遞會通過副本繼續(xù)進行，消息不會丟失，保證了分布式集群中消息的可靠傳遞。

2 實驗分析

通過實驗驗證分析kafka分布式集群的可靠性。實驗環(huán)境為三臺kafka服務(wù)器與一臺zookeeper服務(wù)器。實驗過程為：

1）啟動zookeeper服務(wù)與3臺kafka服務(wù)；

2）創(chuàng)建1個生產(chǎn)者與1個消費者，并創(chuàng)建1個topic；

3）針對創(chuàng)建的topic將其設(shè)置為3個副本模式，并觀察哪一個節(jié)點為leader節(jié)點，哪兩個節(jié)點為follower節(jié)點；

4）生產(chǎn)者發(fā)送信息，手動終止leader節(jié)點kafka服務(wù)，觀察消費者有無正常接收消息；

5）手動終止還存活的kafka服務(wù)節(jié)點中的一個，觀察消費者是否可以正常接收消息；

6）手動終止最后一個存活的kafka服務(wù)節(jié)點，觀察消費者是否可以正常接收消息。

通過上述實驗，得到實驗結(jié)果如下：當(dāng)?shù)谝淮谓K止leader節(jié)點后，zookeeper會重新選舉出新的leader節(jié)點，生產(chǎn)者發(fā)布消息后，訂閱此topic的消費者仍可以正常的接收消息。當(dāng)終止的 服務(wù)節(jié)點不為leader節(jié)點時，leader節(jié)點不會重新選舉，消費者仍可接收訂閱的消息。當(dāng)終止所有kafka服務(wù)節(jié)點時，生產(chǎn)者生產(chǎn)的消息無法發(fā)布到broker中，消費者也無法從broker中訂閱消息。

通過上述實驗結(jié)果，可以得出kafka分布式集群通過多副本，選取leader節(jié)點對多個副本進行管理的形式，保證集群消息傳遞具有一定的可靠性。

3 結(jié)束語

分布式消息隊列在如今互聯(lián)網(wǎng)企業(yè)得到越來越多的應(yīng)用。Kafka為分布式集群中消息傳遞提供了一定的可靠性保證。本文分析了kafka的基本結(jié)構(gòu)，并對其副本模式進行了分析。實驗驗證了其副本模式下生產(chǎn)者消費者消息的傳遞是否能容忍一定的故障。為后續(xù)深入研究kafka提供前期支持。

參考文獻：

[1] Goodhope K K G， Goodhope K. Building LinkedIn's Real-time Activity Data Pipeline[J]. Bulletin of the Technical Committee on Data Engineering， 2012（2）.

[2] 胡華平， 金士堯， 王召福. 分布式系統(tǒng)的可信性研究[J]. 計算機工程與科學(xué)， 1998（1）： 48-53.

[3] Patel D， Khasib F， Sadooghi I， et al. Towards In-Order and Exactly-Once Delivery Using Hierarchical Distributed Message Queues[C]//IEEE/ACM International Symposium on Cluster， Cloud and Grid Computing. IEEE， 2014：883-892.

[4] 馬建剛， 黃濤， 汪錦嶺， 等. 面向大規(guī)模分布式計算發(fā)布訂閱系統(tǒng)核心技術(shù)[J]. 軟件學(xué)報， 2006， 17（1）： 134-147.

[5] Jay Kreps， Neha Narkhede， Jun Rao. Kafka： a Distributed Messaging System for Log Processing[Z].2011.

[6] 袁佳， 郭燕慧. 基于rabbitmq的海量日志的分布式處理[J]. 軟件， 2013（7）： 19-23.

[7] 戴俊， 朱曉民. 基于 ActiveMQ 的異步消息總線的設(shè)計與實現(xiàn)[J]. 計算機系統(tǒng)應(yīng)用， 2010（8）： 254-257.

[8] Okorafor E， Patrick M K. Availability of Jobtracker machine in hadoop/mapreduce zookeeper coordinated clusters[J]. Advanced Computing： An International Journal （ACIJ）， 2012， 3（3）： 19-30.

[9] Apacher Kafka[EB/OL]. http：//kafka.apache.org/.

[10] Chen J， Arumaithurai M， Fu X， et al. Reliable publish/subscribe in content-centric networks[C]//Proceedings of the 3rd ACM SIGCOMM workshop on Information-centric networking. ACM， 2013： 21-26.

[11] 馬浩然. 基于NS3的分布式消息系統(tǒng)Kafka的仿真實現(xiàn)[J]. 軟件， 2015（1）.