[[189864]]

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

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

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

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

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

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

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

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

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

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

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

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()方法，對相關(guān)數(shù)據(jù)進行處理。并依據(jù)數(shù)據(jù)發(fā)布到kafka的情況，處理回調(diào)機制。在數(shù)據(jù)發(fā)送失敗時，定義failedSend()方法;當(dāng)數(shù)據(jù)發(fā)送成功時，定義successedSend()方法。

Kafka消費的代碼如下：

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

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

結(jié)論

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

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