【乾貨】Kafka實現淘寶億萬級數據統計（下）

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

什麼場景下使用？消息訂閱和發佈嗎，好像redis也支持，功能是否有重疊？

1

消息隊列

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

Web應用：部署在雲服務器上，爲我的電腦或者移動用戶提供的訪問體驗。

SQL數據庫：爲Web應用提供數據持久化以及數據查詢。

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

惋惜好景不長。隨着用戶的迅速增加，全部的訪問都直接經過SQL數據庫使得它不堪重負，不得不加上緩存服務以下降SQL數據庫的荷載；爲了理解用戶行爲，開始收集日誌並保存到Hadoop上離線處理，同時把日誌放在全文檢索系統中以便快速定位問題；因爲須要給投資方看業務情況，也須要把數據彙總到數據倉庫中以便提供交互式報表。

此時的系統的架構已經盤根錯節了，考慮未來還會加入實時模塊以及外部數據交互，真是痛並快樂着……

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

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

這時候就輪到Kafka出場了。

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

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

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

· 提供Pub/Sub方式的海量消息處理。

· 以高容錯的方式存儲海量數據流。

· 保證數據流的順序。

2

日誌採集

1.技術選型

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

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

· 代碼的侵入性小。對於業務代碼的侵入性小，只須要在Controller的接口上添加註解，而後在其餘模塊對用戶行爲進行採集。

· 重用性。對於相同做用的代碼能夠進行重用。

· 擴展性。可以很好的對系統進行擴展。

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

· 高性能。每秒鐘能夠處理數以千計生產者生成的消息。

· 高擴展性。能夠經過簡單的增長服務器橫向擴展Kafka集羣的容量。

· 分佈式。消息來自數以千計的服務，使用分佈式來解決單機處理海量數據的瓶頸。

· 持久性。Kafka中的消息能夠持久化到硬盤上，這樣能夠防止數據的丟失。

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

2.整體架構

服務端日誌採集系統主要由兩個工程組成：陸金所-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.部署圖

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

4.日誌採集的流程

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

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

消息生產者的具體步驟以下：

· 經過切面攔截用戶的請求。

· 從切面中提取請求頭的基本信息，如設備信息，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對應的地域、經緯度等信息。

六

關於面試問題

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進階架構交流：加入142019080。

直接點擊連接加羣。https://jq.qq.com/?_wv=1027&k=5lXBNZ7 獲取最新學習資料