作者：京東科技 劉紅申

一、事件總線介紹

事件總線，或稱其為資料管道，作為整個風險洞察資料流轉的重要一環，它承擔著風險實時資料統一標準化的重要職責，

在面對復雜多樣的上游資料，事件總線可以將復雜資料進行決議、轉換, 富化、分發等操作，底層核心算子抽象為source、transform、sink三層架構, 支持各層算子插件式擴展, 并支持groovy、python等腳本語言自定義配置，以及自定義jar包的上傳，擁有將上游資料單向接入多向輸出的能力，在數倉與上層應用的開展中，起著承上啟下的作用，

二、事件總線-遇到的技術挑戰與解決方案

技術難點與挑戰

風險洞察平臺運行初期，業務資料接入完全采用定制化代碼處理，通過代碼配置消費MQ訊息，然后根據業務需求，完成其所需欄位的決議，最終資料落入Clickhouse，這種業務接入方式在早期是可以滿足業務所需，但是隨著風險洞察平臺在風控領域的不斷推進，業務的發展與資料不斷膨脹，面對風控資料的復雜多樣性、訊息平臺的差異性，資料接入定制化成本也越來越高，同時資料轉化與計算邏輯的強耦合，大促時期吞吐量已然達到瓶頸，呈現出越來越多的痛點：

1. 資料結構差異性: 隨著風險洞察平臺使用業務方的的不斷增加，業務資料訊息體的復雜性也不盡相同，如復雜場景以天盾反欺詐場景為例，訊息體結構包含物件、物件字串而且還有陣列；簡單場景以內容安全為例，訊息體結構就是簡單平鋪的一層；面對風控資料的復雜多樣性，定制資料的統一標準化已然迫在眉睫；

2. 代碼邏輯重復性: 對訊息體的處理絕大多數逃離不了序列化與反序列化操作，然而隨著業務量的增多以及開發人員的不盡相同，業務代碼是每日劇增且帶有參差性的，邏輯重復，維護成本高；

3. 決議寫入低效性: 同一個MQ訊息可能會對應很多的業務方，不同的業務方所需業務資料又千差萬別，如以天策MQ為例，實時資料中包含著金白條資料，金條與白條資料又區分著各自的業務線，如果單次訂閱MQ訊息，會導致邏輯處理極其復雜，不可維護；然而采用多次訂閱，又無法復用已有邏輯，且導致資料成倍增長，造成資源浪費，同時吞吐能力成為瓶頸；

4. 輸入輸出多樣性: 隨著風險洞察平臺被使用的越來越廣，來自于上游資料的生產方式也出現了多樣性，如JMQ2、FMQ、Kafka以及JMQ4等等，同時又為了給用戶更好的平臺使用體驗，不同業務資料又會被落入不同存盤中，如Clickhouse、R2m、Jes以及訊息佇列，如何快速支持這些組件成為了挑戰；

5. 業務需求易變性: 上游業務頻繁的策略調整與變更，對應到事件總線就意味著決議欄位以及底層表欄位頻繁的增刪改，正如欄位決議完全依賴于硬編碼且不同業務資料耦合著各自的業務邏輯，導致開發人員維護成本極高，開發周期長、上線影響廣；

技術解決方案

研發一套資料流轉服務，用其貫穿資料接入到數倉存盤的整個流程，再結合風險洞察平臺特性，以資料源組件為基礎，作為資料流轉的入口與出口，具體方案如下：

? 資料統一標準化能力：統一標準化入口與出口，上游資料接入時，無論訊息體結構如何，經過事件總線處理后，都輸出為平鋪單層key-value結構；

? 代碼邏輯規范化能力：針對風控策略本身易變的特性，采用靈活度更高的訊息體決議組件Jsonpath，任何訊息體處理第一步就是生成訊息體背景關系物件，后續欄位的提取，都從這個背景關系中獲取；

? 高吞吐決議寫入能力：一次決議，多路復用，MQ主題實作單次接入，根據不同的業務需求通過過濾下沉不同的業務表，如以天策金白條為例，提取金白條各自的INTERFACE_NAME作為條件，下沉到不同的業務表中；又如以高TPS營銷反欺詐場景為例，在下沉表的同時，下沉訊息佇列給Flink計算使用；減少重復決議，同時抽象各種算子，針對不同的數倉寫入可做對應的頻次、批次、大小設定，提升吞吐量；

? 輸入輸出插件化能力：輸入輸出插件化，新的業務需求來時，可以快速擴展相應組件，以應對新需求；

? 低代碼化熱加載能力：針對業務需求的頻繁變更，解決硬編碼問題，減少上線頻次，那就需要開發一套可配置化系統，支持腳本開發與熱加載，同時內置函式插件化，快速擴展共性函式；

三、事件總線-整體架構圖

事件總線-架構介紹

事件總線整體架構抽象為三層，source、transform 和sink， 通過連接器擴展機制實作資料引擎擴展, 并采用責任鏈模式處理資料鏈路, 插件化管理函式、腳本，實作實時訊息接入、過濾、富化、轉換、分發標準化處理， 并通過分組消費、降級機制保證架構高可用，

? 實時資料: 風險核心場景，目前事件總線業務資料的主要來源；

? 事件總線:

? Source：資料輸入層，風險業務資料的主要來源方式，目前大多數來源于JMQ2、JMQ4、FMQ等；

? Transform: 事件總線的核心處理層，同時也是自定義函式與自定義腳本的決議層，該層抽象了大量的算子，如，資料決議算子、過濾算子、富化算子、轉換算子等等當復雜訊息體資料經過一系列算子之后，最侄訓轉化為單層key-value標準結構；

? Sink: 資料輸出層，經Transform組件轉換后，此時的資料可以發實時訊息給各個訊息佇列，也可以存盤到Clickhouse、Es、R2m等資料庫；

? 資料服務: 基于事件總線標準化后沉淀的資料所支撐的平臺應用；

事件總線-核心類圖介紹

事件總線定義了一個頂層父介面IEventHubExecutor，并定義了一個execute方法，其三個主要子介面，IEventHubParse、IEventHubTransform與IEventHubSink分別對應于事件總線的三個組成部分，source、transform和sink，通過實作這三個子介面，便可以完成對不同中間件的適配問題，比如，目前事件總線僅支持決議的資料寫入到Clickhouse，但業務需求需要做檢索，那么很顯然資料存盤在Es要優于存盤在Clickhouse，所以此時需要擴展一個JesEventHubSink來實作IEventHubSink即可，

其中Context作為背景關系，貫穿了整個事件總線的執行程序，背景關系中包含了決議程序中所需要的一起資訊，比如，從哪里來的資料、要決議哪些欄位、決議好的資料送到那里去等等，

事件總線-自定義函式介紹

自定義函式的實作，其實借助了開源框架Avaitor運算式，Aviator是一個輕量級、高性能的Java運算式執行引擎, 它動態地將運算式編譯成位元組碼并運行，主要用于各種運算式的動態求值，相比Groovy這樣的重量級腳本語言，Aviator是非常輕量級的運算式執行引擎，

? 函式決議器：自定義函式支持腳本撰寫（腳本采用groovy，同時為了更加“親民”，采用Java語法）與Jar包上傳兩種方式；

? 函式編譯器：編譯腳本與決議jar包，生成對應的AvaitorFunction實體；

? 函式注冊器：將生成的AvaitorFunction實體注冊到Avaitor的背景關系中；

? 函式執行器：通過實作FunctionExecutor，便可以對函式方便的呼叫；

事件總線-動態分組、一鍵降級與流量監控介紹

分組消費

事件總線決議能力的提升，也很大一部分歸結于分組消費的設計，對流量做到靈活分流，對機器做到物盡其用，動態分組，又分為物理分組與邏輯分組，如下圖：

? 物理分組：單純依靠機器劃分，規定好哪些機器消費哪些主題，如，天盾分組就消費天盾主題，營銷分組就消費營銷主題，

? 邏輯分組：邏輯分組與物理分組的區別在于，邏輯分組在物理分組之上，又抽象出一個消費組的概念，用機器與消費組系結，而非直接與主題系結，這樣帶來的好處就是，可以更加方便的調配流量，如，營銷流量非常大，那么可以直接動態調配，使天盾分組也去消費營銷主題，既能充分利用天盾分組機器，又能提高營銷主題消費能力，

一鍵降級

一鍵降級更多的用于大促期間，但是為了降的更加“人性化”，一鍵降級我們也做了分類：丟棄降級與積壓降級，如下圖：

? 丟棄降級：所降級主題處于消費狀態，顧名思義，事件總線拿到了資料，就直接將資料丟棄，降級期間資料是不可找回的；丟棄降級可用于業務方并不在意一時資料的丟失或者壓測場景，

? 積壓降級：所降級主題處于非消費狀態，降級期間資料積壓在訊息平臺，降級過后，再開啟消費；積壓降級可用于業務方允許降級期間內沒有新資料，但是降級過后資料又可查場景，

流量監控

事件總線的流量監控現依賴于ump，對單個主題以及所有主題的入口都設有埋點，資料在每個關鍵流轉位置決議性能以及流量都能被監控，代碼片段如下：

Profiler.registerInfo(this.getClass().getSimpleName(), UmpUtil.UMP_APP_NAME, false, true);

四、未來展望

自事件總線上線以來，已經經歷了多次大促考驗，大促決議量已達5000w/min,日常決議量也已2000w/min，伴隨著風險洞察平臺被越來越多的部門所使用，事件總線已然成為其重要組成部分，為了更好的提高決議性能，就需要去做更多的探索，同時，目前事件總線做的更多的是對實時資料的處理，未來我們也將推進flink-cdc等技術在事件總線中的應用，

標籤：其他

上一篇：vivo版本發布平臺：帶寬智能調控優化實踐-平臺產品系列03

下一篇：風險洞察之事件總線的探索與演進