監聽MySQL的binlog日誌工具分析：Canal

Canal是阿里巴巴旗下的一款開源專案，利用Java開發。主要用途是基於MySQL資料庫增量日誌解析，提供增量資料訂閱和消費，目前主要支援MySQL。

GitHub地址：https：//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］RowChangeisDdl ［是否是ddl變更操作，比如create table/drop table］sql ［具體的ddl sql］rowDatas ［具體insert/update/delete的變更資料，可為多條，1個binlog event事件可對應多條變更，比如批處理］beforeColumns ［Column型別的陣列］afterColumns ［Column型別的陣列］Columnindex ［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節點的方式進行控制。

關注

微信公眾號：大資料學習與分享

，獲取更對技術乾貨