作者：蹇智華 達觀數據

前言

達觀數據作為一家提供大數據服務的公司，經常會遇到客戶上報數據的需求。這樣的請求不需要馬上返回處理結果， 而是需要后臺將一系列的上報數據進行統一歸檔整理挖掘， 然后將結果數據呈現給客戶。這樣的業務需求需要達觀提供數據暫存服務，也就是說我們需要一個系統在生產者（客戶上報數據）和消費者（后臺數據處理）之間進行溝通，簡而言之叫系統間通信消息系統，這種模型就是經典的生產者（producer）、消費者（consumer）模型。

然而有一個消息系統正好是為了應對這種業務場景而生，它就是kafka。那么kafka到底是一個什么樣的系統？有什么特點？實際吞吐表現又如何？帶著這些問題，我們一起來了解一下。

一, Kafka簡介

首先根據官網介紹，知道kafka是一個分布式流處理平臺，一個可處理企業級發布/訂閱的消息系統，并且具有高容錯性和消費及時性等特點，那么它是怎么做到這一點的呢？接著往下看。

1，主題和日志：

主題(topic)和日志(log)設置是kafka一大特色，一個kafka集群可以創建多個topic， 每個topic都相當于一個消息隊列，這就意味著可以將不同格式的數據發布到不同的topic中，減小消費這些數據時的邏輯難度。那么每個topic中處理的數據結構是怎樣呢？我們先來看一張topic的解剖圖:

圖1：topic原理解析圖

從圖1中可以看到， 消息傳送過來時kafka會通過負載均衡將消息最終寫入到磁盤上一個特定分區（partition）。由于在同一個partition上這些消息都是順序存儲的， 所以對一個特定分區每條消息都會有一個基于起始位置的偏移量（offset）， 因此我們在后續消費時只需要指明從哪個partition中哪個offset開始消費，就能達到重復消費目的。

1）雖然kafka可以通過增加partition方式來增加負載，但是它的數據最終是被寫入到磁盤中。比如機械磁盤寫入效率是很低的， 難道我們需要增大一個topic的負載給它設置更多的partition嗎？

機械磁盤驅動器吞吐量跟尋道延時是強關聯，也就是說，線性讀寫速度遠大于隨機讀寫。例如，在67200rpm SATA RAID-5磁盤陣列中， 隨機寫速度大約是100k/s, 然而線性寫速度可以達到600M/s，后者大約是前者的6000倍。通過圖1可知， kafka采用的即是后者, 利用操作系統read-ahead和write-behind技術，極大提升磁盤訪問性能；設置partition數量固然可以從磁盤讀寫角度增大topic負載，但是partition數量過多會導致cpu計算量增大，所以***辦法是根據不同配置的機器， 不同的業務場景設置不同的partition數量。

2）偏移量offset存儲類型是什么， 如果消息足夠大，offset的值是否會重新置0， 如果置0，后續消費是否會紊亂？

kafka offset 是一個日志序列號（ log sequence number），不必擔心offset 長度問題。那么這個日志序列號到底有多大，舉個例子：如果一個partition一天接收1T日志， 這個offset至少可以使用1百萬年。由于offset足夠用，而且不會被置0，所以從這個角度講消費紊亂情況是不會出現的。

3）寫入磁盤的日志會被***保留嗎？如果想刪除過期消息， 需要怎么操作？

可以通過配置文件中log.retention參數設置消息過期時間，超過過期時間的消息會被系統刪除，刪除的消息不可再被重新消費。

2，分布式集群

通過前文介紹我們已經了解到kafka通過partition和順序讀寫磁盤的方式達到很高吞吐量，可是單臺機器吞吐量再高一旦該機發生故障宕掉就會對業務產生災難性影響，怎么處理這個問題呢？想必你已經知道了，那就是采用集群的方式，一旦一臺機器發生故障客戶端可以選擇鏈接其它機器， 保證業務穩定性。每一個partition 都會有一個服務器來作為***（leader）， 另外一個或者多個服務器（server）來作為跟隨者（follower），leader會處理所有的讀寫請求，而follower則會從leader那里備份數據， 如果一個leader失敗了， 其它的follower會自動選舉一個成為一個新的leader， 所以對于一個server來說，他可能是某些partition下的leader， 而對于另外一些partition來說則是follower,這樣設計可以將負載更好均衡。

1）搭建kafka集群時有沒有什么小細節需要值得注意的？

kafka官網已經有詳細的搭建過程，在此不贅述。建議正式項目中不要采用偽集群（多個broker運行在同一臺物理機上）的搭建方式，而且zookeeper集群和kafka集群***不要出現在同一臺實體機上，這樣會影響kafka順序讀寫效率。

2）在kafka集群中如果一個server失敗， 怎樣保證數據完整性？

在kafka配置文件中有一個復制因子控制參數，如果將該參數設置為N，則表示一份數據會被保存N次，而這些數據被備份到不同server中，所以當設置復制因子為N時即使有N-1臺server失敗，也會保證數據完整性。

3，生產者消費者和消息的順序性：

上面講了那么多，無非是要實現一個隊列的數據結構。對于隊列這種數據結構我們一點也不陌生，由此可以想到對于kafka的一個topic 隊列來說，生產消費邏輯應該是這樣：有很多生產者向topic中寫入數據，另外一端則有許多消費者消費數據。（見圖2）

圖2：生產者消費者原理解析圖

然而實際上kafka生產者消費者模式有它的特殊性，那么kafka這個隊列是怎樣實現入隊和出隊的？接下來我們一起來看看kafka生產者消費者模式。

生產者：生產者（producer）顧名思義，就是向kafka隊列中發布消息的，即入隊操作者。生產者功能是在topic中選擇一個partion 然后向這個partition中發送數據。選擇partition的過程就是一個負載均衡的方式， 比如可以采用輪詢或者自己設定partition選擇函數來實現負載均衡。當然如果使用封裝的api比如（https://github.com/dpkp/kafka-python）就大可不必關心負載均衡問題。會有默認的負載均衡函數來實現這一功能。

消費者： 消費者（consumer）功能是從隊列中讀取數據并進行相應邏輯處理，但是kafka消費者有特殊之處。kafka增加了一個組（group）的概念，一個topic可以有多個group， 當多個consumer從屬于一個組時，一條消息將被發往所有組，但是在組內，這條消息只會被一個consumer消費。由此說來一個group才是一個真正“邏輯消費者（logic consumer）”。相關邏輯如圖3所示。

消息順序性：通過圖3我們知道消息的消費情況，那么一個消息流消費情況會是怎樣的？其實在高等級api中由于指定了負載均衡規則，同一個生產者發布兩條不同消息數據時會根據相應規則發送到一個特定partition中，在消費時會按照同樣規則從partition中取出數據，這樣就能保證兩條數據消費的先后順序，從而保證了消息順序性。

1）對于一個具有多個consumer的topic，我要實現一條消息被多個consumer消費和一條消息只被一個consumer消費，那我需要怎么設置group？

將多個consumer設置為同一個組可以實現一條消息只被多個consumer消費， 將所有的consumer都設置為不同組，一條消息將會被所有consumer消費。

2）如果有一批數據消費時必須嚴格按照入隊先后順序來消費，需要怎樣設置生產者和消費者。

如果數據量小，可以將topic設置為一個partition；如果數據量較大，可以將一個生產者寫死負載均衡函數，將數據發送到一個特定partition上，消費數據時指定消費者消費的partition，和offset來順序消費數據。

圖3：多個消費者組時消息流向原理圖

二, Kafka性能測試：

kafka是跨語言消息隊列系統，github上提供了Java， Python等多種語言客戶端，為了簡單起見，我們這里采用kafka-python（https://github.com/dpkp/kafka-python）作為客戶端來鏈接kafka集群做測試。

測試環境：

1， broker 數量：3
2， 備份因子數：2
3， 磁盤信息：200G普通機械硬盤
4， cpu參數：8核8線程
5， 語言： Python2.7
6, 客戶端： kafka-python
7, partition 數量： 5

單進程producer 發送10條消息測試（如圖4）：

圖4：一個生產者發送消息延時結果圖

統計上圖數據可知平均延時：0.004488707，也就是說qps可以達到2000，這樣的成績無疑是驚人的。那么在多進程情況下kafka表現還會好嗎？我們設置10個進程，看看kafka在10個進程下的延時會有較大的變化嗎？如圖5（打印消息過多，截取部分結果圖）：

圖5：多個生產者發送消息延時結果圖（部分）

由圖5可知10 個進程每個進程發送10條消息，平均延時為0.00050380466秒, qps接近200000，由于kafka支持數千個客戶端同時讀寫，所以kafka吞吐能力是驚人的，更多測試歡迎大家去完成。

三，kafka在達觀數據的應用介紹

1，在垂直搜索中的應用：

我們知道搜索引擎需要定時對文檔進行更新， 如果我們把需要更新內容暫存到 kafka，這樣索引更新時,只需要從對應 partition 中從上一次取過的 offset 處繼續取數據，就能達到增量更新目的，而過期數據會被自動清理， 減少了操作冗余性和復雜性。

2，在用戶畫像以及相關推薦中的應用：

和用戶畫像之前上報的用戶點擊行為數據不同，相關推薦之前的海量 item 數據上報對數據準確性要求更高，試想如果一條 item 數據因為處理失敗而沒有正確入庫，那么相關推薦時就永遠不會出現這條 item， 所以這就對“可回滾”提出了更加嚴格要求。然而在 kafka 中，也只需要將消費的 offset 重新置為消費失敗時的 offset，修復入庫問題重新消費即可。

當然 kafka 還有更加廣泛的應用，這里就不一一討論，根據官網的介紹，kafka 在網站行為追蹤（Website Activity Tracking）、數據監控， 流處理等眾多方面有特長，如果你對 kafka 原理有研究或者有實際應用方面有心得，歡迎來討論，謝謝！

關于達觀數據

達觀數據專注于企業大數據技術服務，以***的多層智能挖掘算法，實現對海量用戶行為和文本數據的深入分析和挖掘，為企業提供智能文本分析、精準用戶行為建模、個性化推薦、智能搜索等***數據挖掘功能。