干貨｜Kafka在大數(shù)據(jù)環(huán)境中的應(yīng)用

作者：馮燕青 2017-04-28 11:45:16

大數(shù)據(jù)

Kafka 我們生活在一個(gè)數(shù)據(jù)爆炸的時(shí)代，數(shù)據(jù)的巨量增長給我們的業(yè)務(wù)處理帶來了壓力，同時(shí)巨量的數(shù)據(jù)也給我們帶來了十分可觀的財(cái)富。隨著大數(shù)據(jù)將各個(gè)行業(yè)用戶、運(yùn)營商、服務(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)的壓力，也變得越來越突出。

創(chuàng)新互聯(lián)公司長期為上千多家客戶提供的網(wǎng)站建設(shè)服務(wù)，團(tuán)隊(duì)從業(yè)經(jīng)驗(yàn)10年，關(guān)注不同地域、不同群體，并針對不同對象提供差異化的產(chǎn)品和服務(wù)；打造開放共贏平臺(tái)，與合作伙伴共同營造健康的互聯(lián)網(wǎng)生態(tài)環(huán)境。為牡丹企業(yè)提供專業(yè)的網(wǎng)站設(shè)計(jì)、成都做網(wǎng)站，牡丹網(wǎng)站改版等技術(shù)服務(wù)。擁有十年豐富建站經(jīng)驗(yàn)和眾多成功案例,為您定制開發(fā)。

我們生活在一個(gè)數(shù)據(jù)爆炸的時(shí)代，數(shù)據(jù)的巨量增長給我們的業(yè)務(wù)處理帶來了壓力，同時(shí)巨量的數(shù)據(jù)也給我們帶來了十分可觀的財(cái)富。隨著大數(shù)據(jù)將各個(gè)行業(yè)用戶、運(yùn)營商、服務(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ù)比較簡單。即便是數(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è)場景。

場景一：我們開發(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ū)有幾萬個(gè)這種互聯(lián)網(wǎng)關(guān)。當(dāng)信息采集平臺(tái)將這些變化的數(shù)據(jù)信息寫入或更新到數(shù)據(jù)庫時(shí)候，會(huì)給數(shù)據(jù)庫代理非常大的壓力，甚至可以直接將數(shù)據(jù)庫搞掛掉。這就對我們的數(shù)據(jù)采集系統(tǒng)提出了很高的要求。如何穩(wěn)定高效地把消息更新到數(shù)據(jù)庫這一要求擺了出來。

場景二：數(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ì)對數(shù)據(jù)中心的造成很大的壓力，也造成很大的資源浪費(fèi)。

為了解決以上場景提出的問題，我們需要這樣一個(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ù)庫。完成數(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ù)，并對獲取的數(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()方法，對相關(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 next = iterator.next();
   
  
  
  

   
  
  
  
                messageProcess();
   
  
  
  

   
  
  
  
                 }      
   
  
  
  

   
  
  
  
    } 
  
 
 
 

Kafka消費(fèi)者會(huì)和kafka集群建立一個(gè)連接。從kafka讀取數(shù)據(jù)，調(diào)用messageProcess()方法，對獲取的數(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ù)庫，數(shù)據(jù)常常丟失。在引入kafka后，更新程序能夠結(jié)合能力自主處理消息，不會(huì)引起數(shù)據(jù)丟失，關(guān)系型數(shù)據(jù)庫的壓力波動(dòng)不會(huì)發(fā)生過于顯著的變化，不會(huì)出現(xiàn)數(shù)據(jù)庫掛起鎖死現(xiàn)象。

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

當(dāng)前題目：干貨｜Kafka在大數(shù)據(jù)環(huán)境中的應(yīng)用
本文路徑：http://www.5511xx.com/article/dhhgjcj.html