一．了解淘寶Kafka架構(gòu)

在ActiveMQ、RabbitMQ、RocketMQ、Kafka消息中間件之間，我們?yōu)槭裁匆x擇Kafka?下面詳細(xì)介紹一下，2012年9月份我在支付寶做余額寶研發(fā)，2013年6月支付寶正式推出余額寶，2013年8月?lián)沃Ц秾毺詫毑势表?xiàng)目經(jīng)理帶領(lǐng)兄弟們一起做研發(fā)，期間需要與淘寶和500萬對接競彩接口數(shù)據(jù)，業(yè)余時(shí)間與淘寶的同事溝通，了解天貓?jiān)陔娚坦?jié)如何處理這些大數(shù)據(jù)的？技術(shù)架構(gòu)上采用了哪些策略呢？

一、應(yīng)用無狀態(tài)(淘寶session框架)

二、有效使用緩存(Tair)

三、應(yīng)用拆分(HSF)

四、數(shù)據(jù)庫拆分(TDDL)

五、異步通信(Notify)

六、非結(jié)構(gòu)化數(shù)據(jù)存儲(chǔ) ( TFS,NOSQL)

七、監(jiān)控、預(yù)警系統(tǒng)

八、配置統(tǒng)一管理

天貓的同事把大致的架構(gòu)跟我描述了一番，心有感悟。咱們來看一下2018年雙11當(dāng)天的成交額。

二．kafka實(shí)現(xiàn)天貓億萬級數(shù)據(jù)統(tǒng)計(jì)架構(gòu)

Flume是Cloudera提供的一個(gè)高可用的，高可靠的，分布式的海量日志采集、聚合和傳輸?shù)南到y(tǒng)，F(xiàn)lume支持在日志系統(tǒng)中定制各類數(shù)據(jù)發(fā)送方，用于收集數(shù)據(jù)；同時(shí)，F(xiàn)lume提供對數(shù)據(jù)進(jìn)行簡單處理，并寫到各種數(shù)據(jù)接受方（可定制）的能力

Data Access：數(shù)據(jù)通道

Computing：計(jì)算

Persistence：執(zhí)行保存方式

spout：表示一個(gè)流的源頭，產(chǎn)生tuple

bolt：處理輸入流并產(chǎn)生多個(gè)輸出流，可以做簡單的數(shù)據(jù)轉(zhuǎn)換計(jì)算，復(fù)雜的流處理一般需要經(jīng)過多個(gè)bolt進(jìn)行處理

為什么不能用分布式文件HDFS集群？

1、實(shí)時(shí)性：hdfs的實(shí)時(shí)性沒有kafka高。

2、消費(fèi)量的記錄：hdfs不會(huì)記錄你這個(gè)塊文件消費(fèi)到了哪里，而基于zookeeper的kafka會(huì)記錄你消費(fèi)的點(diǎn)。

3、并發(fā)消費(fèi)：hdfs不支持并發(fā)消費(fèi)，而kafka支持并發(fā)消費(fèi)，即多個(gè)consumer.

4、彈性且有序：當(dāng)數(shù)據(jù)量會(huì)很大，而且處理完之后就可以刪除時(shí)，頻繁的讀寫會(huì)對hdfs中NameNode造成很大的壓力。而kafka的消費(fèi)點(diǎn)是記錄在zookeeper的，并且kafka的每條數(shù)據(jù)都是有“坐標(biāo)”的，所以消費(fèi)的時(shí)候只要這個(gè)“坐標(biāo)”向后移動(dòng)就行了，而且刪除的時(shí)候只要把這個(gè)“坐標(biāo)”之前的數(shù)據(jù)刪掉即可。

三．什么是Kafka?

通過上圖就可以了解到，生產(chǎn)者Producers(農(nóng)民和廚師)，消費(fèi)主題top(魚，骨頭，草，香蕉),消費(fèi)者Comsumer(貓，狗，老牛，猴子)，生產(chǎn)者根據(jù)消費(fèi)主題獲取自己想要的食物

四．Kafka架構(gòu)原理

五．Kafka能幫我們解決什么問題？

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

一．消息隊(duì)列

假設(shè)你意氣風(fēng)發(fā)，要開發(fā)新一代的互聯(lián)網(wǎng)應(yīng)用，以期在互聯(lián)網(wǎng)事業(yè)中一展宏圖。借助云計(jì)算，很容易開發(fā)出如下原型系統(tǒng)：

Web應(yīng)用：部署在云服務(wù)器上，為個(gè)人電腦或者移動(dòng)用戶提供的訪問體驗(yàn)。

SQL數(shù)據(jù)庫：為Web應(yīng)用提供數(shù)據(jù)持久化以及數(shù)據(jù)查詢。

這套架構(gòu)簡潔而高效，很快能夠部署到百度云等云計(jì)算平臺(tái)，以便快速推向市場。互聯(lián)網(wǎng)不就是講究小步快跑嘛！

好景不長。隨著用戶的迅速增長，所有的訪問都直接通過SQL數(shù)據(jù)庫使得它不堪重負(fù)，不得不加上緩存服務(wù)以降低SQL數(shù)據(jù)庫的荷載；為了理解用戶行為，開始收集日志并保存到Hadoop上離線處理，同時(shí)把日志放在全文檢索系統(tǒng)中以便快速定位問題；由于需要給投資方看業(yè)務(wù)狀況，也需要把數(shù)據(jù)匯總到數(shù)據(jù)倉庫中以便提供交互式報(bào)表。此時(shí)的系統(tǒng)的架構(gòu)已經(jīng)盤根錯(cuò)節(jié)了，考慮將來還會(huì)加入實(shí)時(shí)模塊以及外部數(shù)據(jù)交互，真是痛并快樂著……

這時(shí)候，應(yīng)該跑慢一些，讓靈魂跟上來。

本質(zhì)上，這是一個(gè)數(shù)據(jù)集成問題。沒有任何一個(gè)系統(tǒng)能夠解決所有的事情，所以業(yè)務(wù)數(shù)據(jù)根據(jù)不同用途存而放在不同的系統(tǒng)，比如歸檔、分析、搜索、緩存等。數(shù)據(jù)冗余本身沒有任何問題，但是不同系統(tǒng)之間像意大利面條一樣復(fù)雜的數(shù)據(jù)同步卻是挑戰(zhàn)。

這時(shí)候就輪到Kafka出場了。

Kafka可以讓合適的數(shù)據(jù)以合適的形式出現(xiàn)在合適的地方。Kafka的做法是提供消息隊(duì)列，讓生產(chǎn)者單往隊(duì)列的末尾添加數(shù)據(jù)，讓多個(gè)消費(fèi)者從隊(duì)列里面依次讀取數(shù)據(jù)然后自行處理。之前連接的復(fù)雜度是O(N^2)，而現(xiàn)在降低到O(N)，擴(kuò)展起來方便多了：

在Kafka的幫助下，你的互聯(lián)網(wǎng)應(yīng)用終于能夠支撐飛速增長的業(yè)務(wù)，成為下一個(gè)BAT指日可待。

以上故事說明了Kafka主要用途是數(shù)據(jù)集成，或者說是流數(shù)據(jù)集成，以Pub/Sub形式的消息總線形式提供。但是，Kafka不僅僅是一套傳統(tǒng)的消息總線，本質(zhì)上Kafka是分布式的流數(shù)據(jù)平臺(tái)，因?yàn)橐韵绿匦远?/p>

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

以高容錯(cuò)的方式存儲(chǔ)海量數(shù)據(jù)流。

保證數(shù)據(jù)流的順序。

二．日志采集

隨著互聯(lián)網(wǎng)的不斷發(fā)展，用戶所產(chǎn)生的行為數(shù)據(jù)被越來越多的網(wǎng)站重視，如何對于用戶信息進(jìn)行采集則越來越受到重視，下面就為大家介紹基于Kafka的服務(wù)端用戶行為日志采集方式。

1. 技術(shù)選型

服務(wù)端日志采集主要通過在Controller的接口中進(jìn)行埋點(diǎn)，然后通過AOP技術(shù)、Kafka消息系統(tǒng)以及l(fā)ogback對用戶行為進(jìn)行采集。

之所以使用AOP技術(shù)是因?yàn)锳OP的以下重要特定：

代碼的侵入性小。對于業(yè)務(wù)代碼的侵入性小，只需要在Controller的接口上添加注解，然后在其他模塊對用戶行為進(jìn)行采集。

重用性。對于相同作用的代碼可以進(jìn)行重用。

擴(kuò)展性。能夠很好的對系統(tǒng)進(jìn)行擴(kuò)展。

由于使用異步方式對用戶行為信息進(jìn)行收集，因此需要使用消息中間件。目前消息中間件非常多，比較流行的有ActiveMQ、ZeroMQ、RabbitMQ、Kafka等。每個(gè)消息中間件都有各種的優(yōu)勢劣勢，之所以使用Kafka消息中間件，是因?yàn)橐韵聨c(diǎn)因素：

高性能。每秒鐘可以處理數(shù)以千計(jì)生產(chǎn)者生成的消息。

高擴(kuò)展性?？梢酝ㄟ^簡單的增加服務(wù)器橫向擴(kuò)展Kafka集群的容量。

分布式。消息來自數(shù)以千計(jì)的服務(wù)，使用分布式來解決單機(jī)處理海量數(shù)據(jù)的瓶頸。

持久性。Kafka中的消息可以持久化到硬盤上，這樣可以防止數(shù)據(jù)的丟失。

因?yàn)橛脩舻男袨閿?shù)據(jù)最終是以日志的形式持久化的，因此使用logback對日志持久化到日志服務(wù)器中。

2.總體架構(gòu)

圖1 總體架構(gòu)圖

服務(wù)端日志采集系統(tǒng)主要由兩個(gè)工程組成：陸金所-bi-core和lu-bi-service。由于中國平安陸金所使用dubbo框架，因此有服務(wù)提供方和服務(wù)消費(fèi)方。lu-bi-core被web、wap和mainsite服務(wù)消費(fèi)方依賴。此外，lu-bi-service也依賴于lu-bi-core，主要是依賴于其中的一些實(shí)體類及工具類。

lu-bi-core工程為Kafka消息的生產(chǎn)者，主要封裝實(shí)現(xiàn)切面的具體邏輯，其主要職責(zé)如下：

解析用戶請求的Request信息：從Request中提取用戶的基本信息，如設(shè)備型號(hào)、用戶的供應(yīng)商、ip、設(shè)備的分辨率、設(shè)備平臺(tái)、設(shè)備的操作系統(tǒng)、設(shè)備id、app渠道等。

接口對應(yīng)的參數(shù)：通過切面可以提取接口的參數(shù)值，從而知道用戶的業(yè)務(wù)信息。

應(yīng)用層返回的結(jié)果信息：因?yàn)榍忻媸褂肁fterReturning方式，因此可以獲取用層的返回結(jié)果，從返回結(jié)果中可以提取有用的信息。

用戶的基本信息：用戶的id信息。

信息格式化：將信息轉(zhuǎn)化成JSON字符串。

發(fā)送消息：將最終需要發(fā)送的消息放入本地阻塞隊(duì)列中，通過另一個(gè)線程異步從阻塞隊(duì)列中獲取消息并發(fā)送到Kafka Broker中。

lu-bi-service工程為Kafka消息的消費(fèi)者，其主要職責(zé)如下：

實(shí)時(shí)從Kafka中拉取最新的數(shù)據(jù)。

將JSON字符串轉(zhuǎn)化成，方便進(jìn)一步對用信息進(jìn)行加工。

對用戶的ip進(jìn)行解析，獲取ip對應(yīng)的地區(qū)以及經(jīng)緯度信息。

將加工好的最終信息持久化到log文件中。

3.部署圖

圖2 部署圖

上圖為陸金所與日志系統(tǒng)系統(tǒng)相關(guān)的部署圖，App、Wap和Mainsite服務(wù)器集群分別對應(yīng)不同終端的應(yīng)用。Kafka集群使用杭研的集群，目前有10個(gè)Broker。日志服務(wù)器有兩臺(tái)，通過Kafka的均衡策略對日志進(jìn)行消費(fèi)。

4.日志采集的流程

日志采集流程圖如下所示：

圖3 日志打點(diǎn)流程圖

上圖為消息生產(chǎn)者和消息消費(fèi)者共同組成的流程圖。

消息生產(chǎn)者的具體步驟如下：

通過切面攔截用戶的請求。

從切面中提取請求頭的基本信息，如設(shè)備信息，cookie信息，ip信息等。

提取請求的接口參數(shù)信息。

從接口返回值中提取相關(guān)信息，如id，pvid等。

將提取的信息封裝成JSON字符串，放到阻塞隊(duì)列中，假如阻塞隊(duì)列溢出會(huì)有三次重試機(jī)制。

異步線程從本地阻塞隊(duì)列中獲取數(shù)據(jù)，并將信息組裝發(fā)送到Kafka的Broker中，此時(shí)消息生產(chǎn)者結(jié)束。

消息消費(fèi)者的具體步驟如下：

實(shí)時(shí)從Kafka Broker中批量拉取消息。

將拉取的消息轉(zhuǎn)化成對象。

解析ip對應(yīng)的國家、省份、城市、經(jīng)緯度信息。

對不同業(yè)務(wù)場景的信息進(jìn)一步解析。

將日志信息轉(zhuǎn)化成JSON字符串，持久化到log文件中。

5. 相關(guān)配置

application-XXX.properties：該配置放Kafka的相關(guān)屬性，包括topic、groupId、server等信息。

lu-log-msg.xml：該配置放在app-web，mainsite-web，wap-web的src/main/resources目錄下，主要是初始化kafka生產(chǎn)者的信息。

lu-bi-service.xml：該配置放在lu-bi-service工程的src/main/resources目錄下，主要用于加載kafka消費(fèi)者的配置信息，并且啟動(dòng)kafka消費(fèi)者服務(wù)。

logback.xml：該配置放在lu-bi-service工程的src/main/resources目錄下，主要用于聲明日志文件存放的目錄，需要持久化的日志的package路徑，以及日志持久化的格式。

ip_conf.txt：該配置放在lu-bi-service工程的src/main/resources目錄下，用于解析ip對應(yīng)的地域、經(jīng)緯度等信息。

六．關(guān)于面試問題

1.Redis和Kafka區(qū)別？

作者跟大家舉個(gè)例子：

老板有個(gè)好消息要告訴大家，公司要發(fā)放年終獎(jiǎng)，有兩個(gè)辦法:

1.到會(huì)議室每個(gè)座位上挨個(gè)兒告訴每個(gè)人。什么？張三去上廁所了？那張三就只能錯(cuò)過好消息了！

2.老板把消息寫到會(huì)議上的黑板報(bào)上，誰想知道就來看一下，什么？張三請假了？沒關(guān)系，我一周之后才擦掉，總會(huì)看見的！什么張三請假兩周？那就算了，我反正只保留一周，不然其他好消息沒地方寫了

redis用第一種辦法，kafka用第二種辦法，知道什么區(qū)別了吧

Redis PUB/SUB使用場景：

1. 消息持久性需求不高

2. 吞吐量要求不高

3. 可以忍受數(shù)據(jù)丟失

4. 數(shù)據(jù)量不大

Kafka使用場景：

上面以外的其他場景：）

1. 高可靠性

2. 高吞吐量

3. 持久性高

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

有關(guān)測試結(jié)論

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

RocketMQ也表現(xiàn)不俗，吞吐量在11.6w/s，磁盤IO %util已接近100%。RocketMQ的消息寫入內(nèi)存后即返回ack，由單獨(dú)的線程專門做刷盤的操作，所有的消息均是順序?qū)懳募?/p>

RabbitMQ的吞吐量5.95w/s，CPU資源消耗較高。它支持AMQP協(xié)議，實(shí)現(xiàn)非常重量級，為了保證消息的可靠性在吞吐量上做了取舍。我們還做了RabbitMQ在消息持久化場景下的性能測試，吞吐量在2.6w/s左右。

在服務(wù)端處理同步發(fā)送的性能上，Kafka>RocketMQ>RabbitMQ

寫在最后

如今都在談?wù)摵卸嗫膳拢P者作為一個(gè)過來人，卻有不同的看法：寒冬不可怕，在寒冬里沒有生存能力，才是最可怕的。

因此小編總結(jié)了這幾年在阿里的工作經(jīng)驗(yàn)并結(jié)合目前互聯(lián)網(wǎng)最主流的Java架構(gòu)技術(shù)，最后錄制了七大Java架構(gòu)技術(shù)專題視頻（源碼閱讀、分布式架構(gòu)、微服務(wù)、性能優(yōu)化、阿里項(xiàng)目實(shí)戰(zhàn)、Devops、并發(fā)編程）分享在我的裙171662117中，并且每晚我都會(huì)在群內(nèi)直播講解這些架構(gòu)技術(shù)的底層實(shí)現(xiàn)原理，感興趣的程序員們可以加群找管理員獲取。