「從零單排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…（歷史文章查閱非常方便）