更多技術交流、求職機會，歡迎關註位元組跳動數據平臺微信公眾號，回覆【1】進入官方交流群摘要位元組數據中台DataLeap的Data Catalog系統通過接收MQ中的近實時消息來同步部分元數據。Apache Atlas對於實時消息的消費處理不滿足性能要求，內部使用Flink任務的處理方案在ToB場景 ...

更多技術交流、求職機會，歡迎關註位元組跳動數據平臺微信公眾號，回覆【1】進入官方交流群

摘要

位元組數據中台DataLeap的Data Catalog系統通過接收MQ中的近實時消息來同步部分元數據。Apache Atlas對於實時消息的消費處理不滿足性能要求，內部使用Flink任務的處理方案在ToB場景中也存在諸多限制，所以團隊自研了輕量級非同步消息處理框架，很好的支持了位元組內部和火山引擎上同步元數據的訴求。本文定義了需求場景，並詳細介紹框架的設計與實現。

背景

動機

位元組數據中台DataLeap的Data Catalog系統基於Apache Atlas搭建，其中Atlas通過Kafka獲取外部系統的元數據變更消息。在開源版本中，每台伺服器支持的Kafka Consumer數量有限，在每日百萬級消息體量下，經常有長延時等問題，影響用戶體驗。

在2020年底，我們針對Atlas的消息消費部分做了重構，將消息的消費和處理從後端服務中剝離出來，並編寫了Flink任務承擔這部分工作，比較好的解決了擴展性和性能問題。然而，到2021年年中，團隊開始重點投入私有化部署和火山公有雲支持，對於Flink集群的依賴引入了可維護性的痛點。

在仔細的分析了使用場景和需求，並調研了現成的解決方案後，我們決定投入人力自研一個消息處理框架。當前這個框架很好的支持了位元組內部以及ToB場景中Data Catalog對於消息消費和處理的場景。

本文會詳細介紹框架解決的問題，整體的設計，以及實現中的關鍵決定。

需求定義

使用下麵的表格將具體場景定義清楚。

設計

概念說明

MQ Type：Message Queue的類型，比如Kafka與RocketMQ。後續內容以Kafka為主，設計一定程度相容其他MQ。
Topic：一批消息的集合，包含多個Partition，可以被多個Consumer Group消費。
Consumer Group：一組Consumer，同一Group內的Consumer數據不會重覆消費。
Consumer：消費消息的最小單位，屬於某個Consumer Group。
Partition：Topic中的一部分數據，同一Partition內消息有序。同一Consumer Group內，一個Partition只會被其中一個Consumer消費。
Event：由Topic中的消息轉換而來，部分屬性如下。

Event Type：消息的類型定義，會與Processor有對應關係；

Event Key：包含消息Topic、Partition、Offset等元數據，用來對消息進行Hash操作；
Processor：消息處理的單元，針對某個Event Type定製的業務邏輯。
Task：消費消息並處理的一條Pipeline，Task之間資源是相互獨立的。

框架架構

整個框架主要由MQ Consumer, Message Processor和State Manager組成。

MQ Consumer：負責從Kafka Topic拉取消息，並根據Event Key將消息投放到內部隊列，如果消息需要延時消費，會被投放到對應的延時隊列；該模塊還負責定時查詢State Manager中記錄的消息狀態，並根據返回提交消息Offset；上報與消息消費相關的Metric。
Message Processor：負責從隊列中拉取消息並非同步進行處理，它會將消息的處理結果更新給State Manager，同時上報與消息處理相關的Metric。
State Manager：負責維護每個Kafka Partition的消息狀態，並暴露當前應提交的Offset信息給MQ Consumer。

實現

線程模型

每個Task可以運行在一臺或多台實例，建議部署到多台機器，以獲得更好的性能和容錯能力。

每台實例中，存在兩組線程池：

Consumer Pool：負責管理MQ Consumer Thread的生命周期，當服務啟動時，根據配置拉起一定規模的線程，併在服務關閉時確保每個Thread安全退出或者超時停止。整體有效Thread的上限與Topic的Partition的總數有關。
Processor Pool：負責管理Message Processor Thread的生命周期，當服務啟動時，根據配置拉起一定規模的線程，併在服務關閉時確保每個Thread安全退出或者超時停止。可以根據Event Type所需要處理的並行度來靈活配置。

兩類Thread的性質分別如下：

Consumer Thread：每個MQ Consumer會封裝一個Kafka Consumer，可以消費0個或者多個Partition。根據Kafka的機制，當MQ Consumer Thread的個數超過Partition的個數時，當前Thread不會有實際流量。
Processor Thread：唯一對應一個內部的隊列，並以FIFO的方式消費和處理其中的消息。

StateManager

在State Manager中，會為每個Partition維護一個優先隊列（最小堆），隊列中的信息是Offset，兩個優先隊列的職責如下：

處理中的隊列：一條消息轉化為Event後，MQ Consumer會調用StateManager介面，將消息Offset 插入該隊列。
處理完的隊列：一條消息處理結束或最終失敗，Message Processor會調用StateManager介面，將消息Offset插入該隊列。

MQ Consumer會周期性的檢查當前可以Commit的Offset，情況枚舉如下：

處理中的隊列堆頂 < 處理完的隊列堆頂或者處理完的隊列為空：代表當前消費回來的消息還在處理過程中，本輪不做Offset提交。
處理中的隊列堆頂 = 處理完的隊列堆頂：表示當前消息已經處理完，兩邊同時出隊，並記錄當前堆頂為可提交的Offset，重覆檢查過程。
處理中的隊列堆頂 > 處理完的隊列堆頂：異常情況，通常是數據回放到某些中間狀態，將處理完的隊列堆頂出堆。

註意：當發生Consumer的Rebalance時，需要將對應Partition的隊列清空

KeyBy與Delay Processing的支持

因源頭的Topic和消息格式有可能不可控制，所以MQ Consumer的職責之一是將消息統一封裝為Event。

根據需求，會從原始消息中拼裝出Event Key，對Key取Hash後，相同結果的Event會進入同一個隊列，可以保證分區內的此類事件處理順序的穩定，同時將消息的消費與處理解耦，支持增大內部隊列數量來增加吞吐。

Event中也支持設置是否延遲處理屬性，可以根據Event Time延遲固定時間後處理，需要被延遲處理的事件會被髮送到有界延遲隊列中，有界延遲隊列的實現繼承了DelayQueue，限制DelayQueue長度, 達到限定值入隊會被阻塞。

異常處理

Processor在消息處理過程中，可能遇到各種異常情況，設計框架的動機之一就是為業務邏輯的編寫者屏蔽掉這種複雜度。Processor相關框架的邏輯會與State Manager協作，處理異常並充分暴露狀態。比較典型的異常情況以及處理策略如下：

處理消息失敗：自動觸發重試，重試到用戶設置的最大次數或預設值後會將消息失敗狀態通知State Manager。
處理消息超時：超時對於吞吐影響較大，且通常重試的效果不明顯，因此當前策略是不會對消息重試，直接通知State Manager 消息處理失敗。
處理消息較慢：上游Topic存在Lag，Message Consumer消費速率大於Message Processor處理速率時，消息會堆積在隊列中，達到隊列最大長度，Message Consumer 會被阻塞在入隊操作，停止拉取消息，類似Flink框架中的背壓。

監控

為了方便運維，在框架層面暴露了一組監控指標，並支持用戶自定義Metrics。其中預設支持的Metrics如下表所示：

線上運維Case舉例

實際生產環境運行時，偶爾需要做些運維操作，其中最常見的是消息堆積和消息重放。

對於Conusmer Lag這類問題的處理步驟大致如下：

查看Enqueue Time，Queue Length的監控確定服務內隊列是否有堆積。
如果隊列有堆積，查看Process Time指標，確定是否是某個Processor處理慢，如果是，根據指標中的Tag 確定事件類型等屬性特征，判斷業務邏輯或者Key設置是否合理；全部Processor 處理慢，可以通過增加Processor並行度來解決。
如果隊列無堆積，排除網路問題後，可以考慮增加Consumer並行度至Topic Partition 上限。

消息重放被觸發的原因通常有兩種，要麼是業務上需要重放部分數據做補全，要麼是遇到了事故需要修複數據。為了應對這種需求，我們在框架層面支持了根據時間戳重置Offset的能力。具體操作時的步驟如下：

使用服務測暴露的API，啟動一臺實例使用新的Consumer GroupId: {newConsumerGroup} 從某個startupTimestamp開始消費
更改全部配置中的 Consumer GroupId 為 {newConsumerGroup}
分批重啟所有實例

總結

為瞭解決位元組數據中台DataLeap中Data Catalog系統消費近實時元數據變更的業務場景，我們自研了輕量級消息處理框架。當前該框架已在位元組內部生產環境穩定運行超過1年，並支持了火山引擎上的數據地圖服務的元數據同步場景，滿足了我們團隊的需求。

下一步會根據優先順序排期支持RocketMQ等其他消息隊列，並持續優化配置動態更新，監控報警，運維自動化等方面。

立即跳轉火山引擎大數據研發治理套件DataLeap官網瞭解詳情

DataLeap的Catalog系統近實時消息同步能力優化

摘要

背景

動機