Kafka對Java程序員有多重要？連阿里都再用它處理億萬級數據統計

一．瞭解淘寶Kafka架構

在ActiveMQ、RabbitMQ、RocketMQ、Kafka消息中間件之間，咱們爲何要選擇Kafka?下面詳細介紹一下，2012年9月份我在支付寶作餘額寶研發，2013年6月支付寶正式推出餘額寶，2013年8月擔任支付寶淘寶彩票項目經理帶領兄弟們一塊兒作研發，期間須要與淘寶和500萬對接競彩接口數據，業餘時間與淘寶的同事溝通，瞭解天貓在電商節如何處理這些大數據的？技術架構上採用了哪些策略呢？

1、應用無狀態(淘寶session框架)

2、有效使用緩存(Tair)

3、應用拆分(HSF)

4、數據庫拆分(TDDL)

5、異步通訊(Notify)

6、非結構化數據存儲 ( TFS,NOSQL)

7、監控、預警系統

8、配置統一管理

天貓的同事把大體的架構跟我描述了一番，心有感悟。我們來看一下2018年雙11當天的成交額。

 

二．kafka實現天貓億萬級數據統計架構

 

Flume是Cloudera提供的一個高可用的，高可靠的，分佈式的海量日誌採集、聚合和傳輸的系統，Flume支持在日誌系統中定製各種數據發送方，用於收集數據；同時，Flume提供對數據進行簡單處理，並寫到各類數據接受方（可定製）的能力

• Data Access：數據通道
• Computing：計算
• Persistence：執行保存方式
• spout：表示一個流的源頭，產生tuple
• bolt：處理輸入流併產生多個輸出流，能夠作簡單的數據轉換計算，複雜的流處理通常須要通過多個bolt進行處理

爲何不能用分佈式文件HDFS集羣？

一、實時性：hdfs的實時性沒有kafka高。

二、消費量的記錄：hdfs不會記錄你這個塊文件消費到了哪裏，而基於zookeeper的kafka會記錄你消費的點。

三、併發消費：hdfs不支持併發消費，而kafka支持併發消費，即多個consumer.

四、彈性且有序：當數據量會很大，並且處理完以後就能夠刪除時，頻繁的讀寫會對hdfs中NameNode形成很大的壓力。而kafka的消費點是記錄在zookeeper的，而且kafka的每條數據都是有「座標」的，因此消費的時候只要這個「座標」向後移動就好了，並且刪除的時候只要把這個「座標」以前的數據刪掉便可。

三．什麼是Kafka?

 

經過上圖就能夠了解到，生產者Producers(農民和廚師)，消費主題top(魚，骨頭，草，香蕉),消費者Comsumer(貓，狗，老牛，猴子)，生產者根據消費主題獲取本身想要的食物

四．Kafka架構原理

 

五．Kafka能幫咱們解決什麼問題？

請高手指明一下kafka解決了什麼問題，什麼場景下使用？消息訂閱和發佈嗎，好像redis也支持，功能是否有重疊？

一．消息隊列

假設你意氣風發，要開發新一代的互聯網應用，以期在互聯網事業中一展宏圖。藉助雲計算，很容易開發出以下原型系統：

• Web應用：部署在雲服務器上，爲我的電腦或者移動用戶提供的訪問體驗。
• SQL數據庫：爲Web應用提供數據持久化以及數據查詢。

 

這套架構簡潔而高效，很快可以部署到百度雲等雲計算平臺，以便快速推向市場。互聯網不就是講究小步快跑嘛！

好景不長。隨着用戶的迅速增加，全部的訪問都直接經過SQL數據庫使得它不堪重負，不得不加上緩存服務以下降SQL數據庫的荷載；爲了理解用戶行爲，開始收集日誌並保存到Hadoop上離線處理，同時把日誌放在全文檢索系統中以便快速定位問題；因爲須要給投資方看業務情況，也須要把數據彙總到數據倉庫中以便提供交互式報表。此時的系統的架構已經盤根錯節了，考慮未來還會加入實時模塊以及外部數據交互，真是痛並快樂着……

 

這時候，應該跑慢一些，讓靈魂跟上來。

本質上，這是一個數據集成問題。沒有任何一個系統可以解決全部的事情，因此業務數據根據不一樣用途存而放在不一樣的系統，好比歸檔、分析、搜索、緩存等。數據冗餘自己沒有任何問題，可是不一樣系統之間像意大利麪條同樣複雜的數據同步倒是挑戰。

這時候就輪到Kafka出場了。

Kafka可讓合適的數據以合適的形式出如今合適的地方。Kafka的作法是提供消息隊列，讓生產者單往隊列的末尾添加數據，讓多個消費者從隊列裏面依次讀取數據而後自行處理。以前鏈接的複雜度是O(N^2)，而如今下降到O(N)，擴展起來方便多了：

 

在Kafka的幫助下，你的互聯網應用終於可以支撐飛速增加的業務，成爲下一個BAT指日可待。

以上故事說明了Kafka主要用途是數據集成，或者說是流數據集成，以Pub/Sub形式的消息總線形式提供。可是，Kafka不只僅是一套傳統的消息總線，本質上Kafka是分佈式的流數據平臺，由於如下特性而著名：

• 提供Pub/Sub方式的海量消息處理。
• 以高容錯的方式存儲海量數據流。
• 保證數據流的順序。

二．日誌採集

隨着互聯網的不斷髮展，用戶所產生的行爲數據被愈來愈多的網站重視，如何對於用戶信息進行採集則愈來愈受到重視，下面就爲你們介紹基於Kafka的服務端用戶行爲日誌採集方式。

1. 技術選型

服務端日誌採集主要經過在Controller的接口中進行埋點，而後經過AOP技術、Kafka消息系統以及logback對用戶行爲進行採集。

之因此使用AOP技術是由於AOP的如下重要特定：

• 代碼的侵入性小。對於業務代碼的侵入性小，只須要在Controller的接口上添加註解，而後在其餘模塊對用戶行爲進行採集。
• 重用性。對於相同做用的代碼能夠進行重用。
• 擴展性。可以很好的對系統進行擴展。

因爲使用異步方式對用戶行爲信息進行收集，所以須要使用消息中間件。目前消息中間件很是多，比較流行的有ActiveMQ、ZeroMQ、RabbitMQ、Kafka等。每一個消息中間件都有各類的優點劣勢，之因此使用Kafka消息中間件，是由於如下幾點因素：

• 高性能。每秒鐘能夠處理數以千計生產者生成的消息。
• 高擴展性。能夠經過簡單的增長服務器橫向擴展Kafka集羣的容量。
• 分佈式。消息來自數以千計的服務，使用分佈式來解決單機處理海量數據的瓶頸。
• 持久性。Kafka中的消息能夠持久化到硬盤上，這樣能夠防止數據的丟失。

由於用戶的行爲數據最終是以日誌的形式持久化的，所以使用logback對日誌持久化到日誌服務器中。

2.整體架構

圖1 整體架構圖

服務端日誌採集系統主要由兩個工程組成：陸金所-bi-core和lu-bi-service。因爲中國平安陸金所使用dubbo框架，所以有服務提供方和服務消費方。lu-bi-core被web、wap和mainsite服務消費方依賴。此外，lu-bi-service也依賴於lu-bi-core，主要是依賴於其中的一些實體類及工具類。

lu-bi-core工程爲Kafka消息的生產者，主要封裝實現切面的具體邏輯，其主要職責以下：

• 解析用戶請求的Request信息：從Request中提取用戶的基本信息，如設備型號、用戶的供應商、ip、設備的分辨率、設備平臺、設備的操做系統、設備id、app渠道等。
• 接口對應的參數：經過切面能夠提取接口的參數值，從而知道用戶的業務信息。
• 應用層返回的結果信息：由於切面使用AfterReturning方式，所以能夠獲取用層的返回結果，從返回結果中能夠提取有用的信息。
• 用戶的基本信息：用戶的id信息。
• 信息格式化：將信息轉化成JSON字符串。
• 發送消息：將最終須要發送的消息放入本地阻塞隊列中，經過另外一個線程異步從阻塞隊列中獲取消息併發送到Kafka Broker中。

lu-bi-service工程爲Kafka消息的消費者，其主要職責以下：

• 實時從Kafka中拉取最新的數據。
• 將JSON字符串轉化成，方便進一步對用信息進行加工。
• 對用戶的ip進行解析，獲取ip對應的地區以及經緯度信息。
• 將加工好的最終信息持久化到log文件中。

3.部署圖

圖2 部署圖

上圖爲陸金所與日誌系統系統相關的部署圖，App、Wap和Mainsite服務器集羣分別對應不一樣終端的應用。Kafka集羣使用杭研的集羣，目前有10個Broker。日誌服務器有兩臺，經過Kafka的均衡策略對日誌進行消費。

4.日誌採集的流程

日誌採集流程圖以下所示：

圖3 日誌打點流程圖

上圖爲消息生產者和消息消費者共同組成的流程圖。

• 消息生產者的具體步驟以下：
• 經過切面攔截用戶的請求。
• 從切面中提取請求頭的基本信息，如設備信息，cookie信息，ip信息等。
• 提取請求的接口參數信息。
• 從接口返回值中提取相關信息，如id，pvid等。
• 將提取的信息封裝成JSON字符串，放到阻塞隊列中，假如阻塞隊列溢出會有三次重試機制。
• 異步線程從本地阻塞隊列中獲取數據，並將信息組裝發送到Kafka的Broker中，此時消息生產者結束。

消息消費者的具體步驟以下：

• 實時從Kafka Broker中批量拉取消息。
• 將拉取的消息轉化成對象。
• 解析ip對應的國家、省份、城市、經緯度信息。
• 對不一樣業務場景的信息進一步解析。
• 將日誌信息轉化成JSON字符串，持久化到log文件中。

5. 相關配置

• application-XXX.properties：該配置放Kafka的相關屬性，包括topic、groupId、server等信息。
• lu-log-msg.xml：該配置放在app-web，mainsite-web，wap-web的src/main/resources目錄下，主要是初始化kafka生產者的信息。
• lu-bi-service.xml：該配置放在lu-bi-service工程的src/main/resources目錄下，主要用於加載kafka消費者的配置信息，而且啓動kafka消費者服務。
• logback.xml：該配置放在lu-bi-service工程的src/main/resources目錄下，主要用於聲明日誌文件存放的目錄，須要持久化的日誌的package路徑，以及日誌持久化的格式。
• ip_conf.txt：該配置放在lu-bi-service工程的src/main/resources目錄下，用於解析ip對應的地域、經緯度等信息。

六．關於面試問題

1.Redis和Kafka區別？

做者跟你們舉個例子：

老闆有個好消息要告訴你們，公司要發放年終獎，有兩個辦法:

1.到會議室每一個座位上挨個兒告訴每一個人。什麼？張三去上廁所了？那張三就只能錯過好消息了！

2.老闆把消息寫到會議上的黑板報上，誰想知道就來看一下，什麼？張三請假了？不要緊，我一週以後才擦掉，總會看見的！什麼張三請假兩週？那就算了，我反正只保留一週，否則其餘好消息沒地方寫了

redis用第一種辦法，kafka用第二種辦法，知道什麼區別了吧

Redis PUB/SUB使用場景：

1. 消息持久性需求不高

2. 吞吐量要求不高

3. 能夠忍受數據丟失

4. 數據量不大

Kafka使用場景：

上面之外的其餘場景：）

1. 高可靠性

2. 高吞吐量

3. 持久性高

Kafka、RabbitMQ、RocketMQ等消息中間件的對比

有關測試結論

Kafka的吞吐量高達17.3w/s，不愧是高吞吐量消息中間件的行業老大。這主要取決於它的隊列模式保證了寫磁盤的過程是線性IO。此時broker磁盤IO已達瓶頸。

RocketMQ也表現不俗，吞吐量在11.6w/s，磁盤IO %util已接近100%。RocketMQ的消息寫入內存後即返回ack，由單獨的線程專門作刷盤的操做，全部的消息均是順序寫文件。

RabbitMQ的吞吐量5.95w/s，CPU資源消耗較高。它支持AMQP協議，實現很是重量級，爲了保證消息的可靠性在吞吐量上作了取捨。咱們還作了RabbitMQ在消息持久化場景下的性能測試，吞吐量在2.6w/s左右。

在服務端處理同步發送的性能上，Kafka>RocketMQ>RabbitMQ

寫在最後

現在都在談論寒冬有多可怕，筆者做爲一個過來人，卻有不一樣的見解：寒冬不可怕，在寒冬裏沒有生存能力，纔是最可怕的。

所以小編總結了這幾年在阿里的工做經驗並結合目前互聯網最主流的Java架構技術，最後錄製了七大Java架構技術專題視頻（源碼閱讀、分佈式架構、微服務、性能優化、阿里項目實戰、Devops、併發編程）分享在個人裙669275137中，而且每晚我都會在羣內直播講解這些架構技術的底層實現原理，感興趣的程序員們能夠加羣找管理員獲取。