作者 | 葛賢亮，單位：中國(guó)移動(dòng)智慧家庭運(yùn)營(yíng)中心

?Labs 導(dǎo)讀

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

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

1、概念介紹

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

1.1 主要特性

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

1.2 Kafka優(yōu)點(diǎn)

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

1.3 Kafka缺點(diǎn)

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

1.4 主要應(yīng)用場(chǎng)景

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

1.5 Kafka為什么這么快？

Kafka可以輕松支持每秒百萬(wàn)級(jí)的寫(xiě)入請(qǐng)求，超過(guò)了大部分的消息中間件，這種特性使得Kafka在日志處理等海量數(shù)據(jù)場(chǎng)景得到廣泛應(yīng)用。

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

2、架構(gòu)設(shè)計(jì)

圖1 架構(gòu)設(shè)計(jì)

3、核心概念

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

• Kafka Cluster：Kafka集群，由一臺(tái)或多臺(tái)服務(wù)器組成；

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

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

• Consumer：消費(fèi)者，用來(lái)讀取Kafka中的數(shù)據(jù)(Record)進(jìn)行消費(fèi)；

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

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

4、工作流程

圖2 工作流程

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

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

?? ACK應(yīng)答機(jī)制

通過(guò)配置request.required.acks屬性來(lái)配置ACK策略：

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

5、Kafka數(shù)據(jù)存儲(chǔ)設(shè)計(jì)

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

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

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

每個(gè)Partition是一個(gè)有序、不可變的消息序列，新的消息只會(huì)被追加到Partition的末尾。在每個(gè)Partition中，通過(guò)offset（偏移量）標(biāo)識(shí)消息。由此可見(jiàn)，在同一個(gè)Partition內(nèi)消息是有序的，在不同Partition之間，不能保證消息被有序消費(fèi)。

Kafka可以通過(guò)log.retention配置項(xiàng)設(shè)定消息日志在集群內(nèi)的留存時(shí)間，默認(rèn)為168小時(shí)（即7天）。

5.2 Partition結(jié)構(gòu)

Partition在服務(wù)器上是以文件夾形式存在的，每個(gè)Partition文件夾內(nèi)會(huì)有多組Segment文件，每組Segment文件又包含.index、.log、.timeindex三個(gè)文件，其中.log是實(shí)際存儲(chǔ)消息日志的地方，而.index和.timeindex為索引文件，用于檢索消息。

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

Segment對(duì)應(yīng)一個(gè)文件（實(shí)現(xiàn)上對(duì)應(yīng)兩個(gè)文件，一個(gè)數(shù)據(jù)文件，一個(gè)索引文件），一個(gè)Partition對(duì)應(yīng)一個(gè)文件夾，一個(gè)Partition內(nèi)理論上可以包含任意多個(gè)Segment。

如果不引入Segment ，所有消息日志都直接寫(xiě)在Partition文件內(nèi)，會(huì)導(dǎo)致Partition文件一直增大。同時(shí)，在做data purge時(shí)，需要把文件的前面部分給刪除，不符合Kafka文件的順序?qū)憙?yōu)化設(shè)計(jì)方案。引入Segment后，消息日志被分散在多個(gè)Segment中， 每次做data purge，只需要把舊的Segment整個(gè)文件刪除即可，保證了每個(gè)Segment的順序?qū)憽?/p>

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

Partition中的每條消息包含三個(gè)屬性：offset、MessageSize、data，其中offset表示消息在這個(gè)Partition中的偏移量，offset 不是實(shí)際存儲(chǔ)位置，而是邏輯上一個(gè)值，用來(lái)唯一標(biāo)識(shí)Partition內(nèi)的一條消息，相當(dāng)于消息id；MessageSize表示消息內(nèi)容data的大小；data為消息的具體內(nèi)容。 

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

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

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

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

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

6、生產(chǎn)者&消費(fèi)者設(shè)計(jì)

6.1 負(fù)載均衡（Partition會(huì)均衡分布到不同Broker上） 

由于一個(gè)Topic可以有多個(gè)Partition，不同Partition均衡分布在不同的Broker上。基于該特性，生產(chǎn)者可通過(guò)隨機(jī)、輪詢或hash等策略，將消息平均發(fā)送至多個(gè)Broker中，實(shí)現(xiàn)負(fù)載均衡。

6.2 批量發(fā)送

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

6.3 壓縮（Snappy、Gzip、LZ4）

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

7、總結(jié)

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