1. 程式人生 > >「從零單排canal 05」 server模組原始碼解析

「從零單排canal 05」 server模組原始碼解析

基於1.1.5-alpha版本,具體原始碼筆記可以參考我的github:https://github.com/saigu/JavaKnowledgeGraph/tree/master/code_reading/canal

本文將對canal的server模組進行分析,跟之前一樣,我們帶著幾個問題來看原始碼:

  • CanalServer有幾種使用方式?
  • 控制檯Admin、客戶端client是如何與CanalServer互動的?
  • CanalServerWithNetty和CanalServerWithEmbedded究竟有什麼關係?
  • Canal事件消費的特色協議,非同步流式api(get/ack/rollback協議)的設計是如何實現的?

server模組內的結構如下:

 

主要分為了三個包:

  • admin包:

這個包的CanalAdmin介面定義了canalServer上暴露給canal-admin控制檯使用的一些服務介面。

上一篇deployer模組解析中提到的CanalAdminController就是實現了CanalAdmin介面(把這個介面的實現放在deployer模組是挺奇怪的)。 Admin包中使用了netty作為服務端(CanalAdminWithNetty類中實現),接受控制檯Admin的請求,返回當前canalServer的一些執行狀態。

  • server包:

server模組的核心包,本文重點解析的部分,需要了解CanalServerWithEmbedded 和CanalServerWithNetty。

  • spi包:

定義了canalServer的監控內容 通過spi實現,比如專案中的Prometheus子模組實現了監控能力,我們不展開分析。

1.從CanalServer的架構說起

CanalServer目前支援兩種模式:

  • serverMode = tcp的Server-Client模式
  • serverMode = kafak 或 rocketMQ 的 Server-MQ-Client模式

為了大家能充分理解canalServer的結構,這裡精心製作了一個canalServer的架構圖(如果覺得這圖不錯,給本文點個贊吧)。

1.1 Server-Client模式

架構如圖所示:

 

我們可以清楚的看到Server模組中各個模組的關係與能力:

  • CanalServerWithEmbedde維護了具體的instance任務,負責對binlog進行訂閱、過濾、快取,就是之前的文章介紹過的parser-sink-store的方式。
  • CanalServerWithNetty作為服務端,接收CanalClient的請求,將binlog的訊息傳送給client。
  • CanalAdminWithNetty作為admin的伺服器,接收控制檯Admin的控制操作、查詢狀態操作等,啟停或顯示當前CanalServer以及instance的狀態。

1.2 Server-MQ-Client模式

架構如圖所示:

 

主體部分與Server-client模式一致,主要區別如下:

  • 不需要CanalServerWithNetty,改為CanalMQProducer投遞訊息給訊息佇列
  • 不使用CanalClient,改為MqClient獲取訊息佇列的訊息進行消費

這種模式相比於Server-client模式

  • 下游解耦,利用訊息佇列的特性,可以支援多個客戶端廣播消費、叢集消費、重複消費等
  • 會增加系統的複雜度,增加一些延遲

具體模式的選擇,需要根據具體的使用場景來決定。

2.server包

admin包和spi包都不屬於核心邏輯,因此我們重點關注server包的程式碼。

 

我們看到,server包下面分為了embedded包、exception包、netty包和幾個介面類。

其中,最頂層的設計就要從CanalServer介面入手。

 

它的實現類有兩個,CanalServerWithEmbedded 和 CanalServerWithNetty。

它們之間的區別官方文件給了一些說明。

 

那麼,對於官方文件中提到的Embedded(嵌入式)的自主開發是怎麼使用呢?

跟我們上面提到的Server-Client模式和Server-MQ-Client模式完全不同,採用了一種無server的架構,如下圖所示。

 

我們可以看到,這種模式沒有了Canal-Server,直接在自己的應用中引入canal,然後使用CanalServerWithEmbedded進行資料抓取和訂閱。

當然,這種方式開發成本有點高,一般也不會去這樣使用。

對於CanalServerWithEmbedded 和 CanalServerWithNetty,官方文件裡面實際上沒有解釋的特別到位,只講了區別,沒有講聯絡。

這兩個實現類除了官方文件中說明的區別之外,還有很大的聯絡。

可以看看我們上文介紹的架構圖,對於Server-Client模式下的模組聯絡

 

實際上,真正的執行邏輯是在CanalServerWithEmbedded中的,CanalServerWithNetty中持有了CanalServerWithEmbedded物件,委託embedded進行相關邏輯處理,CanalServerWithNetty更多的作用是充當服務端與CanalClient進行互動。

3. CanalServerWithNetty類

下面,我們先看看CanalServerWithNetty類。

3.1 單例構建

使用 private構造器 + 靜態內部類 來實現一個單例模式,保證了一個CanalServer內部只有一個CanalServerWithNetty。

同時,我們能看到內部持有一個CanalServerWithEmbedded物件,用來處理相關請求,驗證了我們上面的說明。

 

3.2 啟動邏輯 start()

原始碼如下:

 

主要流程如下:

  • 啟動embeddedServer
  • 建立bootstrap例項,設定netty相關配置

引數NioServerSocketChannelFactory也是Netty的API,接受2個執行緒池引數,第一個執行緒池是Accept執行緒池,第二個執行緒池是woker執行緒池,Accept執行緒池接收到client連線請求後,會將代表client的物件轉發給worker執行緒池處理。這裡屬於netty的知識,不熟悉的可以暫時不必深究,簡單認為netty使用執行緒來處理客戶端的高併發請求即可。

  • 構造對應的pipeline,包括解碼處理、身份驗證、建立netty的 seesionHandler(真正處理客戶端請求,seesionHandler的實現是核心邏輯)

pipeline實際上就是netty對客戶端請求的處理器鏈,可以類比JAVA EE程式設計中Filter的責任鏈模式,上一個filter處理完成之後交給下一個filter處理,只不過在netty中,不再是filter,而是ChannelHandler。

  • 啟動netty,監聽port埠,然後客戶端對 這個埠的請求可以被接收到

對於 netty的相關知識 ,本文 不深入展開,簡單理解 為一個高效能伺服器即可,可以監聽 埠請求,並 進行相應的處理。

重點在於sessionHandler的處理。

3.3 邏輯分發SessionHandler類

canalServer的處理邏輯顯然都在sessionHandler裡面,而這個handler在構建時,傳入了embeddedServer。

前面我們提過,serverWithNetty的處理邏輯是委派給embeddedServer的,所以這裡就非常順理成章了,讓handler維護embeddedServer例項,進行邏輯處理。

sessionHandler繼承了netty的SimpleChannelHandler類,重寫了messageReceived方法,接收到不同請求後,委託embeddedServer用不同方法進行處理 。

這個方法裡面的程式碼非常冗長,而本質都是委託給embeddedServer去處理,因此,我們看下主幹邏輯即可。

 

可以看到,根據不同的packet型別,最終都是委託給embeddedServer進行處理,這裡只是做一個邏輯的判斷和分發。

3.4 CanalServerWithNetty小結

到此,我們已經瞭解了CanalServerWithNetty是如何啟動的。

並且,它的主要定位就是充當伺服器,接收客戶端的請求,然後做訊息分發,委託給CanalServerEmbedded進行處理。

下面,我們來看下CanalServerEmbedded的相關實現。

4. CanalServerEmbedded類

4.1 基本認識

  • 非完全單例模式,這裡使用public的構造器,使用者還是有機會自己new物件出來的,應用是用來獨立引入進行開發的時候使用。
  • 維護了instance的物件容器
  • 繼承了CanalServer和CanalService介面

 

CannalServer介面其實就是就是start()和stop()方法,沒有特別的地方,主要是start()配置了一個MigrateMap.makeComputingMap,

當需要某個instance的時候,就會呼叫apply方法用instanceGenerator建立對應的instance。

 

我們重點看下CanalService介面定義的方法。

 

每個方法的入參都帶來clientIdentity,這個是客戶端的身份標示

 

目前canal只支援一個客戶端對一個instance進行訂閱,clientId全部寫死為1001,據說以後可能會支援多使用者訂閱。

瞭解CanalService定義的方法在CanalServerEmbedded中如何實現,基本也就能看清CanalServerEmbedded的全貌了。

尤其是,你能理解官網wiki中介紹的canal核心功能——非同步消費流式api(get/ack/rollback協議) 設計。

 

4.2 subscribe方法

主要步驟:

  • 根據客戶端標識clientIdentity中的destination,找到對應的instance
  • 通過instance的metaManager記錄下當前這個客戶端在訂閱
  • 通過instace的metaManage獲取當前訂閱binlog的position位置。如果是第一次訂閱,那麼metaManage沒有position資訊,就從eventStore獲取第一個binlog的position,然後更新到metaManager
  • 通知下訂閱關係變化

 

這裡需要注意一下metaManager,這是一個介面,有多種實現方式,包括基於記憶體、基於檔案、基於記憶體+zookeeper混合、基於zookeeper等,都在meta模組中,這裡就簡單瞭解下概念即可。

  • MemoryMetaManager:位點資訊儲存在記憶體中
  • ZookeeperMetaManage:位點資訊儲存在zk上
  • PeriodMixedMetaManager:前面兩種的混合,儲存在記憶體中,然後位點資訊定期重新整理到zk上

 

我們在叢集模式下,default-instance.xml使用的是基於PeriodMixedMetaManager的實現。

4.3 unsubscribe方法

這個方法比較簡單,就不放原始碼了。

就是找到instance對應的metaManager,然後呼叫unsubscribe方法取消這個客戶端的訂閱。

需要注意的是,取消訂閱,instance本身仍然是在執行的,可以有新的client來訂閱這個instance。

4.4 getWithoutAck方法

先解釋幾個概念。

我們用的叢集版canalServer,預設是使用PeriodMixedMetaManager來管理位點資訊,也就是MemoryMetaManager + zookeeperMetaManager。

其中,對於客戶端消費instance訊息的情況,內部維護了一個物件MemoryClientIdentityBatch進行記錄

 

回到這個方法來說,這個方法用於客戶端獲取binlog訊息,大致流程如下:

  • 根據clientIdentity的destination獲取對應的instance
  • 獲取到流式資料中的最後一批獲取的位置positionRanges(跟batchId有關聯,就是上面那個map裡面的)
  • 從cananlEventStore裡面獲取binlog,轉化為event。一般是從最後的一個batchId位置開始,如果之前沒有batchId,那麼就從cursor記錄的消費位點開始;如果cursor為空,那隻能從eventStore的第一條訊息開始。
  • event轉化為entry,並生成新的batchId,組合成message返回給客戶端

注意在eventStore獲取event的時候,使用者可以自己設定batchSize和超時時間timeout。為了儘量提高效率,一般一次獲取一批binlog,而不是獲取一條。這個批次的大小(batchSize)由客戶端指定。同時客戶端可以指定超時時間,在超時時間內,如果獲取到了batchSize的binlog,會立即返回。 如果超時了還沒有獲取到batchSize指定的binlog個數,也會立即返回。特別的,如果沒有設定超時時間,如果沒有獲取到binlog也立即返回。具體eventStore的獲取邏輯,我們下次講到這個模組再展開。

 

4.5 get方法

這個方法主要是用於客戶端獲取binlog訊息,與getWithoutAck基本一致。

主要區別在於,客戶端獲取batch後,自動ack,這樣相對來說肯定更快,但是無法保證可靠性。

在專案中看起來暫時沒有使用,我們就不展開了。

4.6 ack方法

進行 batch id 的確認。確認之後,小於等於此 batchId 的 Message 都會被確認。

  • 從metaManager中移除batchId對應的記錄
  • 記錄已經成功消費到的binlog位置,以便下一次獲取的時候可以從這個位置開始
  • 已經ack的資料,在eventStore中清除

 

4.7 rollback

rollback有兩個方法,回滾所有和回滾指定batchId,不過從原始碼來看,目前回滾指定指定batchId也是回滾所有。

回滾的本質,就是把所有還沒ack的batchId都清空,流式api被get但是還沒ack的訊息會被重新get。

5.canalMQStarter

在第一節的架構模式中我們分析過了,在啟動過程中,如果serverMode選擇tcp,會啟動canalServerWithNetty,如果serverMode選擇了mq,就會啟動cannalMQStarter。

所以從模組組成來說,canalMQStarter跟canalServerWithNetty是比較相似的。

canalMQStarter也是委託embeddedCanal做處理,同時委託CanalMQProducer把訊息投遞到mq叢集。

canalServerWithNetty也是委託embeddedCanal做處理,然後通過netty來跟canal-client做互動。

如果我們以後應用中要內嵌embeddedCanal,完全可以參照canalMQStarter和canalServerWithNetty的模式來寫。

主要組成如下:

  • 工作執行緒池executorService,對每個instance起一個worker執行緒
  • canalMQWorks,記錄了destination(instance的標識)和worker執行緒的關係
  • CanalServerWithEmbedded
  • CanalMQProducer投遞mq訊息

 

5.1 start方法

這個方法就是前面canalStarter類裡面的start()方法中,對CanalMQStarter.start()的呼叫。

具體做了三件事情:

  • 獲取CanalServerWithEmbedded的單例物件
  • 對應每個instance啟動一個worker執行緒CanalMQRunnable
  • 註冊ShutdownHook,退出時關閉執行緒池和mqProducer

 

這裡主要看看CanalMQRunnable做了些什麼。

5.2 CanalMQRunnable

這是一個內部類,就是看看worker裡面做了什麼

 

只有一個worker方法,主要邏輯非常清晰:

  • 給自己建立一個身份標識,作為client
  • 根據destination獲取對應instance,如果沒有就sleep,等待產生(比如從別的server那邊HA過來一個instance)
  • 構建一個MQ的destination物件,載入相關mq的配置資訊,用作mqProducer的入參
  • 在embeddedCanal中註冊這個訂閱客戶端
  • 開始執行,並通過embededCanal進行流式get/ack/rollback協議,進行資料消費

 

 

6.總結

回到開頭的幾個問題,相信文中都已經做了解答。

  • CanalServer有幾種使用方式?

可以獨立部署(推薦),可以使用Server-Client模式 和 Server-MQ-Client模式兩種。

可以內嵌部署開發(embedded,難度較高)。

  • 控制檯Admin、客戶端client是如何與CanalServer互動的?

控制檯Admin通過CanalAdminWithNetty與服務端互動 客戶端client通過CanalServerWithNetty與服務端互動。

  • CanalServerWithNetty和CanalServerWithEmbedded究竟有什麼關係?

CanalServerWithEmbedded是真正核心邏輯(parser-sink-store)處理的地方 。CanalServerWithNetty持有CanalServerWithEmbedded物件,接收client的請求然後轉發給CanalServerWithEmbedded物件處理。

  • Canal事件消費的特色協議,非同步流式api(get/ack/rollback協議)的設計是如何實現的?

CanalServerWithEmbedded集成了CanalService介面,實現了具體的get/ack/rollback協議

 

都看到最後了,原創不易,點個關注,點個贊吧~
文章持續更新,可以微信搜尋「阿丸筆記 」第一時間閱讀,回覆關鍵字【學習】有我準備的一線大廠面試資料。
知識碎片重新梳理,構建Java知識圖譜:github.com/saigu/JavaK…(歷史文章查閱非常方便)