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

?Labs 導讀

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

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

1、概念介紹

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

1.1 主要特性

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

1.2 Kafka優(yōu)點

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

1.3 Kafka缺點

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

1.4 主要應用場景

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

1.5 Kafka為什么這么快？

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

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

2、架構設計

圖1 架構設計

3、核心概念

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

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

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

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

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

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

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

4、工作流程

圖2 工作流程

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

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

?? ACK應答機制

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

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

5、Kafka數(shù)據(jù)存儲設計

5.1 Topic和數(shù)據(jù)日志

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

圖3 Topic和數(shù)據(jù)日志-分區(qū)數(shù)據(jù)日志文件結構

每個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對應一個文件（實現(xiàn)上對應兩個文件，一個數(shù)據(jù)文件，一個索引文件），一個Partition對應一個文件夾，一個Partition內理論上可以包含任意多個Segment。

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

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

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

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

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

5.5 數(shù)據(jù)文件索引（分段索引、稀疏存儲）

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

圖4 數(shù)據(jù)文件索引-稀疏索引

6、生產者&消費者設計

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

由于一個Topic可以有多個Partition，不同Partition均衡分布在不同的Broker上?；谠撎匦裕a者可通過隨機、輪詢或hash等策略，將消息平均發(fā)送至多個Broker中，實現(xiàn)負載均衡。

6.2 批量發(fā)送

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

6.3 壓縮（Snappy、Gzip、LZ4）

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

7、總結

Kafka憑借其架構與性能優(yōu)勢，愈來愈受到眾廠商的青睞。依托其完善的社區(qū)環(huán)境，Kafka構建了龐大而成熟的生態(tài)，已成為大數(shù)據(jù)及流計算領域中至關重要的一環(huán)。