在現有的工程專案中，經常需要對使用者的實時log進行處理，實時log主要是通過kafka進行記錄，分兩個部分，一部分通過kafka進行切片後儲存在hdfs中，另一部分也可以提供實時的kafka資料，方便實時的處理，因此先簡單介紹一下kafka的基本的工作原理

訊息佇列：

訊息佇列技術是分散式應用間交換資訊的一種技術。訊息佇列可駐留在記憶體或磁碟上, 佇列儲存訊息直到它們被應用程式讀走。通過訊息佇列，應用程式可獨立地執行--它們不需要知道彼此的位置、或在繼續執行前不需要等待接收程式接收此訊息。在分散式計算環境中，為了整合分散式應用，開發者需要對異構網路環境下的分散式應用提供有效的通訊手段。為了管理需要共享的資訊，對應用提供公共的資訊交換機制是重要的。常用的訊息佇列技術是 Message Queue。

Message Queue 的通訊模式

1. 點對點通訊：點對點方式是最為傳統和常見的通訊方式，它支援一對一、一對多、多對多、多對一等多種配置方式，支援樹狀、網狀等多種拓撲結構。

2. 多點廣播：MQ 適用於不同型別的應用。其中重要的，也是正在發展中的是"多點廣播"應用，即能夠將訊息傳送到多個目標站點 (Destination List)。可以使用一條 MQ 指令將單一訊息傳送到多個目標站點，並確保為每一站點可靠地提供資訊。MQ 不僅提供了多點廣播的功能，而且還擁有智慧訊息分發功能，在將一條訊息傳送到同一系統上的多個使用者時，MQ 將訊息的一個複製版本和該系統上接收者的名單傳送到目標 MQ 系統。目標 MQ 系統在本地複製這些訊息，並將它們傳送到名單上的佇列，從而儘可能減少網路的傳輸量。

3. 釋出/訂閱 (Publish/Subscribe) 模式：釋出/訂閱功能使訊息的分發可以突破目的佇列地理指向的限制，使訊息按照特定的主題甚至內容進行分發，使用者或應用程式可以根據主題或內容接收到所需要的訊息。釋出/訂閱功能使得傳送者和接收者之間的耦合關係變得更為鬆散，傳送者不必關心接收者的目的地址，而接收者也不必關心訊息的傳送地址，而只是根據訊息的主題進行訊息的收發。

4. 群集 (Cluster)：為了簡化點對點通訊模式中的系統配置，MQ 提供 Cluster(群集) 的解決方案。群集類似於一個域 (Domain)，群集內部的佇列管理器之間通訊時，不需要兩兩之間建立訊息通道，而是採用群集 (Cluster) 通道與其它成員通訊，從而大大簡化了系統配置。此外，群集中的佇列管理器之間能夠自動進行負載均衡，當某一佇列管理器出現故障時，其它佇列管理器可以接管它的工作，從而大大提高系統的高可靠性。

回頁首

Apache Kafka 原理

Kafka 是一個訊息系統，原本開發自 LinkedIn，用作 LinkedIn 的活動流（Activity Stream）和運營資料處理管道（Pipeline）的基礎。現在它已被多家公司作為多種型別的資料管道和訊息系統使用。活動流資料是幾乎所有站點在對其網站使用情況做報表時都要用到的資料中最常規的部分。活動資料包括頁面訪問量（Page View）、被檢視內容方面的資訊以及搜尋情況等內容。這種資料通常的處理方式是先把各種活動以日誌的形式寫入某種檔案，然後週期性地對這些檔案進行統計分析。運營資料指的是伺服器的效能資料（CPU、IO 使用率、請求時間、服務日誌等等資料)，總的來說，運營資料的統計方法種類繁多。

• Kafka 專用術語

Broker：Kafka 叢集包含一個或多個伺服器，這種伺服器被稱為 broker。

Topic：每條釋出到 Kafka 叢集的訊息都有一個類別，這個類別被稱為 Topic。（物理上不同 Topic 的訊息分開儲存，邏輯上一個 Topic 的訊息雖然保存於一個或多個 broker 上，但使用者只需指定訊息的 Topic 即可生產或消費資料而不必關心資料存於何處）。

Partition：Partition 是物理上的概念，每個 Topic 包含一個或多個 Partition。

Producer：負責釋出訊息到 Kafka broker。

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

Consumer Group：每個 Consumer 屬於一個特定的 Consumer Group（可為每個 Consumer 指定 group name，若不指定 group name 則屬於預設的 group）。

• Kafka 互動流程

Kafka 是一個基於分散式的訊息釋出-訂閱系統，它被設計成快速、可擴充套件的、持久的。與其他訊息釋出-訂閱系統類似，Kafka 在主題當中儲存訊息的資訊。生產者向主題寫入資料，消費者從主題讀取資料。由於 Kafka 的特性是支援分散式，同時也是基於分散式的，所以主題也是可以在多個節點上被分割槽和覆蓋的。

資訊是一個位元組陣列，程式設計師可以在這些位元組陣列中儲存任何物件，支援的資料格式包括 String、JSON、Avro。Kafka 通過給每一個訊息繫結一個鍵值的方式來保證生產者可以把所有的訊息傳送到指定位置。屬於某一個消費者群組的消費者訂閱了一個主題，通過該訂閱消費者可以跨節點地接收所有與該主題相關的訊息，每一個訊息只會傳送給群組中的一個消費者，所有擁有相同鍵值的訊息都會被確保發給這一個消費者。

Kafka 設計中將每一個主題分割槽當作一個具有順序排列的日誌。同處於一個分割槽中的訊息都被設定了一個唯一的偏移量。Kafka 只會保持跟蹤未讀訊息，一旦訊息被置為已讀狀態，Kafka 就不會再去管理它了。Kafka 的生產者負責在訊息佇列中對生產出來的訊息保證一定時間的佔有，消費者負責追蹤每一個主題 (可以理解為一個日誌通道) 的訊息並及時獲取它們。基於這樣的設計，Kafka 可以在訊息佇列中儲存大量的開銷很小的資料，並且支援大量的消費者訂閱。

回頁首

利用 Apache Kafka 系統架構的設計思路

• 示例：網路遊戲

假設我們正在開發一個線上網路遊戲平臺，這個平臺需要支援大量的線上使用者實時操作，玩家在一個虛擬的世界裡通過互相協作的方式一起完成每一個任務。由於遊戲當中允許玩家互相交易金幣、道具，我們必須確保玩家之間的誠信關係，而為了確保玩家之間的誠信及賬戶安全，我們需要對玩家的 IP 地址進行追蹤，當出現一個長期固定 IP 地址忽然之間出現異動情況，我們要能夠預警，同時，如果出現玩家所持有的金幣、道具出現重大變更的情況，也要能夠及時預警。此外，為了讓開發組的資料工程師能夠測試新的演算法，我們要允許這些玩家資料進入到 Hadoop 叢集，即載入這些資料到 Hadoop 叢集裡面。

對於一個實時遊戲，我們必須要做到對儲存在伺服器記憶體中的資料進行快速處理，這樣可以幫助實時地發出預警等各類動作。我們的系統架設擁有多臺伺服器，記憶體中的資料包括了每一個線上玩家近 30 次訪問的各類記錄，包括道具、交易資訊等等，並且這些資料跨伺服器儲存。

我們的伺服器擁有兩個角色：首先是接受使用者發起的動作，例如交易請求，其次是實時地處理使用者發起的交易並根據交易資訊發起必要的預警動作。為了保證快速、實時地處理資料，我們需要在每一臺機器的記憶體中保留歷史交易資訊，這意味著我們必須在伺服器之間傳遞資料，即使接收使用者請求的這臺機器沒有該使用者的交易資訊。為了保證角色的鬆耦合，我們使用 Kafka 在伺服器之間傳遞資訊 (資料)。

• Kafka 特性

Kafka 的幾個特性非常滿足我們的需求：可擴充套件性、資料分割槽、低延遲、處理大量不同消費者的能力。這個案例我們可以配置在 Kafka 中為登陸和交易配置同一個主題。由於 Kafka 支援在單一主題內的排序，而不是跨主題的排序，所以我們為了保證使用者在交易前使用實際的 IP 地址登陸系統，我們採用了同一個主題來儲存登陸資訊和交易資訊。

當用戶登陸或者發起交易動作後，負責接收的伺服器立即發事件給 Kafka。這裡我們採用使用者 id 作為訊息的主鍵，具體事件作為值。這保證了同一個使用者的所有的交易資訊和登陸資訊被髮送到 Kafka 分割槽。每一個事件處理服務被當作一個 Kafka 消費者來執行，所有的消費者被配置到了同一個消費者群組，這樣每一臺伺服器從一些 Kafka 分割槽讀取資料，一個分割槽的所有資料被送到同一個事件處理伺服器 (可以與接收伺服器不同)。當事件處理伺服器從 Kafka 讀取了使用者交易資訊，它可以把該資訊加入到儲存在本地記憶體中的歷史資訊列表裡面，這樣可以保證事件處理伺服器在本地記憶體中呼叫使用者的歷史資訊並做出預警，而不需要額外的網路或磁碟開銷。

圖 1. 遊戲設計圖

>為了多執行緒處理，我們為每一個事件處理伺服器或者每一個核建立了一個分割槽。Kafka 已經在擁有 1 萬個分割槽的叢集裡測試過。

• 切換回 Kafka

上面的例子聽起來有點繞口：首先從遊戲伺服器傳送資訊到 Kafka，然後另一臺遊戲伺服器的消費者從主題中讀取該資訊並處理它。然而，這樣的設計解耦了兩個角色並且允許我們管理每一個角色的各種功能。此外，這種方式不會增加負載到 Kafka。測試結果顯示，即使 3 個結點組成的叢集也可以處理每秒接近百萬級的任務，平均每個任務從註冊到消費耗時 3 毫秒。

上面例子當發現一個事件可疑後，傳送一個預警標誌到一個新的 Kafka 主題，同樣的有一個消費者服務會讀取它，並將資料存入 Hadoop 叢集用於進一步的資料分析。

因為 Kafka 不會追蹤訊息的處理過程及消費者佇列，所以它在消耗極小的前提下可以同時處理數千個消費者。Kafka 甚至可以處理批量級別的消費者，例如每小時喚醒一次一批睡眠的消費者來處理所有的資訊。

Kafka 讓資料存入 Hadoop 叢集變得非常簡單。當擁有多個數據來源和多個數據目的地時，為每一個來源和目的地配對地編寫一個單獨的資料通道會導致混亂髮生。Kafka 幫助 LinkedIn 規範了資料通道格式，並且允許每一個系統獲取資料和寫入資料各一次，這樣極大地減少資料通道的複雜性和操作耗時。

LinkedIn 的架構師 Jay Kreps 說：“我最初是在 2008 年完成鍵值對資料儲存方式後開始的，我的專案是嘗試執行 Hadoop，將我們的一些處理過程移動到 Hadoop 裡面去。我們在這個領域幾乎沒有經驗，花了幾個星期嘗試把資料匯入、匯出，另外一些事件花在了嘗試各種各樣的預測性演算法使用上面，然後，我們開始了漫漫長路”。

• 與 Flume 的區別

Kafka 與 Flume 很多功能確實是重複的。以下是評估兩個系統的一些建議：

1. Kafka 是一個通用型系統。你可以有許多的生產者和消費者分享多個主題。相反地，Flume 被設計成特定用途的工作，特定地向 HDFS 和 HBase 傳送出去。Flume 為了更好地為 HDFS 服務而做了特定的優化，並且與 Hadoop 的安全體系整合在了一起。基於這樣的結論，Hadoop 開發商 Cloudera 推薦如果資料需要被多個應用程式消費的話，推薦使用 Kafka，如果資料只是面向 Hadoop 的，可以使用 Flume。

2. Flume 擁有許多配置的來源 (sources) 和儲存池 (sinks)。然後，Kafka 擁有的是非常小的生產者和消費者環境體系，Kafka 社群並不是非常支援這樣。如果你的資料來源已經確定，不需要額外的編碼，那你可以使用 Flume 提供的 sources 和 sinks，反之，如果你需要準備自己的生產者和消費者，那你需要使用 Kafka。

3. Flume 可以在攔截器裡面實時處理資料。這個特性對於過濾資料非常有用。Kafka 需要一個外部系統幫助處理資料。

4. 無論是 Kafka 或是 Flume，兩個系統都可以保證不丟失資料。然後，Flume 不會複製事件。相應地，即使我們正在使用一個可以信賴的檔案通道，如果 Flume agent 所在的這個節點宕機了，你會失去所有的事件訪問能力直到你修復這個受損的節點。使用 Kafka 的管道特性不會有這樣的問題。

5. Flume 和 Kafka 可以一起工作的。如果你需要把流式資料從 Kafka 轉移到 Hadoop，可以使用 Flume 代理 (agent)，將 kafka 當作一個來源 (source)，這樣可以從 Kafka 讀取資料到 Hadoop。你不需要去開發自己的消費者，你可以使用 Flume 與 Hadoop、HBase 相結合的特性，使用 Cloudera Manager 平臺監控消費者，並且通過增加過濾器的方式處理資料。