Canal 源碼走讀

前言

canal 是什麼？ 引用一下官方回答：

阿里巴巴mysql數據庫binlog的增量訂閱&消費組件

canal 能作什麼？

基於日誌增量訂閱&消費支持的業務：

1. 數據庫鏡像
2. 數據庫實時備份
3. 多級索引 (賣家和買家各自分庫索引)
4. search build
5. 業務cache刷新
6. 價格變化等重要業務消息

好比 LZ 目前就使用 canal 實現數據實時複製，搜索引擎數據構建等功能。既然要使用，就好好的研究一下。

時間有限，一塊兒來簡單看看。

軟件架構

關於 canla 的工做原理，我就不展開了，有興趣的能夠看看官方文檔，或者這個 ppt ： docs.google.com/presentatio…

說白了， canal 就是假裝成 mysql 的 slave，dump binlog，解析 binlog，而後傳遞給應用程序，整體仍是蠻簡單的。

好，咱們來看看 canal 的代碼架構。

咱們看到，canal server 內部由幾個模塊組成， 最外部的是 Server，該 Server 接收 Canal Client 請求，並返回 Client 數據。一個 Server 就是一個 JVM。每一個 Server 內部由多個 CanalInstance，每一個 CanalInstance 其實就是咱們設置的 destination，一般是一個數據庫。

每一個 CanalInstance 內部由 5 個模塊，分別是 parser 解析，sink 過濾，store 存儲，metaManager 元數據管理，Alarm 報警。

這 5 個模塊是幹嗎的呢？

簡單說一下：

當 Canal Server 啓動後，會根據配置啓動 N 個 CanalInstance， 每一個 CanalInstance 都會使用 socket 鏈接 mysql，dump binlog，而後將數據交給 parser 解析，sink 過濾，store 存儲，當 client 鏈接時，會從 zk 上讀取該 client 的信息，而 metaManager 元數據管理就是管理 zk（固然有多種實現，好比存儲在文件中） 信息的，若是發生錯誤了，就調用 Alarm 發送報警信息（你能夠接入本身公司的監控系統），目前是打印日誌。

Canal 啓動流程

canal 代碼量目前有 6 萬多行，去除 2 個 ProtocolBuffer 生成類大概 1.7 萬行，也還有 4.3 萬行，代碼仍是很多的。

啓動過程也比較繞。這裏我簡單畫了一個流程圖：

解釋一下這個圖：

canal 腳本從 CanalLauncher main 方法啓動，而後調用 CanalController 的 start 方法，CanalController 調用 InstanceConfigMonitor 的 start 方法，最後調用 canal 關鍵組件 CanalServerWithEmbedded 的 start 方法。

在 Canal 內部， 有 CanalServerWithEmbedded 和 CanalServerWithNetty，前者是沒有 Server 端口的，是一個無故口的代理。後者是基於 Netty 實現的服務器，在 channelRead 方法中，會調用 CanalServerWithEmbedded 的相關方法。

CanalServerWithEmbedded 是單例的， 內部會有多個 CanalInstance， 他有多個實現，獨立版本中使用的是 CanalInstanceWithSpring 版本，基於 Spring 管理組件的生命週期。

每一個 CanalInstance 內部有 5 個組件，也就是上面說的幾個組件，他們會分別啓動。

其中，比較關鍵的是 parser，sink，store。

CanalEventParser 啓動後，會啓動一個叫作 parseThread 線程，不停的循環。主要是：構造與 mysql 的鏈接，而後啓動心跳線程，而後開始 dump binlog。

dump 出來的 binlog 經過 disruptor 無鎖隊列發佈，內部由 3 個消費者按照順序消費 binlog，處理完以後，交給了 sink 模塊。

而後是 sink，這個比較簡單，就不說了。sink 處理完以後，交給了 store 模塊。

store 模式是一個相似 RingBuffer 的循環數組，存儲着從 mysql dump 出來的數據，client 也是從這裏獲取數據的。該數組維護着 3 個指針，get，put， ack。

這裏比較奇怪的是，爲何不使用責任鏈模式夠組裝組件？

Canal 數據流向

看了啓動流程，再來看看 canal 內部運行的數據流向是什麼樣子的。我這裏簡單畫了一個圖。

獨立版本的 Canal 使用 Netty 暴露端口，使用本身構造的 SessionHandler 處理 TCP 請求，SessionHandler 將請求交給 CanalServerWithEmbedded 來處理。

咱們看 CanalServerWithEmbedded 的一些方法，例如 subscribe，get，ack 等，都是和 client 對應的方法，也就是說，CanalServerWithEmbedded 是和 client 打交道的一個類。

CanalServerWithEmbedded 內部管理全部的 CanalInstance，經過 Client 的信息，找到 Client 訂閱的 CanalInstance，而後調用 CanalInstance 內部的 Store 模塊，也就是那個 RingBuffer 的 get 方法，獲取 RingBuffer 的數據。

從 Myslq 的角度看，MysqlConnection 從 Myslq dump 數據，交給 parser 解析，parser 解析完，交給 sink，sink 處理完，交給 store 保存，等待 client 前來獲取。

看完了數據流向，若是對哪裏有什麼疑問，就能夠看看哪一個模塊對應的代碼是什麼，直接看是看就行了。

總結

花了點時間看了看 Canal 的代碼，整體上仍是很是好的，只是有些地方有點疑問，例如 parser，sink，store 爲何不使用過濾器模式。

Client 和 CanalServerWithEmbedded 爲何不使用 RPC 的方式交互，這樣更簡單明瞭。

代碼裏回調方法太多太長，影響閱讀。

但整體瑕不掩瑜，值得一讀。