「從零單排canal 01」 canal 10分鐘入門（基於1.1.4版本）

1.簡介

canal [kə'næl]，譯意爲水道/管道/溝渠，主要用途是基於 MySQL 數據庫增量日誌解析，提供增量數據 訂閱 和 消費。應該是阿里雲DTS（Data Transfer Service）的開源版本。

2.提供的能力

Canal與DTS提供的功能基本類似：

1）基於Mysql的Slave協議實時dump binlog流，解析爲事件發送給訂閱方。

2）單Canal instance，單DTS數據訂閱通道均只支持訂閱一個RDS，提供給一個消費者。

3）可使用canal-client客戶端進行消息消費。

4）也能夠經過簡單配置，也能夠不須要自行使用canal-client消費，能夠選擇直接投遞到kafka或者RocketMQ集羣，用戶只須要使用消息隊列的consumer消費便可。

5）成功消費消息後須要進行Ack，以確保一致性，服務端則會維護客戶端目前的消費位點。

3.工做原理

MySQL的主從複製分紅三步：

• master將改變記錄到二進制日誌(binary log)中（這些記錄叫作二進制日誌事件，binary log events，能夠經過show binlog events進行查看）；
• slave將master的binary log events拷貝到它的中繼日誌(relay log)；
• slave重作中繼日誌中的事件，將改變反映它本身的數據。

 

canal 就是模擬了這個過程。

• canal模擬 MySQL slave 的交互協議，假裝本身爲 MySQL slave ，向 MySQL master 發送 dump 協議；
• MySQL master 收到 dump 請求，開始推送 binary log 給 slave (即 canal )；
• canal 解析 binary log 對象(原始爲 byte 流)；

 

4. canal 架構

4.1 admin版本總體架構

canal 1.1.4開始支持admin管理，經過canal-admin爲canal提供總體配置管理、節點運維等面向運維的功能，提供相對友好的WebUI操做界面，方便更多用戶快速和安全的操做，替代了過去繁瑣的配置文件管理。

總體部署架構以下。

 

• 多個canal-server能夠組成集羣模式，每一個instance任務經過zookeeper在集羣中實現高可用
• 經過多個集羣，能夠實現同步資源的物理隔離
• 能夠直接抓取消費投遞MQ，能夠實現生產/消費解耦、消息堆積、消息回溯
• 能夠抓取消費投遞給canal-client，在用戶的服務中進行消息處理，減小中間過程

4.2 canal-server架構

 

說明：

• server表明一個canal-server運行實例，對應於一個jvm
• instance對應於一個數據隊列，是真正的變動抓取的實體 （1個server能夠對應多個instance)

Instance模塊

• EventParser ：數據源接入，模擬slave協議和master進行交互，協議解析
• EventSink ：Parser和Store連接器，進行數據過濾，加工，分發的工做
• EventStore ：數據存儲
• MetaManager：增量訂閱&消費信息管理器

1)EventParser子模塊

EventParser模塊的類圖設計以下

 

每一個EventParser都會關聯兩個內部組件：CanalLogPositionManager ， CanalHAController

• CanalLogPositionManager：記錄binlog最後一次解析成功位置信息，主要是描述下一次canal啓動的位點
• CanalHAController：支持Mysql主備，判斷當前該連哪一個mysql（基於Heartbeat實現，主庫失去心跳則連備庫）

EventParser根據HAController獲知連到哪裏，經過LogPositionManager獲知從哪一個位點開始解析，以後便經過Mysql Slave協議拉取binlog進行解析，推入EventSink

2）EventSink子模塊

目前只提供了一個帶有實際做用的實現：GroupEventSink

GroupEventSink用於將多個instance上的數據進行歸併，經常使用於分庫後的多數據源歸併。

3）EventStore子模塊

EventStore的類圖以下

 

官方提供的實現類是
MemoryEventStoreWIthBuffer，內部採用的是一個RingBuffer：

 

• Put : Sink模塊進行數據存儲的最後一次寫入位置
• Get : 數據訂閱獲取的最後一次提取位置
• Ack : 數據消費成功的最後一次消費位置

這些位點信息經過MetaManager進行管理。這也解釋了爲何一個canal instance只能支撐一個消費者：EventStore的RingBuffer只爲一個消費者維護信息。

4.3 客戶端使用

數據格式已經在前文給出，Canal和DTS客戶端均採起：

拉取事件 -> 消費 -> 消費成功後ACK

這樣的消費模式，並支持消費不成功時進行rollback，從新消費該數據。

下面是一段簡單的客戶端調用實例（略去異常處理）：

// 建立CanalConnector， 鏈接到localhost:11111

CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),11111), destination, "", "");

connector.connect(); // 鏈接

connector.subscribe(); // 開始訂閱binlog

// 開始循環拉取

while (running) {

 Message message = connector.getWithoutAck(1024); // 獲取指定數量的數據

 long batchId = message.getId();

 for (Entry entry : message.getEntries()){

 // 對每條消息進行處理

 }

 connector.ack(batchId); // ack

}

5.總結分析

5.1 優勢

1）性能優異、功能全面

• canal 1.1.x 版本（release_note）,性能與功能層面有較大的突破,重要提高包括:
• 總體性能測試&優化,提高了150%. #726
• 原生支持prometheus監控 #765
• 原生支持kafka消息投遞 #695
• 原生支持aliyun rds的binlog訂閱 (解決自動主備切換/oss binlog離線解析) （沒法拒絕它的理由！）
• 原生支持docker鏡像 #801

2）運維方便

• canal 1.1.4版本，迎來最重要的WebUI能力，引入canal-admin工程，支持面向WebUI的canal動態管理能力，支持配置、任務、日誌等在線白屏運維能力
• Standalone的一體化解決方案，無外部服務依賴，運維更簡單，在某種程度上也意味着更穩定。
• 開箱即用，節約開發與定製成本。
• 有良好的管理控制平臺與監控系統（若是你已經有promethus監控，能夠秒接canal監控）

3）多語言支持

• canal 特別設計了 client-server 模式，交互協議使用 protobuf 3.0 , client 端可採用不一樣語言實現不一樣的消費邏輯
• canal 做爲 MySQL binlog 增量獲取和解析工具，可將變動記錄投遞到 MQ 系統中，好比 Kafka/RocketMQ，能夠藉助於 MQ 的多語言能力

5.2 缺點

• 單instance/訂閱通道只支持訂閱單個數據庫，並只能支持單客戶端消費。每當咱們須要新增一個消費端->MySQL的訂閱：對於Canal而言，就要給MySQL接一個「Slave」，可能會對主庫有必定影響。
• 消息的Schema很弱，全部消息的Schema均相同，客戶端須要提早知道各個表消息的Schema與各字段的上下文才能正確消費。

好了，花了10分鐘應該對canal有大體瞭解了，下一期，阿丸計劃手把手教你搭建canal集羣和admin管理平臺，記得關注哦。

 

都看到最後了，原創不易，點個關注，點個贊吧～

知識碎片從新梳理，構建Java知識圖譜： github.com/saigu/JavaK…（歷史文章查閱很是方便）