http://www.cnblogs.com/darkprince/p/5114425.html

開發Kafka通用資料平臺中介軟體

(含本次專案全部程式碼及資源)

目錄：

一. Kafka概述

二. Kafka啟動命令

三.我們為什麼使用Kafka

四. Kafka資料平臺中介軟體設計及程式碼解析

五.未來Kafka開發任務

一. Kafka概述

　　Kafka是Linkedin於2010年12月份建立的開源訊息系統，它主要用於處理活躍的流式資料。活躍的流式資料在web網站應用中非常常見，這些活動資料包括頁面訪問量（Page View）、被檢視內容方面的資訊以及搜尋情況等內容。 這些資料通常以日誌的形式記錄下來，然後每隔一段時間進行一次統計分析。

　　傳統的日誌分析系統是一種離線處理日誌資訊的方式，但若要進行實時處理，通常會有較大延遲。而現有的訊息佇列系統能夠很好的處理實時或者近似實時的應用，但未處理的資料通常不會寫到磁碟上，這對於Hadoop之類，間隔時間較長的離線應用而言，在資料安全上會出現問題。Kafka正是為了解決以上問題而設計的，它能夠很好地進行離線和線上應用。

1.1 Kfka部署結構：

(圖1)

1.2 Kafka關鍵字：

•Broker ： Kafka訊息伺服器，訊息中心。一個Broker可以容納多個Topic。

•Producer ：訊息生產者，就是向Kafka broker發訊息的客戶端。

•Consumer ：訊息消費者，向Kafka broker取訊息的客戶端。

•Zookeeper ：管理Producer，Broker，Consumer的動態加入與離開。

•Topic ：可以為各種訊息劃分為多個不同的主題，Topic就是主題名稱。Producer可以針對某個主題進行生產，Consumer可以針對某個主題進行訂閱。

•Consumer Group： Kafka採用廣播的方式進行訊息分發，而Consumer叢集在消費某Topic時， Zookeeper會為該叢集建立Offset消費偏移量，最新Consumer加入並消費該主題時，可以從最新的Offset點開始消費。

•Partition：Kafka採用對資料檔案切片（Partition）的方式可以將一個Topic可以分佈儲存到多個Broker上，一個Topic可以分為多個Partition。在多個Consumer併發訪問一個partition會有同步鎖控制。

(圖2)

1.3 訊息收發流程：

•啟動Zookeeper及Broker.

•Producer連線Broker後，將訊息釋出到Broker中指定Topic上(可以指定Patition)。

•Broker叢集接收到Producer發過來的訊息後，將其持久化到硬碟，並將訊息該保留指定時長（可配置），而不關注訊息是否被消費。

•Consumer連線到Broker後，啟動訊息泵對Broker進行偵聽，當有訊息到來時，會觸發訊息泵迴圈獲取訊息，獲取訊息後Zookeeper將記錄該Consumer的訊息Offset。

1.4 Kafka特性：

•高吞吐量

•負載均衡：通過zookeeper對Producer,Broker,Consumer的動態加入與離開進行管理。

•拉取系統：由於kafka broker會持久化資料，broker沒有記憶體壓力，因此，consumer非常適合採取pull的方式消費資料

•動態擴充套件：當需要增加broker結點時，新增的broker會向zookeeper註冊，而producer及consumer會通過zookeeper感知這些變化，並及時作出調整。

•訊息刪除策略：資料檔案將會根據broker中的配置要求,保留一定的時間之後刪除。kafka通過這種簡單的手段,來釋放磁碟空間。

二. Kafka啟動命令：

啟動Zookeeper服務：

zookeeper-server-start.bat ../../config/zookeeper.properties

啟動Broker服務：

kafka-server-start.bat ../../config/server.properties

通過Zookeeper的協調在Broker中建立一個Topic(主題)

kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testTopic

查詢當前Broker中某個指定主題的配置資訊

kafka-run-class.bat kafka.admin.TopicCommand --describe --zookeeper localhost:2181 --topic testTopic

啟動一個數據生產者Producer

kafka-console-producer.bat --broker-list localhost:9092 --topic testTopic  

啟動一個數據消費者Consumer

kafka-console-consumer.bat --zookeeper localhost:2181 --topic testTopic --from-beginning

Zookeeper配置檔案，zookeeper.properties配置片段

Broker配置檔案，server.properties配置片段

關於kafka收發訊息相關的配置項

1.在Broker Server中屬性(這些屬性需要在Server啟動時載入)：

//每次Broker Server能夠接收的最大包大小，該引數要與consumer的fetch.message.max.bytes屬性進行匹配使用

* message.max.bytes 1000000(預設) 

//Broker Server中針對Producer傳送方的資料緩衝區。Broker Server會利用該緩衝區迴圈接收來至Producer的資料 包，緩衝區過小會導致對該資料包的分段數量增加，但不會影響資料包尺寸限制問題。

socket.send.buffer.bytes 100 * 1024(預設)

//Broker Server中針對Consumer接收方的資料緩衝區。意思同上。

socket.receive.buffer.bytes 100 * 1024(預設)

//Broker Server中針對每次請求最大的緩衝區尺寸，包括Prodcuer和Consumer雙方。該值必須大於 message.max.bytes屬性

* socket.request.max.bytes 100 * 1024 * 1024(預設)

2.在Consumer中的屬性(這些屬性需要在程式中配置Consumer時設定)

//Consumer用於接收來自Broker的資料緩衝區,意思同socket.send.buffer.bytes。

socket.receive.buffer.bytes 64 * 1024(預設)

//Consumer用於每次接收訊息包的最大尺寸，該屬性需要與Broker中的message.max.bytes屬性配對使用

* fetch.message.max.bytes 1024 * 1024(預設)

3.在Producer中的屬性(這些屬性需要在程式中配置Consumer時設定)

//Producer用於傳送資料緩衝區,意思同socket.send.buffer.bytes。

send.buffer.bytes 100 * 1024（預設）

三. 我們為什麼使用Kafka

當前專案中，我們更希望從企業獲得儘可能多的有價值資料。最直接獲取大資料的方式是採用寫應用直連目標企業資料庫來獲得資料。但這種方式在實際應用中，會由於企業擔心開放本地資料庫而導致的安全隱患很難實施。另外，這種方式會與企業本地資料庫結構耦合度過高，會出現多家企業多個應用的情況，缺少統一的資料互動平臺，導致後期維護困難。

3.1 Kafka在當前專案中問題：

當前案例，我們想把某企業的本地資料實時同步到資料中心中，之後對這些資料進行二次分析處理。我們的目標是建立統一的資料同步平臺，便於在日後的多企業多系統中能有統一的實施標準，所以選用了Kafka訊息系統作為支撐。

Producer(資料傳送方)以獨立執行緒方式常駐某企業內部應用中，依靠一定的時間週期，從本地資料庫獲得資料並推送至Broker中。而Consumer(快銷組資料接收方)也是獨立與WEB框架常駐記憶體，獲得資料訊息後儲存至資料中心中。

但目前Kafka在實施中面臨以下問題：

1.Producer/Consumer均獨立於Web框架，Producer依靠訊息片輪詢檢索/傳送最新資料，執行效率低。

2.Producer會直接針對某企業內部資料庫表結構操作，導致程式碼與企業業務耦合度過高，而無法平滑移植到其他企業系統中。

3.由於Producer/Consumer是獨立於Web框架的，在外圍負責資料的採集及推送，與Web專案主程式無切合度。

4.目前針對Kafka的資料傳輸異常處理比較簡陋，當Broker或 Zookeeper等出現異常時，有可能會導致資料安全性問題。

3.2實現目標：

針對以上問題，我們要實現如下目標：

1.把Producer/Consumer的資料推送/獲取的過程封裝成Class或者Jar包的形式，供Java Web框架呼叫，從而形成與企業內部Web應用或計算中心資料分析Web應用融合一體。

2.資料的推送/獲取只針對Java Object物件，不要針對資料庫表結構，不能與企業特有資料耦合度過高，形成通用的資料介面。Producer需要對Object進行序列化，Consumer需要對序列化後的二進位制資訊進行反序列化重建Object返回給呼叫者。

3.訊息的推送/獲取的整個生命週期中，要把重要事件通知給外部呼叫者，比如：Broker,Zookeeper是否有異常，資料推送/獲取是否成功，如果失敗需要保留失敗記錄便於進行後期資料恢復等。(需要在中介軟體中建立回撥機制通知呼叫者)

4.可對多企業多應用進行平滑移植，移植過程中儘可能保持整體Kafka資料平臺結構的零修改。

四. Kafka資料平臺中介軟體設計

4.1解決方案：

基於以上待完成目標，我們有了以下解決方案。

3.2 實現要點：

KfkProducer(資料生產者)

•KfkProducer物件需要在Web框架中的Application_OnStart()中啟動，常駐程序，只與Broker連線一次，資料傳送過程不能與Broker建立連線。（實踐中發現Kafka的 Broker如果有異常，重啟Broker後Producer不用再次連線即可傳送）

•Web框架可以隨時呼叫推送介面將物件(Object)推送至Broker.

•Object序列化後形成二進位制資訊，並且要保證在Consumer所處框架中能順利還原.

•可傳送多種物件(Object,File ,Byte[]等),簡化外圍框架針對待發送資料所做操作，簡化呼叫介面。

•資料傳送使用Kafka中最新的非同步式資料傳送API，不能由於傳送時間過長或Broker異常等問題阻塞呼叫者。

•需要對整個傳送生命期進行跟蹤反饋異常資訊，若傳送失敗，需要將待發送資料使用回撥機制通知到框架呼叫者。

•詳細測試Broker,或Zookeeper產生異常時，Producer可能會出現的情況。

•在針對多企業多應用中，可依靠Topic進行區分資料主題，這樣可實現多應用部署時框架零修改問題。

KfkConsumer(資料消費者)

•KfkConsumer需要在計算中心內部Web框架中的Application_OnStart()中啟動，常駐程序，只與Broker連線一次，並啟動訊息泵等待訊息到來。（實踐中發現Kafka的 Broker如果有異常，重啟Broker後Consumer不用再次連線即可正常獲取訊息）

•需要定義回撥介面，該回調介面由外圍框架程式註冊處理程式，當資料訊息到來時，Consumer需要把資料傳送至該介面，之後由呼叫者處理。

•呼叫者需要註冊所接受的物件型別，因為Broker中同一Topic下會有各種資料物件(UserInfo,CompanyInfo,ProductInfo...)存在，所以必須提供接收物件的註冊介面，以方便呼叫者有針對性的獲取。

•資料到來時，要針對傳送方序列化的二進位制資訊進行反序列化操作，並能準確還原成原始物件。

•需要對整個接收生命期進行跟蹤反饋異常資訊，若訊息泵停止或異常，需要通知到框架呼叫者。

實現以上要點後，需要將KfkProducer及KfkConsumer物件打包成Jar包的形式，更靈活的部署到企業本地Web框架及計算中心內部Web框架中。

3.3 程式碼實現及分析：

3.3.1 KfkProducer 物件：資料生產者物件，封裝了關於資料傳送的相關功能。

3.3.2 KfkConsumer物件：資料消費者物件，封裝了關於資料接收的相關功能。

3.3.3 KfkMsg物件：資料訊息物件，封裝了資料物件的序列化/反序列化操作，構造多種型別的傳送物件，封裝傳送協議等操作。

3.3.4 SerializeUtils物件：序列化操作工具類，完成在Jar包內部對外部物件的序列化/反序列化基礎從操作。

3.3.5 呼叫者Web框架部署：

KfkProducer部署：

KfkConsumer部署：

五. 未來Kafka中介軟體

目前該中介軟體只完成了初級階段功能，很多功能都不完善不深入，隨著應用業務的拓展及Kafka未來版本功能支援，。以Kafka訊息中介軟體為中心的大資料處理平臺還有很多工去實現。

一般在網際網路中所流動的資料由以下幾種型別：

•需要實時響應的交易資料，使用者提交一個表單，輸入一段內容，這種資料最後是存放在關係資料庫(Oracle, MySQL)中的，有些需要事務支援。

•活動流資料，準實時的，例如頁面訪問量、使用者行為、搜尋情況等。我們可以針對這些資料廣播、排序、個性化推薦、運營監控等。這種資料一般是前端伺服器先寫檔案，然後通過批量的方式把檔案倒到Hadoop(離線資料分析平臺)這種大資料分析器裡面，進行慢慢的分析。

•各個層面程式產生的日誌，例如http的日誌、tomcat的日誌、其他各種程式產生的日誌。這種資料一個是用來監控報警，還有就是用來做分析。

謝謝觀賞！

注：基於全球開源共享理念，本人會分享更多原創及譯文，讓更多的IT人從中受益，與大家一起進步！

基因Cloud 原創，轉發請註明出處

[email protected] (工作繁忙，有事發郵件，QQ不加，非要事勿擾，多謝！)

2015 / 06 / 14