監聽MySQL的binlog日誌工具分析：Canal
- 2020 年 10 月 20 日
- 筆記
- canal, MySQL, mysql binlog, 大數據
Canal是阿里巴巴旗下的一款開源項目，利用Java開發。主要用途是基於MySQL資料庫增量日誌解析，提供增量數據訂閱和消費，目前主要支援MySQL。
GitHub地址：//github.com/alibaba/canal
在介紹Canal內部原理之前，首先來了解一下MySQL Master/Slave同步原理：
- MySQL master啟動binlog機制，將數據變更寫入二進位日誌（binary log, 其中記錄叫做二進位日誌事件binary log events，可以通過show binlog events進行查看）
- MySQL slave（I/O thread）將master的binary log events拷貝到它的中繼日誌（relay log）
- MySQL slave（SQL thread）重放relay log中事件，將數據變更反映它自己的數據中
Canal工作原理：
- Canal模擬MySQL slave的交互協議，偽裝自己為MySQL slave，向MySQL master發送dump協議
- MySQL master收到dump請求，開始推送binary log給slave（也就是canal）
- Canal解析binary log對象（原始為byte流）
簡而言之，Canal是通過模擬成為MySQL的slave，監聽MySQL的binlog日誌來獲取數據。當把MySQL的binlog設置為row模式以後，可以獲取到執行的每一個Insert/Update/Delete的腳本，以及修改前和修改後的數據，基於這個特性，Canal就能高效的獲取到MySQL數據的變更。
Canal架構：
說明：
server代表一個Canal運行實例，對應於一個jvm
instance對應於一個數據隊列（1個server對應1..n個instance)
EventParser：數據源接入，模擬slave協議和master進行交互，協議解析
EventSink：Parser和Store連接器，主要進行數據過濾，加工，分發的工作
EventStore：負責存儲
MemoryMetaManager：增量訂閱和消費資訊管理器
Event Parser設計：
整個parser過程大致可分為以下幾步：
- Connection獲取上一次解析成功的log position（如果是第一次啟動，則獲取初始指定的位置或者是當前資料庫的binlog log position）
- Connection建立連接，向MySQL master發送BINLOG_DUMP請求
- MySQL開始推送binary Log接收到的binary Log
- 通過BinlogParser進行協議解析，補充一些特定資訊。如補充欄位名字、欄位類型、主鍵資訊、unsigned類型處理等
- 將解析後的數據傳入到EventSink組件進行數據存儲（這是一個阻塞操作，直到存儲成功）
- 定時記錄binary Log位置，以便重啟後繼續進行增量訂閱
如果需要同步的master宕機，可以從它的其他slave節點繼續同步binlog日誌，避免單點故障。
Event Sink設計：
EventSink主要作用如下：
數據過濾：支援通配符的過濾模式，表名，欄位內容等
數據路由/分發：解決1:n（1個parser對應多個store的模式）
數據歸併：解決n:1（多個parser對應1個store）
數據加工：在進入store之前進行額外的處理，比如join
數據1:n業務
為了合理的利用資料庫資源， 一般常見的業務都是按照schema進行隔離，然後在MySQL上層或者dao這一層面上，進行一個數據源路由，屏蔽資料庫物理位置對開發的影響，阿里系主要是通過cobar/tddl來解決數據源路由問題。所以，一般一個資料庫實例上，會部署多個schema，每個schema會有由1個或者多個業務方關注。
數據n:1業務
同樣，當一個業務的數據規模達到一定的量級後，必然會涉及到水平拆分和垂直拆分的問題，針對這些拆分的數據需要處理時，就需要鏈接多個store進行處理，消費的位點就會變成多份，而且數據消費的進度無法得到儘可能有序的保證。所以，在一定業務場景下，需要將拆分後的增量數據進行歸併處理，比如按照時間戳/全局id進行排序歸併。
Event Store設計：
支援多種存儲模式，比如Memory記憶體模式。採用記憶體環裝的設計來保存消息，借鑒了Disruptor的RingBuffer的實現思路。
RingBuffer設計：
定義了3個cursor：
put：Sink模組進行數據存儲的最後一次寫入位置（同步寫入數據的cursor）
get：數據訂閱獲取的最後一次提取位置（同步獲取的數據的cursor）
ack：數據消費成功的最後一次消費位置
借鑒Disruptor的RingBuffer的實現，將RingBuffer拉直來看：
實現說明：
- put/get/ack cursor用於遞增，採用long型存儲。三者之間的關係為put>=get>=ack
- buffer的get操作，通過取余或者&操作。(&操作：cusor & (size – 1) , size需要為2的指數，效率比較高)
Instance設計：
instance代表了一個實際運行的數據隊列，包括了EventPaser、EventSink、EventStore等組件。抽象了CanalInstanceGenerator，主要是考慮配置的管理方式：
manager方式：和你自己的內部web console/manager系統進行對接。(目前主要是公司內部使用)
spring方式：基於spring xml + properties進行定義，構建spring配置。
Server設計：
server代表了一個Canal運行實例，為了方便組件化使用，特意抽象了Embeded(嵌入式)/Netty(網路訪問)的兩種實現。
增量訂閱/消費設計：
具體的協議格式，可參見：CanalProtocol.proto。數據對象格式：EntryProtocol.proto
Entry Header logfileName [binlog文件名] logfileOffset [binlog position] executeTime [binlog里記錄變更發生的時間戳] schemaName [資料庫實例] tableName [表名] eventType [insert/update/delete類型] entryType [事務頭BEGIN/事務尾END/數據ROWDATA] storeValue [byte數據,可展開，對應的類型為RowChange] RowChange isDdl [是否是ddl變更操作，比如create table/drop table] sql [具體的ddl sql] rowDatas [具體insert/update/delete的變更數據，可為多條，1個binlog event事件可對應多條變更，比如批處理] beforeColumns [Column類型的數組] afterColumns [Column類型的數組] Column index [column序號] sqlType [jdbc type] name [column name] isKey [是否為主鍵] updated [是否發生過變更] isNull [值是否為null] value [具體的內容，注意為文本]
針對上述的補充說明：
1.可以提供資料庫變更前和變更後的欄位內容，針對binlog中沒有的name、isKey等資訊進行補全
2.可以提供ddl的變更語句
Canal HA機制：
Canal的HA實現機制是依賴zookeeper實現的，主要分為Canal server和Canal client的HA。
Canal server:為了減少對MySQL dump的請求，不同server上的instance要求同一時間只能有一個處於running狀態，其他的處於standby狀態。
Canal client:為了保證有序性，一份instance同一時間只能由一個Canal client進行get/ack/rollback操作，否則客戶端接收無法保證有序。
Canal Server HA架構圖：
大致步驟：
- Canal server要啟動某個Canal instance時都先向Zookeeper進行一次嘗試啟動判斷 (實現：創建EPHEMERAL節點，誰創建成功就允許誰啟動)
- 創建Zookeeper節點成功後，對應的Canal server就啟動對應的Canal instance，沒有創建成功的Canal instance就會處於standby狀態
- 一旦Zookeeper發現Canal server A創建的節點消失後，立即通知其他的Canal server再次進行步驟1的操作，重新選出一個Canal server啟動instance
- Canal client每次進行connect時，會首先向Zookeeper詢問當前是誰啟動了Canal instance，然後和其建立鏈接，一旦鏈接不可用，會重新嘗試connect
Canal Client的方式和Canal server方式類似，也是利用Zookeeper的搶佔EPHEMERAL節點的方式進行控制。
關注微信公眾號：大數據學習與分享，獲取更對技術乾貨
