一種分布式消息隊列研究與測試

于金良　朱志祥　李聰穎

摘 要：為了解決實時流式數(shù)據(jù)的采集問題，研究了一種分布式消息隊列Kafka，它可以實時采集流式數(shù)據(jù)，處理數(shù)據(jù)時先從它訂閱，就可以以流數(shù)據(jù)的形式處理數(shù)據(jù)。該隊列具有部署簡單、易于管理、吞吐量高、高容錯性等優(yōu)點。經測試，該隊列可以滿足實際生產中對吞吐量的需求。

關鍵詞：分布式；消息隊列；主題；流數(shù)據(jù)

中圖分類號：TP274.2 文獻標識碼：A 文章編號：2095-1302（2016）08-00-03

0 引 言

當前大數(shù)據(jù)處理的數(shù)據(jù)形式主要分為兩種。一種是固定的批量數(shù)據(jù)，這種數(shù)據(jù)是一個文件或者一個數(shù)據(jù)庫，其數(shù)據(jù)量是固定的，我們只需一次讀取，然后進行計算；另外一種是活躍的流式數(shù)據(jù)，這種數(shù)據(jù)是一個數(shù)據(jù)流，是實時生成、可傳輸?shù)模缫粋€網站的流量page views、用戶搜索的內容等，這些數(shù)據(jù)是實時的，且數(shù)據(jù)量很大，其采集比較困難，傳統(tǒng)的消息隊列很難應用到此種場景中。要想處理這些數(shù)據(jù)就需要先采集這些數(shù)據(jù)，本文介紹了一種可以實時采集這些數(shù)據(jù)的分布式消息隊列Kafka。

1 簡介

Kafka是LinkedIn于2010年12月份開源的消息系統(tǒng)，它主要用于處理活躍的流式數(shù)據(jù)。活躍的流式數(shù)據(jù)在Web網站的應用中十分常見，包括網站的pv、用戶訪問的內容、搜索的內容等。這些數(shù)據(jù)通常以日志的形式記錄下來，然后每隔一段時間進行一次統(tǒng)計處理。

傳統(tǒng)的日志分析系統(tǒng)提供了一種離線處理日志信息的可擴展方案，但若要進行實時處理，通常會有較大延遲。而現(xiàn)有的消息（隊列）系統(tǒng)能夠很好地處理實時或者近似實時的應用，但未處理的數(shù)據(jù)通常不會寫到磁盤上，這對于Hadoop之類（一小時或者一天只處理一部分數(shù)據(jù)）的離線應用而言，可能會存在些許問題。Kafka正是為了解決以上問題而設計的，它能夠很好地提供離線和在線應用。

Kafka對消息按topic進行歸類保存，消息的發(fā)布者稱為生產者（Producer），消息的接收者稱為消費者（Consumer）。Kafka是分布式，它的集群由多個Kafka實例組成，每個實例是一個Broker。Kafka集群的信息及生產者與消費者的元數(shù)據(jù)都由Zookeeper保存，它本身無需保存這些數(shù)據(jù)。

2 設計原理

Kafka的設計初衷是希望作為一個統(tǒng)一的數(shù)據(jù)收集平臺，能夠實時收集數(shù)據(jù)、支撐大數(shù)據(jù)，并具備良好的容錯能力。

2.1 存儲

Kafka不會在消息被消費后直接刪除，而是將消息持久化在磁盤中，使用文件存儲消息，而文件系統(tǒng)的優(yōu)化幾乎是不可能的，為了提高性能，采用了緩存/直接內存映射的方法。為了減少對磁盤的訪問次數(shù)，Broker將數(shù)據(jù)暫時緩存起來，當消息的數(shù)量達到一定值時，再flush到磁盤中，減少了在磁盤I/O上消耗的時間。

2.2 高吞吐量

為了提高Kafka的吞吐量，它采用了批量傳輸發(fā)送的方法，即生產者發(fā)布消息時，先將消息緩存，當達到一定量時，批量發(fā)送到Broker；對于消費者也一樣，Broker會批量發(fā)送多條消息。考慮到網絡I/O，Kafka將在網絡上傳輸?shù)臄?shù)據(jù)進行壓縮，它支持的壓縮方式有gzip/snappy等。而在創(chuàng)建主題時，可以分為多個分區(qū)，進一步提高讀寫的吞吐量。

2.3 負載均衡

生產者根據(jù)用戶指定的算法，將消息發(fā)送到指定的分區(qū)；存在多個分區(qū)，每個分區(qū)都有自己的副本，每個副本分布在不同的Broker節(jié)點上；多個分區(qū)需要選取出主分區(qū)，主分區(qū)負責讀寫，并由Zookeeper負責故障恢復；通過Zookeeper管理Broker與消費者的動態(tài)加入與離開。

2.4 自動擴容

由于在大數(shù)據(jù)行業(yè)中數(shù)據(jù)量的大小難以估計，Kafka支持集群的橫向擴展，當需要增加Broker結點時，新增的Broker、生產者、消費者會向Zookeeper注冊，并及時作出調整。

3 技術架構

Kafka是使用scala語言開發(fā)的，同時支持多種編程語言的客戶端（c++、java、python、go等），其總體架構如圖1所示。

Kafka的消息分為以下幾個層次：

（1）主題（Topic）：一類消息，例如頁面瀏覽日志，用戶搜索日志等都可以以主題的形式存在，Kafka集群能夠同時負責一個或多個主題的分發(fā)。

（2）分區(qū)（Partition）：是在主題物理上的分組，一個主題可以分為多個分區(qū)，每個分區(qū)是一個有序的隊列。分區(qū)中的每條消息都會被分配一個有序的id（offset）。

（3） 消息（Message）：最小訂閱單元。

具體流程如下所示：

（1）生產者根據(jù)指定的分區(qū)方法（round-robin、hash等），將消息發(fā)布到指定主題的分區(qū)中；

（2）Kafka集群接收到生產者發(fā)過來的消息后，將其持久化到硬盤，并保留消息指定時長（可配置），而不關注消息是否被消費；

（3）消費者從Kafka集群拉數(shù)據(jù)，并控制獲取消息的offset。

4 主題與分區(qū)

一個topic是對一組消息的歸納。Kafka對每個主題的日志進行了分區(qū)，如圖2所示。

每個分區(qū)都由一系列有序的、不可變的消息組成，這些消息被連續(xù)追加到分區(qū)中。分區(qū)中的每個消息都有一個連續(xù)的序列號叫做offset，用來在分區(qū)中唯一的標識這個消息。

在一個可配置的時間段內，Kafka集群保留所有發(fā)布的消息，不管這些消息有沒有被消費。比如將消息的保存策略設置為2天，那么在一個消息發(fā)布到Kafka的兩天時間內，它都可以被消費。之后它將被丟棄以釋放空間。Kafka的性能是和數(shù)據(jù)量無關的常量級，所以保留太多的數(shù)據(jù)并不是問題。

實際上每個消費者唯一需要維護的是已消費消息在消息隊列中的位置，即offset。一般情況下隨著消費者不斷的讀取消息，offset的值隨之不斷增加，其實消費者可以以任意的順序讀取消息，比如它可以將offset設置成一個舊值來重讀之前的消息。

結合以上特點可以發(fā)現(xiàn)，Kafka消費者是輕量級的，它們可以在不對集群和其他消費者造成影響的情況下讀取消息。

將日志分區(qū)可以達到以下目的：首先這使得每個日志的數(shù)量不會太大，可以在單個服務上保存。另外每個分區(qū)可以單獨發(fā)布和消費，為并發(fā)操作主題提供了一種可能。

4.1 分布式

每個分區(qū)在Kafka集群的若干服務中都有副本，這些持有副本的服務可以共同處理數(shù)據(jù)和請求，副本數(shù)量可以配置。副本使Kafka具備了容錯能力。

每個分區(qū)都由一個服務器作為主服務，零或若干服務器作為從服務，主服務負責處理消息的讀和寫，從服務則復制主服務，若主服務宕機了，從服務中的一臺則會自動成為主服務。集群中的每個服務都會同時扮演兩個角色：作為它所持有的一部分分區(qū)的主，同時作為其他分區(qū)的從，這樣集群就會有較好的負載均衡。

4.2 生產者

生產者（Producer）將消息發(fā)布到它指定的主題中，并決定發(fā)布到哪個分區(qū)中。一般是由負載均衡機制隨機選擇一個分區(qū)，也可通過特定的分區(qū)函數(shù)來選擇分區(qū)。使用更多的是第二種。

4.3 消費者

發(fā)布消息通常有兩種模式：隊列模式（queuing）和發(fā)布-訂閱模式（publish-subscribe）。隊列模式中，消費者（Consumers）可以同時從服務端讀取消息，每個消息只被其中一個消費者讀到；發(fā)布-訂閱模式中消息被廣播到所有的消費者中。多個消費者可以加入到一個消費者組中，共同競爭一個主題，主題中的消息將被分發(fā)到組中的一個成員中。同一組中的消費者可以在不同的程序中、不同的機器上。如果所有的消費者都在一個組中，這就成為了傳統(tǒng)的隊列模式，在消費者組中實現(xiàn)負載均衡。

5 性能測試

通過一臺4核、8 G內存的臺式機，向一個擁有5臺Broker的Kafka集群發(fā)布、訂閱消息。消息的平均大小為100個字節(jié)。測試結果如下：

生產者：每秒可發(fā)布30萬條消息，且可通過調整request.required.acks參數(shù)來保證數(shù)據(jù)的可靠性。

消費者：每秒可消費5萬條數(shù)據(jù)。通過使用java語言編寫Kafka生產者與消費者對Kafka性能進行測試。

6 結 語

分布式消息隊列Kafka具有部署簡單、易于管理、高吞吐量、高容錯性等優(yōu)點，經測試，可以滿足實際生產中對吞吐量的需求。

參考文獻

[1]孫韓林.一種基于云計算的網絡流量分析系統(tǒng)結構[J].西安郵電大學學報，2013，18（4）：75-79.

[2]金澈清，錢衛(wèi)寧，周傲英.流數(shù)據(jù)分析與管理綜述[J].軟件學報，2004，15（8）：1172-1181.

[3]曹婧華，冉彥中，許志軍.分布式消息隊列的設計與實現(xiàn)[J].河南科技大學學報（自然科學版），2010，31（4）：35-38，109.

[4]王博，陳莉君.Hadoop遠程過程調用機制的分析和應用[J].西安郵電學院學報，2012，17（6）：74-77.

[5]盧本捷.分布式消息隊列的理論、實現(xiàn)與應用[D].武漢：華中科技大學，2004.

[6]崔小燕.Linux集群系統(tǒng)分析[J].西安郵電學院學報，2006，11（5）：103-106.

[7]于自強.海量流數(shù)據(jù)挖掘相關問題研究[D].濟南：山東大學，2015.

[8]謝曉燕，張靜雯.一種基于Linux集群技術的負載均衡方法[J].西安郵電大學學報，2014，19（3）：64-68.

[9]陸慶，周世杰，秦志光，等.消息隊列中間件系統(tǒng)中消息隊列與消息分發(fā)技術研究[J].計算機應用研究，2003（8）：51-53.