阿里开源Canal--①简介
canal是阿里巴巴旗下的一款開源項目,純Java開發。基于數據庫增量日志解析,提供增量數據訂閱&消費,目前主要支持了MySQL(也支持mariaDB)。
背景
早期,阿里巴巴B2B公司因為存在杭州和美國雙機房部署,存在跨機房同步的業務需求。不過早期的數據庫同步業務,主要是基于trigger的方式獲取增量變更,不過從2010年開始,阿里系公司開始逐步的嘗試基于數據庫的日志解析,獲取增量變更進行同步,由此衍生出了增量訂閱&消費的業務,從此開啟了一段新紀元。ps. 目前內部使用的同步,已經支持mysql5.x和oracle部分版本的日志解析
基于日志增量訂閱&消費支持的業務:
項目介紹
名稱:canal [k?'n?l]
譯意: 水道/管道/溝渠
語言: 純java開發
定位: 基于數據庫增量日志解析,提供增量數據訂閱&消費,目前主要支持了mysql
關鍵詞: mysql binlog parser / real-time / queue&topic
工作原理
mysql主備復制實現
?
從上層來看,復制分成三步:
canal的工作原理:
?
原理相對比較簡單:
架構
?
說明:
- server代表一個canal運行實例,對應于一個jvm
- instance對應于一個數據隊列 (1個server對應1..n個instance)
instance模塊:
- eventParser (數據源接入,模擬slave協議和master進行交互,協議解析)
- eventSink (Parser和Store鏈接器,進行數據過濾,加工,分發的工作)
- eventStore (數據存儲)
- metaManager (增量訂閱&消費信息管理器)
知識科普
mysql的Binlay Log介紹
- (http://dev.mysql.com/doc/refman/5.5/en/binary-log.html)
- (http://www.taobaodba.com/html/474_mysqls-binary-log_details.html)
簡單點說:
- mysql的binlog是多文件存儲,定位一個LogEvent需要通過binlog filename + binlog position,進行定位
- mysql的binlog數據格式,按照生成的方式,主要分為:statement-based、row-based、mixed。
?
mysql> show variables like 'binlog_format';+---------------+-------+| Variable_name | Value |+---------------+-------+| binlog_format | ROW |+---------------+-------+1 row in set (0.00 sec)目前canal支持所有模式的增量訂閱(但配合同步時,因為statement只有sql,沒有數據,無法獲取原始的變更日志,所以一般建議為ROW模式)
EventParser設計
大致過程:
?
整個parser過程大致可分為幾步:
// 0. write command number
// 1. write 4 bytes bin-log position to start at
// 2. write 2 bytes bin-log flags
// 3. write 4 bytes server id of the slave
// 4. write bin-log file name
// 補充字段名字,字段類型,主鍵信息,unsigned類型處理
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進行排序歸并.
EventStore設計
- 1. 目前僅實現了Memory內存模式,后續計劃增加本地file存儲,mixed混合模式
- 2. 借鑒了Disruptor的RingBuffer的實現思路
RingBuffer設計:
?
定義了3個cursor
- Put : Sink模塊進行數據存儲的最后一次寫入位置
- Get : 數據訂閱獲取的最后一次提取位置
- Ack : 數據消費成功的最后一次消費位置
借鑒Disruptor的RingBuffer的實現,將RingBuffer拉直來看:
?
?
image.png
實現說明:
- Put/Get/Ack cursor用于遞增,采用long型存儲
- 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(網絡訪問)的兩種實現
- Embeded : 對latency和可用性都有比較高的要求,自己又能hold住分布式的相關技術(比如failover)
- Netty : 基于netty封裝了一層網絡協議,由canal server保證其可用性,采用的pull模型,當然latency會稍微打點折扣,不過這個也視情況而定。(阿里系的notify和metaq,典型的push/pull模型,目前也逐步的在向pull模型靠攏,push在數據量大的時候會有一些問題)
增量訂閱/消費設計
?
具體的協議格式,可參見:CanalProtocol.proto
get/ack/rollback協議介紹:
- Message getWithoutAck(int batchSize),允許指定batchSize,一次可以獲取多條,每次返回的對象為Message,包含的內容為:
a. batch id 唯一標識
b. entries 具體的數據對象,對應的數據對象格式:EntryProtocol.proto - void rollback(long batchId),顧命思議,回滾上次的get請求,重新獲取數據。基于get獲取的batchId進行提交,避免誤操作
- void ack(long batchId),顧命思議,確認已經消費成功,通知server刪除數據。基于get獲取的batchId進行提交,避免誤操作
canal的get/ack/rollback協議和常規的jms協議有所不同,允許get/ack異步處理,比如可以連續調用get多次,后續異步按順序提交ack/rollback,項目中稱之為流式api.
流式api設計的好處:
- get/ack異步化,減少因ack帶來的網絡延遲和操作成本 (99%的狀態都是處于正常狀態,異常的rollback屬于個別情況,沒必要為個別的case犧牲整個性能)
- get獲取數據后,業務消費存在瓶頸或者需要多進程/多線程消費時,可以不停的輪詢get數據,不停的往后發送任務,提高并行化. (作者在實際業務中的一個case:業務數據消費需要跨中美網絡,所以一次操作基本在200ms以上,為了減少延遲,所以需要實施并行化)
流式api設計:
?
- 每次get操作都會在meta中產生一個mark,mark標記會遞增,保證運行過程中mark的唯一性
- 每次的get操作,都會在上一次的mark操作記錄的cursor繼續往后取,如果mark不存在,則在last ack cursor繼續往后取
- 進行ack時,需要按照mark的順序進行數序ack,不能跳躍ack. ack會刪除當前的mark標記,并將對應的mark位置更新為last ack cusor
- 一旦出現異常情況,客戶端可發起rollback情況,重新置位:刪除所有的mark, 清理get請求位置,下次請求會從last ack cursor繼續往后取
數據對象格式:EntryProtocol.proto
?
EntryHeaderlogfileName [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 [具體的內容,注意為文本]說明:
- 可以提供數據庫變更前和變更后的字段內容,針對binlog中沒有的name,isKey等信息進行補全
- 可以提供ddl的變更語句
HA機制設計
canal的ha分為兩部分,canal server和canal client分別有對應的ha實現
- canal server: 為了減少對mysql dump的請求,不同server上的instance要求同一時間只能有一個處于running,其他的處于standby狀態.
- canal client: 為了保證有序性,一份instance同一時間只能由一個canal client進行get/ack/rollback操作,否則客戶端接收無法保證有序。
整個HA機制的控制主要是依賴了zookeeper的幾個特性,watcher和EPHEMERAL節點(和session生命周期綁定),可以看下我之前zookeeper的相關文章。
Canal Server:
?
image.png
大致步驟:
Canal Client的方式和canal server方式類似,也是利用zookeeper的搶占EPHEMERAL節點的方式進行控制.
作者:擼碼小丑
鏈接:https://www.jianshu.com/p/87944efe1005
來源:簡書
著作權歸作者所有。商業轉載請聯系作者獲得授權,非商業轉載請注明出處。
來源:https://www.jianshu.com/p/87944efe1005
總結
以上是生活随笔為你收集整理的阿里开源Canal--①简介的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: 分期是怎么分期的
- 下一篇: IDEA的maven项目报错BeanCr