作者：海源、仕祿、肖恩等

Kafka在美團數據平臺承擔著統一的數據緩存和分發的角色，隨著數據量的增長，集群規模的擴大，Kafka面臨的挑戰也愈發嚴峻。

1. 現狀和挑戰

1.1 現狀

Kafka是一個開源的流處理平臺，我們首先了解一下Kafka在美團數據平臺的現狀。

圖1-1 Kafka在美團數據平臺的現狀如圖1-1所示，藍色部分描述了Kafka在數據平臺定位為流存儲層。主要的職責是做數據的緩存和分發，它會將收集到的日志分發到不同的數據系統里，這些日志來源于系統日志、客戶端日志以及業務數據庫。下游的數據消費系統包括通過ODS入倉提供離線計算使用、直接供實時計算使用、通過DataLink同步到日志中心，以及做OLAP分析使用。

Kafka在美團的集群規模總體機器數已經超過了15000+臺，單集群的最大機器數也已經到了2000+臺。在數據規模上，天級消息量已經超過了30+P，天級消息量峰值也達到了4+億/秒。不過隨著集群規模的增大，數據量的增長，Kafka面臨的挑戰也愈發嚴峻，下面講一下具體的挑戰都有哪些。

1.2 挑戰

圖1-2 Kafka在美團數據平臺面臨的挑戰

如圖1-2所示，具體的挑戰可以概括為兩部分：

第一部分是慢節點影響讀寫，這里慢節點參考了HDFS的一個概念，具體定義指的是讀寫延遲TP99大于300ms的Broker。造成慢節點的原因有三個：

1. 集群負載不均衡會導致局部熱點，就是整個集群的磁盤空間很充裕或者ioutil很低，但部分磁盤即將寫滿或者ioutil打滿。
2. PageCache容量，比如說，80GB的PageCache在170MB/s的寫入量下僅能緩存8分鐘的數據量。那么如果消費的數據是8分鐘前的數據，就有可能觸發慢速的磁盤訪問。
3. Consumer客戶端的線程模型缺陷會導致端到端延時指標失真。例如當Consumer消費的多個分區處于同一Broker時，TP90可能小于100ms，但是當多個分區處于不同Broker時，TP90可能會大于1000ms。

第二部分是大規模集群管理的復雜性，具體表現有4類問題：

1. 不同Topic之間會相互影響，個別Topic的流量突增，或者個別消費者的回溯讀會影響整體集群的穩定性。
2. Kafka原生的Broker粒度指標不夠健全，導致問題定位和根因分析困難。
3. 故障感知不及時，處理成本較高。
4. Rack級別的故障會造成部分分區不可用。

2. 讀寫延遲優化

接下來我們先介紹一下針對讀寫延遲問題，美團數據平臺做了哪些優化。首先從宏觀層面，我們將受影響因素分為應用層和系統層，然后詳細介紹應用層和系統層存在的問題，并給出對應的解決方案，包括流水線加速、Fetcher隔離、遷移取消和Cgroup資源隔離等，下面具體介紹各種優化方案的實現。

2.1 概覽

圖2-1 Kafka讀寫延遲優化概覽

圖2-1是針對讀寫延遲碰到的問題以及對應優化方案的概覽圖。我們把受影響的因素分為應用層和系統層。

應用層主要包括3類問題：

1）Broker端負載不均衡，例如磁盤使用率不均衡、ioutil不均衡等問題。個別磁盤負載升高影響整個Broker的請求受到影響。

2）Broker的數據遷移存在效率問題和資源競爭問題。具體來講，包括以下3個層面：

• 遷移只能按批次串行提交，每個批次可能存在少量分區遷移緩慢，無法提交下個批次，導致遷移效率受影響。
• 遷移一般在夜間執行，如果遷移拖到了午高峰還未完成，可能會顯著影響讀寫請求。
• 遷移請求和實時拉取存在共用Fetcher線程的問題導致分區遷移請求可能會影響實時消費請求。

3）Consumer端單線程模型存在缺陷導致運維指標失真，并且單Consumer消費的分區數不受限制，消費能力不足就無法跟上實時最新的數據，當消費的分區數增多時可能會引起回溯讀。

系統層也主要包括3類問題：

1）PageCache污染。Kafka利用內核層提供的ZeroCopy技術提升性能，但是內核層無法區分實時讀寫請求和回溯讀請求，導致磁盤讀可能污染PageCache，影響實時讀寫。

2）HDD在隨機讀寫負載下性能差。HDD對于順序讀寫友好，但是面對混合負載場景下的隨機讀寫，性能顯著下降。

3）CPU和內存等系統資源在混部場景下的資源競爭問題。在美團大數據平臺，為了提高資源的利用率，IO密集型的服務（比如Kafka）會和CPU密集型的服務（比如實時計算作業）混布，混布存在資源競爭，影響讀寫延遲。

以上提到的問題，我們采取了針對性的策略。比如應用層的磁盤均衡、遷移流水線加速、支持遷移取消和Consumer異步化等。系統層的Raid卡加速、Cgroup隔離優化等。此外，針對HDD隨機讀寫性能不足的問題，我們還設計并實現了基于SSD的緩存架構。

2.2 應用層

① 磁盤均衡

圖2-2 Kafka應用層磁盤均衡

磁盤熱點導致兩個問題：

• 實時讀寫延遲變高，比如說TP99請求處理時間超過300ms可能會導致實時作業發生消費延遲問題，數據收集擁堵問題等。
• 集群整體利用率不足，雖然集群容量非常充裕，但是部分磁盤已經寫滿，這個時候甚至會導致某些分區停止服務。

針對這兩個問題，我們采用了基于空閑磁盤優先的分區遷移計劃，整個計劃分為3步，由組件Rebalancer統籌管理：

1. 生成遷移計劃。Rebalancer通過目標磁盤使用率和當前磁盤使用率（通過Kafka Monitor上報）持續生成具體的分區遷移計劃。
2. 提交遷移計劃。Rebalancer向Zookeeper的Reassign節點提交剛才生成的遷移計劃，Kafka的Controller收到這個Reassign事件之后會向整個Kafka Broker集群提交Reassign事件。
3. 檢查遷移計劃。Kafka Broker負責具體執行數據遷移任務，Rebalancer負責檢查任務進展。

如圖2-2所示，每塊Disk持有3個分區是一個相對均衡的狀態，如果部分Disk持有4個分區，比如Broker1-Disk1和Broker4-Disk4；部分Disk持有2個分區，比如Broker2-Disk2，Broker3-Disk3，Reblanacer就會將Broker1-Disk1和Broker4-Disk4上多余的分區分別遷移到Broker2-Disk2和Broker3-Disk3，最終盡可能地保證整體磁盤利用率均衡。

② 遷移優化

雖然基于空閑磁盤優先的分區遷移實現了磁盤均衡，但是遷移本身仍然存在效率問題和資源競爭問題。接下來，我們會詳細描述我們采取的針對性策略。

1. 采取流水線加速策略優化遷移緩慢引起的遷移效率問題。
2. 支持遷移取消解決長尾分區遷移緩慢引起的讀寫請求受影響問題。
3. 采取Fetcher隔離緩解數據遷移請求和實時讀寫請求共用Fetcher線程的問題。

優化一，流水線加速

圖2-3 流水線加速

如圖2-3所示，箭頭以上原生Kafka版本只支持按批提交，比如說一批提交了四個分區，當TP4這個分區一直卡著無法完成的時候，后續所有分區都無法繼續進行。采用流水線加速之后，即使TP4這個分區還沒有完成，可以繼續提交新的分區。在相同的時間內，原有的方案受阻于TP4沒有完成，后續所有分區都沒辦法完成，在新的方案中，TP4分區已經遷移到TP11分區了。圖中虛線代表了一個無序的時間窗口，主要用于控制并發，目的是為了和原有的按組提交的個數保持一致，避免過多的遷移影響讀寫請求服務。

優化二，遷移取消

圖2-4-1 遷移問題

如圖2-4-1所示，箭頭左側描述了因為遷移影響的三種線上類型。第一種是因為遷移會觸發最舊讀，同步大量的數據，在這個過程中會首先將數據回刷到PageCache上引起PageCache污染，導致某個實時讀的分區發生Cache Miss，觸發磁盤度進而影響讀寫請求；第二種是當存在某些異常節點導致遷移Hang住時，部分運維操作無法執行，比如流量上漲觸發的Topic自動擴分區。因為在Kafka遷移過程中這類運維操作被禁止執行。第三種和第二種類似，它的主要問題是當目標節點Crash，Topic擴分區也無法完成，用戶可能一直忍受讀寫請求受影響。

圖2-4-2 遷移取消

針對上面提到的3種問題，我們支持了遷移取消功能。管理員可以調用遷移取消命令，中斷正在遷移的分區，針對第一種場景，PageCache就不會被污染，實時讀得以保證；在第二、三種場景中，因為遷移取消，擴分區得以完成。遷移取消會刪除未完成遷移的分區，刪除可能會導致磁盤IO出現瓶頸影響讀寫，因此我們通過支持平滑刪除避免大量刪除引起的性能問題。

優化三，Fetcher隔離

圖2-5 Fetcher隔離

如圖2-5，綠色代表實時讀，紅色代表延時讀。當某一個Follower的實時讀和延時讀共享同一個Fetcher時，延時讀會影響實時讀。因為每一次延時讀的數據量是顯著大于實時讀的，而且延時讀容易觸發磁盤讀，可能數據已經不在PageCache中了，顯著地拖慢了Fetcher的拉取效率。針對這種問題，我們實施的策略叫Fetcher隔離。也就是說所有ISR的Follower共享Fetcher，所有非ISR的Follower共享Fetcher，這樣就能保證所有ISR中的實時讀不會被非ISR的回溯讀所影響。

③ Consumer異步化

圖2-6 Kafka-Broker分階段延時統計模型

在講述Consumer異步化前，需要解釋下圖2-6展示的Kafka-Broker分階段延時統計模型。Kafka-Broker端是一個典型的事件驅動架構，各組件通過隊列通信。請求在不同組件流轉時，會依次記錄時間戳，最終就可以統計出請求在不同階段的執行耗時。

具體來說，當一個Kafka的Producer或Consumer請求進入到Kafka-Broker時，Processor組件將請求寫入RequestQueue，RequestHandler從RequestQueue拉取請求進行處理，在RequestQueue中的等待時間是RequestQueueTime，RequestHandler具體的執行時間是LocalTime。當RequestHandler執行完畢后會將請求傳遞給DelayedPurgatory組件中，該組件是一個延時隊列。

當觸發某一個延時條件完成了以后會把請求寫到ResponseQueue中，在DelayedPurgatory隊列持續的時間為RemoteTime，Processor會不斷的從ResponseQueue中將數據拉取出來發往客戶端，標紅的ResponseTime是可能會被客戶端影響的，因為如果客戶端接收能力不足，那么ResponseTime就會一直持續增加。從Kafka-Broker的視角，每一次請求總的耗時時RequestTotalTime，包含了剛才所有流程分階段計時總和。

圖2-7 Consumer異步化

ResponseTime持續增加的主要問題是因為Kafka原生Consumer基于NIO的單線程模型存在缺陷。如圖2-7所示，在Phase1，User首先發起Poll請求，Kafka-Client會同時向Broker1、Broker2和Broker3發送請求，Broker1的數據先就緒時，Kafka Client將數據寫入CompleteQueue，并立即返回，而不是繼續拉取Broker2和Broker3的數據。后續的Poll請求會直接從CompleteQueue中讀取數據，然后直接返回，直到CompleteQueue被清空。在CompleteQueue被清空之前，即使Broker2和Broker3的端的數據已經就緒，也不會得到及時拉取。如圖中Phase2，因為單線程模型存在缺陷導致WaitFetch這部分時長變大，導致Kafka-Broker的RespnseTime延時指標不斷升高，帶來的問題是無法對服務端的處理瓶頸進行精準的監控與細分。

圖2-8 引入異步拉取線程

針對這個問題，我們的改進是引入異步拉取線程。異步拉取線程會及時地拉取就緒的數據，避免服務端延時指標受影響，而且原生Kafka并沒有限制同時拉取的分區數，我們在這里做了限速，避免GC和OOM的發生。異步線程在后臺持續不斷地拉取數據并放到CompleteQueue中。

2.3 系統層

① Raid卡加速

圖2-9 Raid卡加速

HDD存在隨機寫性能不足的問題，表現為延時升高，吞吐降低。針對這個問題我們引入了Raid卡加速。Raid卡自帶緩存，與PageCache類似，在Raid這一層會把數據Merge成更大的Block寫入Disk，更加充分利用順序寫HDD的帶寬，借助Raid卡保證了隨機寫性能。

② Cgroup隔離優化

圖2-10 Cgroup隔離

為了提高資源利用率，美團數據平臺將IO密集型應用和CPU密集型應用混合部署。IO密集型應用在這里指的就是Kafka，CPU密集型應用在這里指的是Flink和Storm。但是原有的隔離策略存在兩個問題：首先是物理核本身會存在資源競爭，在同一個物理核下，共享的L1Cache和L2Cache都存在競爭，當實時平臺CPU飆升時會導致Kafka讀寫延時受到影響；其次，Kafka的HT跨NUMA，增加內存訪問耗時，如圖2-10所示，跨NUMA節點是通過QPI去做遠程訪問，而這個遠程訪問的耗時是40ns。

針對這兩個問題，我們改進了隔離策略，針對物理核的資源競爭，我們新的混布策略保證Kafka獨占物理核，也就是說在新的隔離策略中，不存在同一個物理核被Kafka和Flink同時使用；然后是保證Kafka的所有超線程處于同一側的NUMA，避免Kafka跨NUMA帶來的訪問延時。通過新的隔離策略，Kafka的讀寫延時不再受Flink CPU飆升的影響。

2.4 混合層-SSD新緩存架構

圖2-11 Page污染引起的性能問題

背景和挑戰

Kafka利用操作系統提供的ZeroCopy技術處理數據讀取請求，PageCache容量充裕時數據直接從PageCache拷貝到網卡，有效降低了讀取延時。但是實際上，PageCache的容量往往是不足的，因為它不會超過一個機器的內存。容量不足時，ZeroCopy就會觸發磁盤讀，磁盤讀不僅顯著變慢，還會污染PageCache影響其他讀寫。

如圖2-11中左半部分所示，當一個延遲消費者去拉取數據時，發現PageCache中沒有它想要的數據，這個時候就會觸發磁盤讀。磁盤讀后會將數據回寫到PageCache，導致PageCache污染，延遲消費者消費延遲變慢的同時也會導致另一個實時消費受影響。因為對于實時消費而言，它一直讀的是最新的數據，最新的數據按正常來說時不應該觸發磁盤讀的。

選型和決策

針對這個問題，我們這邊在做方案選型時提供了兩種方案：

方案一，讀磁盤時不回寫PageCache，比如使用DirectIO，不過Java并不支持；

方案二，在內存和HDD之間引入中間層，比如SSD。眾所周知，SSD和HDD相比具備良好的隨機讀寫能力，非常適合我們的使用場景。針對SSD的方案我們也有兩種選型：

方案一，可以基于操作系統的內核實現，這種方案SSD與HDD存儲空間按照固定大小分塊，并且SSD與HDD建立映射關系，同時會基于數據局部性原理，Cache Miss后數據會按LRU和LFU替換SSD中部分數據，業界典型方案包括OpenCAS和FlashCache。其優勢是數據路由對應用層透明，對應用代碼改動量小，并且社區活躍可用性好；但是問題在于局部性原理并不滿足Kafka的讀寫特性，而且緩存空間污染問題并未得到根本解決，因為它會根據LRU和LFU去替換SSD中的部分數據。

方案二，基于Kafka的應用層去實現，具體就是Kafka的數據按照時間維度存儲在不同設備上，對于近實時數據直接放在SSD上，針對較為久遠的數據直接放在HDD上，然后Leader直接根據Offset從對應設備讀取數據。這種方案的優勢是它的緩存策略充分考慮了Kafka的讀寫特性，確保近實時的數據消費請求全部落在SSD上，保證這部分請求處理的低延遲，同時從HDD讀取的數據不回刷到SSD防止緩存污染，同時由于每個日志段都有唯一明確的狀態，因此每次請求目的明確，不存在因Cache Miss帶來的額外性能開銷。同時劣勢也很明顯，需要在Server端代碼上進行改進，涉及的開發以及測試的工作量較大。

圖2-13 KafkaSSD新緩存架構

具體實現

下面來介紹一下SSD新緩存架構的具體實現。

1. 首先新的緩存架構會將Log內的多個Segment按時間維度存儲在不同的存儲設備上，如圖2-14中的紅圈1，新緩存架構數據會有三種典型狀態，一種叫Only Cache，指的是數據剛寫進SSD，還未同步到HDD上；第2個是Cached，指數據既同步到了HDD也有一部分緩存在SSD上；第三種類型叫WithoutCache，指的是同步到了HDD但是SSD中已經沒有緩存了。
2. 然后后臺異步線程持續地將SSD數據同步到HDD上。
3. 隨著SSD的持續寫入，當存儲空間達到閾值后，會按時間順序刪除距當前時間最久的數據，因為SSD的數據空間有限。
4. 副本可根據可用性要求靈活開啟是否寫入SSD。
5. 從HDD讀取的數據是不會回刷到SSD上的，防止緩存污染。

圖2-14 SSD新緩存架構細節優化

細節優化

介紹了具體實現之后，再來看一下細節優化。

1. 首先是關于日志段同步，就是剛才說到的Segment，只同步Inactive的日志段，Inactive指的是現在并沒有在寫的日志段，低成本解決數據一致性問題。
2. 其次是做同步限速優化，在SSD向HDD同步時是需要限速的，同時保護了兩種設備，不會影響其他IO請求的處理。

3. 大規模集群管理優化

3.1 隔離策略

美團大數據平臺的Kafka服務于多個業務，這些業務的Topic混布在一起的話，很有可能造成不同業務的不同Topic之間相互影響。此外，如果Controller節點同時承擔數據讀寫請求，當負載明顯變高時，Controller可能無法及時控制類請求，例如元數據變更請求，最終可能會造成整個集群發生故障。針對這些相互影響的問題，我們從業務、角色和優先級三個維度來做隔離優化。

圖3-1 隔離優化

• 第一點是業務隔離，如圖3-1所示，每一個大的業務會有一個獨立的Kafka集群，比如外賣、到店、優選。
• 第二點是分角色隔離，這里Kafka的Broker和Controller以及它們依賴的組件Zookeeper是部署在不同機器上的，避免之間相互影響。
• 第三點是分優先級，有的業務Topic可用性等級特別高，那么我們就可以給它劃分到VIP集群，給它更多的資源冗余去保證其可用性。

3.2 全鏈路監控

隨著集群規模增長，集群管理碰到了一系列問題，主要包括兩方面：

• Broker端延時指標無法及時反應用戶問題。

1. 隨著請求量的增長，Kafka當前提供的Broker端粒度的TP99甚至TP999延時指標都可能無法反應長尾延時。
2. Broker端的延時指標不是端到端指標，可能無法反應用戶的真實問題。

• 故障感知和處理不及時。

圖3-2 全鏈路監控

針對這兩個問題，我們采取的策略是全鏈路監控。全鏈路監控收集和監控Kafka核心組件的指標和日志。全鏈路監控架構如圖3-2所示。當某一個客戶端讀寫請求變慢時，我們通過全鏈路監控可以快速定位到具體慢在哪個環節，全鏈路指標監控如圖3-3所示。

圖3-3 全鏈路指標監控

圖3-4是一個根據全鏈路指標定位請求瓶頸的示例，可以看出服務端RemoteTime占比最高，這說明耗時主要花費在數據復制。日志和指標的解析服務可以自動實時感知故障和慢節點，大部分的故障（內存、磁盤、Raid卡以及網卡等）和慢節點都已經支持自動化處理，還有一類故障是計劃外的故障，比如分區多個副本掛掉導致的不可用，遷移Hang住以及非預期的錯誤日志等，需要人工介入處理。

圖3-4 全鏈路監控指標示例

3.3 服務生命周期管理

圖3-5 服務生命周期管理

美團線上Kafka的服務器規模在萬級別，隨著服務規模的增長，我們對服務和機器本身的管理，也在不斷迭代。我們的自動化運維系統能夠處理大部分的機器故障和服務慢節點，但對于機器和服務本身的管理是割裂的，導致存在兩類問題：

1. 狀態語義存在歧義，無法真實反映系統狀態，往往需要借助日志和指標去找到真實系統是否健康或者異常。
2. 狀態不全面，異常Case需人工介入處理，誤操作風險極大。

為了解決這兩類問題，我們引入了生命周期管理機制，確保能夠真實反映系統狀態。生命周期管理指的是從服務開始運行到機器報廢停止服務的全流程管理，并且做到了服務狀態和機器狀態聯動，無需人工同步變更。而且新的生命周期管理機制的狀態變更由特定的自動化運維觸發，禁止人工變更。

3.4 TOR容災

圖3-6 TOR容災挑戰

我們從工程實現的角度，歸納總結了當前主流圖神經網絡模型的基本范式，實現一套通用框架，以期涵蓋多種GNN模型。以下按照圖的類型（同質圖、異質圖和動態圖）分別討論。

圖3-7 TOR容災

TOR容災保證同一個分區的不同副本不在同一個Rack下，如圖3-7所示，即使Rack1整個發生故障，也能保證所有分區可用。

4 未來展望

過去一段時間，我們圍繞降低服務端的讀寫延遲做了大量的優化，但是在服務高可用方面，依然有一些工作需要完成。未來一段時間，我們會將重心放在提升魯棒性和通過各種粒度的隔離機制縮小故障域。比如，讓客戶端主動對一些故障節點進行避讓，在服務端通過多隊列的方式隔離異常請求，支持服務端熱下盤，網絡層主動反壓與限流等等。

另外，隨著美團實時計算業務整體的發展，實時計算引擎（典型如Flink）和流存儲引擎（典型如Kafka）混合部署的模式越來越難以滿足業務的需求。因此，我們需要在保持當前成本不變的情況下對Kafka進行獨立部署。這就意味著需要用更少的機器（在我們的業務模式下，用原來1/4的機器）來承載不變的業務流量。如何在保障服務穩定的情況下，用更少的機器扛起業務請求，也是我們面臨的挑戰之一。最后，隨著云原生趨勢的來臨，我們也在探索流存儲服務的上云之路。

5 作者簡介

海源、仕祿、肖恩、鴻洛、啟帆、胡榮、李杰等，均來自美團數據科學與平臺部。