更多技術交流、求職機會，歡迎關注位元組跳動資料平臺微信公眾號，回復【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官網了解詳情

轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/509201.html

標籤：大數據

上一篇：Java學習-第一部分-第三階段-第三節：MySQL基礎

下一篇：開源直播課丨高效穩定易用的資料集成框架——ChunJun類加載原理與實作

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

摘要

背景

動機