作者 | 葛賢亮，單位：中國移動智慧家庭運營中心

?Labs 導讀

近年來，互聯網技術發展迅猛，各行各業的信息量急劇膨脹。隨著云計算和算力網絡時代的到來，消息中間件在國內許多行業的關鍵應用中越來越受到重視。在高并發分布式場景下，合理地利用消息中間件往往能起到突破性能瓶頸與化繁為簡的效果。

前期分別從“作用”與“協議”、“傳輸模式”與“消費模式”對消息中間件技術做了簡要的介紹。本期從消息中間件產品角度介紹主流方案的設計與實現。

1、概念介紹

Apache Kafka是一種高吞吐量、分布式、多副本、基于發布/訂閱的消息系統，最初由LinkedIn公司開發，使用Scala語言編寫，目前是Apache的開源項目。Kafka已成為事件流處理動態數據的事實標準。

1.1 主要特性

• 高吞吐量、低延遲：kafka每秒可以處理幾十萬條消息，它的延遲最低只有幾毫秒；
• 可擴展性：kafka集群支持熱擴展，數據遷移、擴容對用戶透明；
• 持久性、可靠性：消息被持久化到本地磁盤，并且支持數據備份防止數據丟失；
• 容錯性：允許集群中節點失敗（若副本數量為n，則允許n-1個節點失敗）；
• 高并發：支持數千個客戶端同時讀寫；
• 分布式架構：Broker、Producer和Consumer都原生自動支持分布式，自動實現負載均衡；
• 支持同步和異步復制兩種高可用機制；
• 支持數據批量發送和拉取；
• 零拷貝技術(zero-copy)：減少 IO 操作步驟，提高系統吞吐量；
• 其他特性：豐富的消息拉取模型、高效訂閱者水平擴展、實時的消息訂閱、億級的消息堆積能力、定期刪除機制。

1.2 Kafka優點

• 客戶端多語言支持：支持Java、.Net、PHP、Ruby、Python、Go等多種語言；
• 高性能：單機寫入TPS約在100萬條/秒，消息大小10個字節；
• 分布式架構，并有replica機制，擁有較高的可用性和可靠性，理論上支持消息無限堆積；
• 支持批處理操作；
• 消費者采用Pull方式獲取消息。單分區內消息有序，通過控制能夠保證所有消息被消費且僅被消費一次；
• 在日志領域比較成熟，被多家公司和多個開源項目使用。

1.3 Kafka缺點

• Kafka單機超過64個分區時，load時會發生明顯的飆高現象。隊列越多，負載越高，發送消息響應時間變長；
• 使用短輪詢方式，實時性取決于輪詢間隔時間，對于不能批處理的消息，需考慮消費線程執行效率；
• 需要引入ZooKeeper，部署成本相比其他MQ較高；
• 不能保證消息100%到達，不支持事務消息。

1.4 主要應用場景

• 消息系統：分布式消息系統，解耦生產者和消費者；
• 日志收集：Kafka常與ELK(Logstash、ElasticSearch、Kibana)一起作為業務系統日志收集方案；
• 業務埋點：對于可靠性要求不那么高的埋點數據（如瀏覽網頁、點擊、跳轉等），可使用Kafka進行傳輸，消費者端收到消息后可根據需求做實時監控分析或裝載到hadoop、數據倉庫中做離線分析和挖掘；
• 運營指標：Kafka可用來傳輸運營監控數據，以便統一歸集分析、集中反饋；
• 流式處理：Kafka提供了完整的流式處理類庫，可以很方便的被集成至應用程序中，為流式處理框架（Flink、Spark、Storm等）提供可靠的數據來源。

1.5 Kafka為什么這么快？

Kafka可以輕松支持每秒百萬級的寫入請求，超過了大部分的消息中間件，這種特性使得Kafka在日志處理等海量數據場景得到廣泛應用。

• 并行處理：Kafka引入了Partition（分區）的概念，每個Topic 可包含一個或多個Partition，不同Partition可位于不同節點中，從而實現多磁盤并發讀寫；
• 順序讀寫：Kafka中每個Partition是一個有序、不可變的消息序列，新的消息只會被追加到Partition的末尾，而一個Partition又被分為多個Segment，清除舊數據時可直接刪除Segment文件，避免隨機寫；
• 頁緩存(Page Cache)：Kafka使用頁緩存技術減少I/O操作次數，即使Kafka進程重啟數據也不會丟失（機器宕機時，頁緩存內的數據未及時寫入磁盤會導致數據丟失，同步刷盤可以規避該問題，但會影響性能，默認使用異步刷盤機制）；
• 零拷貝技術：Kafka使用零拷貝技術，避免數據在內核空間的緩沖區和用戶空間的緩沖區之間進行拷貝；
• 批處理：Kafka支持批處理操作，以減少網絡I/O操作；
• 數據壓縮：Kafka支持Snappy、Gzip、LZ4等算法對數據進行壓縮傳輸。

2、架構設計

圖1 架構設計

3、核心概念

• Producer：生產者，用來向Kafka Broker中發送數據(Record)；

• Kafka Cluster：Kafka集群，由一臺或多臺服務器組成；

• Broker：Broker是指部署了Kafka實例的服務器節點，每個服務器上可安裝一個或多個Kafka實例。每個Kafka集群內的Broker都有一個不重復的編號（如broker-0、broker-1等）；
• Topic：消息主題，用來區分不同類型信息。在每個Broker上可以創建多個Topic；
• Partition：Topic的分區，每個Topic可以有一個或多個 Partition（分區），分區可實現負載均衡，支持并發寫入讀取，提高Kafka的吞吐量。一個分區內的數據只能被一個線程消費；
• Replication：每一個分區可以有多個副本。當主分區(Leader)故障的時候會選擇一個副本(Follower)成為新的Leader。在Kafka中副本的默認最大數量是10個，且副本的數量不能大于Broker的數量，Follower和Leader必須分布在不同的機器上，同一機器上同一分區只能存放一個副本（包括自己）。

• Record：消息記錄。每個Record包含了key、value和 timestamp；

• Consumer：消費者，用來讀取Kafka中的數據(Record)進行消費；

• Consumer Group：消費者組，一個消費者組可以包含一個或多個消費者。在Kafka的設計中一個分區內的數據只能被消費者組中的某一個消費者消費，同一個消費者組的消費者可以消費某個Topic不同分區的數據；

• Segment：實際存儲消息的片段；一個Partition在物理上由一個或者多個Segment構成，每個Segment中保存真實的消息數據。

4、工作流程

圖2 工作流程

Kafka一般工作流程如下（根據ACK應答策略會存在部分差異）：

• 生產者與Leader直接交互，先從集群獲取Topic對應分區的Leader元數據；
• 獲取到Leader分區元數據后進行消息發送；
• Kafka Broker對應的Leader分區收到消息后寫入文件進行持久化；
• Follower拉取Leader消息，進行數據同步；
• Follower完成消息拉取后給Leader回復ACK確認；
• Leader和Follower分區完成數據同步后，Leader分區給生產者回復ACK確認。

?? ACK應答機制

通過配置request.required.acks屬性來配置ACK策略：

• 0代表生產者往集群發送數據不需要等待集群的返回，不確保消息是否發送成功。安全性最低但是效率最高。
• 1（默認）代表生產者往集群發送數據只要Leader應答就可以發送下一條，只確保Leader發送成功（Leader不需要等待Follower完成數據同步即返回生產者ACK確認）。
• all代表生產者往集群發送數據需要所有的Follower都完成數據同步才會發送下一條，確保Leader發送成功和所有的副本都完成備份。安全性最高，但是效率最低。

5、Kafka數據存儲設計

5.1 Topic和數據日志

Topic是同一類別的消息記錄(Record)的集合。在Kafka中，一個Topic又可以被劃分成多個Partition，分區數據日志文件結構如下：

圖3 Topic和數據日志-分區數據日志文件結構

每個Partition是一個有序、不可變的消息序列，新的消息只會被追加到Partition的末尾。在每個Partition中，通過offset（偏移量）標識消息。由此可見，在同一個Partition內消息是有序的，在不同Partition之間，不能保證消息被有序消費。

Kafka可以通過log.retention配置項設定消息日志在集群內的留存時間，默認為168小時（即7天）。

5.2 Partition結構

Partition在服務器上是以文件夾形式存在的，每個Partition文件夾內會有多組Segment文件，每組Segment文件又包含.index、.log、.timeindex三個文件，其中.log是實際存儲消息日志的地方，而.index和.timeindex為索引文件，用于檢索消息。

Q：為什么有了Partition還要有Segment？

Segment對應一個文件（實現上對應兩個文件，一個數據文件，一個索引文件），一個Partition對應一個文件夾，一個Partition內理論上可以包含任意多個Segment。

如果不引入Segment ，所有消息日志都直接寫在Partition文件內，會導致Partition文件一直增大。同時，在做data purge時，需要把文件的前面部分給刪除，不符合Kafka文件的順序寫優化設計方案。引入Segment后，消息日志被分散在多個Segment中， 每次做data purge，只需要把舊的Segment整個文件刪除即可，保證了每個Segment的順序寫。

5.3 Partition的數據文件(offset、MessageSize、data)

Partition中的每條消息包含三個屬性：offset、MessageSize、data，其中offset表示消息在這個Partition中的偏移量，offset 不是實際存儲位置，而是邏輯上一個值，用來唯一標識Partition內的一條消息，相當于消息id；MessageSize表示消息內容data的大小；data為消息的具體內容。 

5.4 數據文件分段Segment（順序讀寫、分段命令、二分查找）

Partition物理上由多個Segment文件組成，每個Segment大小相等（大致）。每個Segment數據文件以該段中最小的offset命名，文件擴展名為.log。這樣在查找指定offset消息的時候，用二分查找法可以快速定位到該消息在哪個Segment數據文件中。 

5.5 數據文件索引（分段索引、稀疏存儲）

Kafka為每個Segment建立了索引文件（文件名與數據文件一致，擴展名為.index）。具體是采用稀疏索引方式，每隔一定字節的數據建立一條索引。這樣做的好處是減少了索引體積，以便保留在內存中；壞處是查詢時命中需要消耗更多時間（相對）。

圖4 數據文件索引-稀疏索引

6、生產者&消費者設計

6.1 負載均衡（Partition會均衡分布到不同Broker上） 

由于一個Topic可以有多個Partition，不同Partition均衡分布在不同的Broker上。基于該特性，生產者可通過隨機、輪詢或hash等策略，將消息平均發送至多個Broker中，實現負載均衡。

6.2 批量發送

生產者在本地內存中進行消息聚合，以單次請求發送批量數據的方式，減少網絡I/O操作（副作用是一定程度上會影響消息實時性，以時延換取吞吐量）。

6.3 壓縮（Snappy、Gzip、LZ4）

生產者通過Snappy、Gzip、LZ4等算法對數據進行壓縮傳輸，減少傳輸數據量，減輕網絡壓力（以CPU資源換取網絡時延的降低）。

7、總結

Kafka憑借其架構與性能優勢，愈來愈受到眾廠商的青睞。依托其完善的社區環境，Kafka構建了龐大而成熟的生態，已成為大數據及流計算領域中至關重要的一環。