[[189864]]

我們生活在一個(gè)數(shù)據(jù)爆炸的時(shí)代，數(shù)據(jù)的巨量增長(zhǎng)給我們的業(yè)務(wù)處理帶來了壓力，同時(shí)巨量的數(shù)據(jù)也給我們帶來了十分可觀的財(cái)富。隨著大數(shù)據(jù)將各個(gè)行業(yè)用戶、運(yùn)營(yíng)商、服務(wù)商的數(shù)據(jù)整合進(jìn)大數(shù)據(jù)環(huán)境，或用戶取用大數(shù)據(jù)環(huán)境中海量的數(shù)據(jù)，業(yè)務(wù)平臺(tái)間的消息處理將變得尤為復(fù)雜。如何高效地采集、使用數(shù)據(jù)，如何減輕各業(yè)務(wù)系統(tǒng)的壓力，也變得越來越突出。在早期的系統(tǒng)實(shí)現(xiàn)時(shí)，業(yè)務(wù)比較簡(jiǎn)單。即便是數(shù)據(jù)量、業(yè)務(wù)量比較大，大數(shù)據(jù)環(huán)境也能做出處理。但是隨著接入的系統(tǒng)增多，數(shù)據(jù)量、業(yè)務(wù)量增大，大數(shù)據(jù)環(huán)境、業(yè)務(wù)系統(tǒng)都可出現(xiàn)一定的瓶頸。下面我們看幾個(gè)場(chǎng)景。

場(chǎng)景一：我們開發(fā)過一個(gè)設(shè)備信息挖掘平臺(tái)。這個(gè)平臺(tái)需要實(shí)時(shí)將采集互聯(lián)網(wǎng)關(guān)采集到的路由節(jié)點(diǎn)的狀態(tài)信息存入數(shù)據(jù)中心。通常一個(gè)網(wǎng)關(guān)一次需要上報(bào)幾十甚至幾百個(gè)變化的路由信息。全區(qū)有幾萬(wàn)個(gè)這種互聯(lián)網(wǎng)關(guān)。當(dāng)信息采集平臺(tái)將這些變化的數(shù)據(jù)信息寫入或更新到數(shù)據(jù)庫(kù)時(shí)候，會(huì)給數(shù)據(jù)庫(kù)代理非常大的壓力，甚至可以直接將數(shù)據(jù)庫(kù)搞掛掉。這就對(duì)我們的數(shù)據(jù)采集系統(tǒng)提出了很高的要求。如何穩(wěn)定高效地把消息更新到數(shù)據(jù)庫(kù)這一要求擺了出來。

場(chǎng)景二：數(shù)據(jù)中心處理過的數(shù)據(jù)需要實(shí)時(shí)共享給幾個(gè)不同的機(jī)構(gòu)。我們常采用的方法是將數(shù)據(jù)批量存放在數(shù)據(jù)采集機(jī)，分支機(jī)構(gòu)定時(shí)來采集;或是分支機(jī)構(gòu)通過JDBC、RPC、http或其他機(jī)制實(shí)時(shí)從數(shù)據(jù)中心獲取數(shù)據(jù)。這兩種方式都存在一定的問題，前者在于實(shí)時(shí)性不足，還牽涉到數(shù)據(jù)完整性問題;后者在于，當(dāng)數(shù)據(jù)量很大的時(shí)候，多個(gè)分支機(jī)構(gòu)同時(shí)讀取數(shù)據(jù)，會(huì)對(duì)數(shù)據(jù)中心的造成很大的壓力，也造成很大的資源浪費(fèi)。

為了解決以上場(chǎng)景提出的問題，我們需要這樣一個(gè)消息系統(tǒng)：

緩沖能力，系統(tǒng)可以提供一個(gè)緩沖區(qū)，當(dāng)有大量數(shù)據(jù)來臨時(shí)，系統(tǒng)可以將數(shù)據(jù)可靠的緩沖起來，供后續(xù)模塊處理;

訂閱、分發(fā)能力，系統(tǒng)可以接收消息可靠的緩存下來，也可以將可靠緩存的數(shù)據(jù)發(fā)布給使用者。

這就要我們找一個(gè)高吞吐的、能滿足訂閱發(fā)布需求的系統(tǒng)。

Kafka是一個(gè)分布式的、高吞吐的、基于發(fā)布/訂閱的消息系統(tǒng)。利用kafka技術(shù)可以在廉價(jià)PC Server上搭建起大規(guī)模的消息系統(tǒng)。Kafka具有消息持久化、高吞吐、分布式、實(shí)時(shí)、低耦合、多客戶端支持、數(shù)據(jù)可靠等諸多特點(diǎn)，適合在線和離線的消息處理。

使用kafka解決我們上述提到的問題。

互聯(lián)網(wǎng)關(guān)采集到變化的路由信息，通過kafka的producer將歸集后的信息批量傳入kafka。Kafka按照接收順序?qū)w集的信息進(jìn)行緩存，并加入待消費(fèi)隊(duì)列。Kafka的consumer讀取隊(duì)列信息，并一定的處理策略，將獲取的信息更新到數(shù)據(jù)庫(kù)。完成數(shù)據(jù)到數(shù)據(jù)中心的存儲(chǔ)。

數(shù)據(jù)中心的數(shù)據(jù)需要共享時(shí)，kafka的producer先從數(shù)據(jù)中心讀取數(shù)據(jù)，然后傳入kafka緩存并加入待消費(fèi)隊(duì)列。各分支結(jié)構(gòu)作為數(shù)據(jù)消費(fèi)者，啟動(dòng)消費(fèi)動(dòng)作，從kafka隊(duì)列讀取數(shù)據(jù)，并對(duì)獲取的數(shù)據(jù)進(jìn)行處理。

Kafka生產(chǎn)的代碼如下：

public void produce(){      
 
                   //生產(chǎn)消息預(yù)處理 
 
        produceInfoProcess();        
 
        pro.send(ProducerRecord,new Callback(){ 
 
                            @Override 
 
                            onCompletion() { 
 
                                     if (metadata == null) { 
 
                                               // 發(fā)送失敗 
 
                                               failedSend(); 
 
                                     } else { 
 
                                               //發(fā)送成功!"  
 
                                               successedSend();      
 
} 
 
                            }                       
 
              });   
 
     }  

消息生產(chǎn)者根據(jù)需求，靈活定義produceInfoProcess()方法，對(duì)相關(guān)數(shù)據(jù)進(jìn)行處理。并依據(jù)數(shù)據(jù)發(fā)布到kafka的情況，處理回調(diào)機(jī)制。在數(shù)據(jù)發(fā)送失敗時(shí)，定義failedSend()方法;當(dāng)數(shù)據(jù)發(fā)送成功時(shí)，定義successedSend()方法。

Kafka消費(fèi)的代碼如下：

public void consumer() { 
 
                       //配置文件 
 
            properties(); 
 
            //獲取當(dāng)前數(shù)據(jù)的迭代器 
 
            iterator = stream.iterator(); 
 
            while (iterator.hasNext()) { 
 
                //取出消息 
 
                MessageAndMetadata<byte[], byte[]> next = iterator.next(); 
 
                messageProcess(); 
 
                 }       
 
    }  

Kafka消費(fèi)者會(huì)和kafka集群建立一個(gè)連接。從kafka讀取數(shù)據(jù)，調(diào)用messageProcess()方法，對(duì)獲取的數(shù)據(jù)靈活處理。

結(jié)論

Kafka的高吞吐能力、緩存機(jī)制能有效的解決高峰流量沖擊問題。實(shí)踐表明，在未將kafka引入系統(tǒng)前，當(dāng)互聯(lián)網(wǎng)關(guān)發(fā)送的數(shù)據(jù)量較大時(shí)，往往會(huì)掛起關(guān)系數(shù)據(jù)庫(kù)，數(shù)據(jù)常常丟失。在引入kafka后，更新程序能夠結(jié)合能力自主處理消息，不會(huì)引起數(shù)據(jù)丟失，關(guān)系型數(shù)據(jù)庫(kù)的壓力波動(dòng)不會(huì)發(fā)生過于顯著的變化，不會(huì)出現(xiàn)數(shù)據(jù)庫(kù)掛起鎖死現(xiàn)象。

依靠kafka的訂閱分發(fā)機(jī)制，實(shí)現(xiàn)了一次發(fā)布，各分支依據(jù)需求自主訂閱的功能。避免了各分支機(jī)構(gòu)直接向數(shù)據(jù)中心請(qǐng)求數(shù)據(jù)，或者數(shù)據(jù)中心依次批量向分支機(jī)構(gòu)傳輸數(shù)據(jù)以致實(shí)時(shí)性不足的情況。kafka提高了實(shí)時(shí)性，減輕了數(shù)據(jù)中心的壓力，提高了效率。