canal是阿里巴巴旗下的一款開(kāi)源項(xiàng)目募寨，純Java開(kāi)發(fā)∩笤幔基于數(shù)據(jù)庫(kù)增量日志解析深滚，提供增量數(shù)據(jù)訂閱&消費(fèi)奕谭，目前主要支持了MySQL（也支持mariaDB）。
背景
早期痴荐，阿里巴巴B2B公司因?yàn)榇嬖诤贾莺兔绹?guó)雙機(jī)房部署血柳，存在跨機(jī)房同步的業(yè)務(wù)需求。不過(guò)早期的數(shù)據(jù)庫(kù)同步業(yè)務(wù)生兆，主要是基于trigger的方式獲取增量變更难捌，不過(guò)從2010年開(kāi)始，阿里系公司開(kāi)始逐步的嘗試基于數(shù)據(jù)庫(kù)的日志解析鸦难，獲取增量變更進(jìn)行同步根吁，由此衍生出了增量訂閱&消費(fèi)的業(yè)務(wù)，從此開(kāi)啟了一段新紀(jì)元合蔽。ps. 目前內(nèi)部使用的同步击敌，已經(jīng)支持mysql5.x和oracle部分版本的日志解析
基于日志增量訂閱&消費(fèi)支持的業(yè)務(wù)：
- 數(shù)據(jù)庫(kù)鏡像
- 數(shù)據(jù)庫(kù)實(shí)時(shí)備份
- 多級(jí)索引 (賣家和買家各自分庫(kù)索引)
- search build
- 業(yè)務(wù)cache刷新
- 價(jià)格變化等重要業(yè)務(wù)消息
項(xiàng)目介紹
名稱：canal [k?'n?l]
譯意： 水道/管道/溝渠
語(yǔ)言： 純java開(kāi)發(fā)
定位： 基于數(shù)據(jù)庫(kù)增量日志解析，提供增量數(shù)據(jù)訂閱&消費(fèi)辈末，目前主要支持了mysql
關(guān)鍵詞： mysql binlog parser / real-time / queue&topic
工作原理
mysql主備復(fù)制實(shí)現(xiàn)
從上層來(lái)看愚争，復(fù)制分成三步：
- master將改變記錄到二進(jìn)制日志(binary log)中（這些記錄叫做二進(jìn)制日志事件，binary log events挤聘，可以通過(guò)show binlog events進(jìn)行查看）；
- slave將master的binary log events拷貝到它的中繼日志(relay log)捅彻；
- slave重做中繼日志中的事件组去，將改變反映它自己的數(shù)據(jù)。
canal的工作原理：
原理相對(duì)比較簡(jiǎn)單：
- canal模擬mysql slave的交互協(xié)議步淹，偽裝自己為mysql slave从隆，向mysql master發(fā)送dump協(xié)議
- mysql master收到dump請(qǐng)求，開(kāi)始推送binary log給slave(也就是canal)
- canal解析binary log對(duì)象(原始為byte流)
架構(gòu)
說(shuō)明：
- server代表一個(gè)canal運(yùn)行實(shí)例缭裆，對(duì)應(yīng)于一個(gè)jvm
- instance對(duì)應(yīng)于一個(gè)數(shù)據(jù)隊(duì)列 （1個(gè)server對(duì)應(yīng)1..n個(gè)instance)
instance模塊：
- eventParser (數(shù)據(jù)源接入键闺，模擬slave協(xié)議和master進(jìn)行交互，協(xié)議解析)
- eventSink (Parser和Store鏈接器澈驼，進(jìn)行數(shù)據(jù)過(guò)濾辛燥，加工，分發(fā)的工作)
- eventStore (數(shù)據(jù)存儲(chǔ))
- metaManager (增量訂閱&消費(fèi)信息管理器)
知識(shí)科普
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)
簡(jiǎn)單點(diǎn)說(shuō)：
- mysql的binlog是多文件存儲(chǔ)缝其，定位一個(gè)LogEvent需要通過(guò)binlog filename + binlog position挎塌，進(jìn)行定位
- mysql的binlog數(shù)據(jù)格式，按照生成的方式内边，主要分為：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支持所有模式的增量訂閱(但配合同步時(shí)嘴高，因?yàn)閟tatement只有sql竿音，沒(méi)有數(shù)據(jù)，無(wú)法獲取原始的變更日志拴驮，所以一般建議為ROW模式)
EventParser設(shè)計(jì)
大致過(guò)程：
整個(gè)parser過(guò)程大致可分為幾步：
- Connection獲取上一次解析成功的位置 (如果第一次啟動(dòng)春瞬，則獲取初始指定的位置或者是當(dāng)前數(shù)據(jù)庫(kù)的binlog位點(diǎn))
- Connection建立鏈接，發(fā)送BINLOG_DUMP指令
// 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 - Mysql開(kāi)始推送Binaly Log
- 接收到的Binaly Log的通過(guò)Binlog parser進(jìn)行協(xié)議解析莹汤，補(bǔ)充一些特定信息
// 補(bǔ)充字段名字快鱼，字段類型，主鍵信息纲岭，unsigned類型處理 - 傳遞給EventSink模塊進(jìn)行數(shù)據(jù)存儲(chǔ)抹竹，是一個(gè)阻塞操作，直到存儲(chǔ)成功
- 存儲(chǔ)成功后止潮，定時(shí)記錄Binaly Log位置
EventSink設(shè)計(jì)
說(shuō)明：
- 數(shù)據(jù)過(guò)濾：支持通配符的過(guò)濾模式窃判，表名，字段內(nèi)容等
- 數(shù)據(jù)路由/分發(fā)：解決1:n (1個(gè)parser對(duì)應(yīng)多個(gè)store的模式)
- 數(shù)據(jù)歸并：解決n:1 (多個(gè)parser對(duì)應(yīng)1個(gè)store)
- 數(shù)據(jù)加工：在進(jìn)入store之前進(jìn)行額外的處理喇闸，比如join
數(shù)據(jù)1:n業(yè)務(wù)
為了合理的利用數(shù)據(jù)庫(kù)資源袄琳， 一般常見(jiàn)的業(yè)務(wù)都是按照schema進(jìn)行隔離，然后在mysql上層或者dao這一層面上燃乍，進(jìn)行一個(gè)數(shù)據(jù)源路由唆樊，屏蔽數(shù)據(jù)庫(kù)物理位置對(duì)開(kāi)發(fā)的影響，阿里系主要是通過(guò)cobar/tddl來(lái)解決數(shù)據(jù)源路由問(wèn)題刻蟹。
所以逗旁，一般一個(gè)數(shù)據(jù)庫(kù)實(shí)例上，會(huì)部署多個(gè)schema舆瘪，每個(gè)schema會(huì)有由1個(gè)或者多個(gè)業(yè)務(wù)方關(guān)注
數(shù)據(jù)n:1業(yè)務(wù)
同樣片效，當(dāng)一個(gè)業(yè)務(wù)的數(shù)據(jù)規(guī)模達(dá)到一定的量級(jí)后，必然會(huì)涉及到水平拆分和垂直拆分的問(wèn)題英古，針對(duì)這些拆分的數(shù)據(jù)需要處理時(shí)淀衣，就需要鏈接多個(gè)store進(jìn)行處理，消費(fèi)的位點(diǎn)就會(huì)變成多份召调，而且數(shù)據(jù)消費(fèi)的進(jìn)度無(wú)法得到盡可能有序的保證膨桥。
所以，在一定業(yè)務(wù)場(chǎng)景下某残，需要將拆分后的增量數(shù)據(jù)進(jìn)行歸并處理国撵，比如按照時(shí)間戳/全局id進(jìn)行排序歸并.
EventStore設(shè)計(jì)
- 1. 目前僅實(shí)現(xiàn)了Memory內(nèi)存模式，后續(xù)計(jì)劃增加本地file存儲(chǔ)玻墅，mixed混合模式
- 2. 借鑒了Disruptor的RingBuffer的實(shí)現(xiàn)思路
RingBuffer設(shè)計(jì)：
定義了3個(gè)cursor
- Put : Sink模塊進(jìn)行數(shù)據(jù)存儲(chǔ)的最后一次寫(xiě)入位置
- Get : 數(shù)據(jù)訂閱獲取的最后一次提取位置
- Ack : 數(shù)據(jù)消費(fèi)成功的最后一次消費(fèi)位置
借鑒Disruptor的RingBuffer的實(shí)現(xiàn)介牙，將RingBuffer拉直來(lái)看：
實(shí)現(xiàn)說(shuō)明：
- Put/Get/Ack cursor用于遞增，采用long型存儲(chǔ)
- buffer的get操作澳厢，通過(guò)取余或者與操作环础。(與操作： cusor & (size - 1) , size需要為2的指數(shù)囚似，效率比較高)
Instance設(shè)計(jì)
instance代表了一個(gè)實(shí)際運(yùn)行的數(shù)據(jù)隊(duì)列，包括了EventPaser,EventSink,EventStore等組件线得。
抽象了CanalInstanceGenerator饶唤，主要是考慮配置的管理方式：
- manager方式： 和你自己的內(nèi)部web console/manager系統(tǒng)進(jìn)行對(duì)接。(目前主要是公司內(nèi)部使用)
- spring方式：基于spring xml + properties進(jìn)行定義贯钩，構(gòu)建spring配置.
Server設(shè)計(jì)
server代表了一個(gè)canal的運(yùn)行實(shí)例募狂，為了方便組件化使用，特意抽象了Embeded(嵌入式) / Netty(網(wǎng)絡(luò)訪問(wèn))的兩種實(shí)現(xiàn)
- Embeded : 對(duì)latency和可用性都有比較高的要求角雷，自己又能hold住分布式的相關(guān)技術(shù)(比如failover)
- Netty : 基于netty封裝了一層網(wǎng)絡(luò)協(xié)議祸穷，由canal server保證其可用性，采用的pull模型勺三，當(dāng)然latency會(huì)稍微打點(diǎn)折扣雷滚，不過(guò)這個(gè)也視情況而定。(阿里系的notify和metaq吗坚，典型的push/pull模型祈远，目前也逐步的在向pull模型靠攏，push在數(shù)據(jù)量大的時(shí)候會(huì)有一些問(wèn)題)
增量訂閱/消費(fèi)設(shè)計(jì)
具體的協(xié)議格式商源，可參見(jiàn)：CanalProtocol.proto
get/ack/rollback協(xié)議介紹：
- Message getWithoutAck(int batchSize)车份，允許指定batchSize，一次可以獲取多條牡彻，每次返回的對(duì)象為Message躬充，包含的內(nèi)容為：
a. batch id 唯一標(biāo)識(shí)
b. entries 具體的數(shù)據(jù)對(duì)象，對(duì)應(yīng)的數(shù)據(jù)對(duì)象格式：EntryProtocol.proto - void rollback(long batchId)讨便，顧命思議，回滾上次的get請(qǐng)求以政，重新獲取數(shù)據(jù)霸褒。基于get獲取的batchId進(jìn)行提交盈蛮，避免誤操作
- void ack(long batchId)废菱，顧命思議，確認(rèn)已經(jīng)消費(fèi)成功抖誉，通知server刪除數(shù)據(jù)殊轴。基于get獲取的batchId進(jìn)行提交袒炉，避免誤操作
canal的get/ack/rollback協(xié)議和常規(guī)的jms協(xié)議有所不同旁理，允許get/ack異步處理，比如可以連續(xù)調(diào)用get多次我磁，后續(xù)異步按順序提交ack/rollback孽文，項(xiàng)目中稱之為流式api.
流式api設(shè)計(jì)的好處：
- get/ack異步化驻襟，減少因ack帶來(lái)的網(wǎng)絡(luò)延遲和操作成本 (99%的狀態(tài)都是處于正常狀態(tài)，異常的rollback屬于個(gè)別情況芋哭，沒(méi)必要為個(gè)別的case犧牲整個(gè)性能)
- get獲取數(shù)據(jù)后沉衣，業(yè)務(wù)消費(fèi)存在瓶頸或者需要多進(jìn)程/多線程消費(fèi)時(shí)，可以不停的輪詢get數(shù)據(jù)减牺，不停的往后發(fā)送任務(wù)豌习，提高并行化. (作者在實(shí)際業(yè)務(wù)中的一個(gè)case：業(yè)務(wù)數(shù)據(jù)消費(fèi)需要跨中美網(wǎng)絡(luò)，所以一次操作基本在200ms以上拔疚，為了減少延遲肥隆，所以需要實(shí)施并行化)
流式api設(shè)計(jì)：
- 每次get操作都會(huì)在meta中產(chǎn)生一個(gè)mark，mark標(biāo)記會(huì)遞增草雕，保證運(yùn)行過(guò)程中mark的唯一性
- 每次的get操作巷屿，都會(huì)在上一次的mark操作記錄的cursor繼續(xù)往后取，如果mark不存在墩虹，則在last ack cursor繼續(xù)往后取
- 進(jìn)行ack時(shí)嘱巾，需要按照mark的順序進(jìn)行數(shù)序ack，不能跳躍ack. ack會(huì)刪除當(dāng)前的mark標(biāo)記诫钓，并將對(duì)應(yīng)的mark位置更新為last ack cusor
- 一旦出現(xiàn)異常情況旬昭，客戶端可發(fā)起rollback情況，重新置位：刪除所有的mark, 清理get請(qǐng)求位置菌湃，下次請(qǐng)求會(huì)從last ack cursor繼續(xù)往后取
數(shù)據(jù)對(duì)象格式：EntryProtocol.proto
Entry
Header
logfileName [binlog文件名]
logfileOffset [binlog position]
executeTime [binlog里記錄變更發(fā)生的時(shí)間戳]
schemaName [數(shù)據(jù)庫(kù)實(shí)例]
tableName [表名]
eventType [insert/update/delete類型]
entryType [事務(wù)頭BEGIN/事務(wù)尾END/數(shù)據(jù)ROWDATA]
storeValue [byte數(shù)據(jù),可展開(kāi)问拘，對(duì)應(yīng)的類型為RowChange]
RowChange
isDdl [是否是ddl變更操作，比如create table/drop table]
sql [具體的ddl sql]
rowDatas [具體insert/update/delete的變更數(shù)據(jù)惧所，可為多條骤坐，1個(gè)binlog event事件可對(duì)應(yīng)多條變更，比如批處理]
beforeColumns [Column類型的數(shù)組]
afterColumns [Column類型的數(shù)組]
Column
index [column序號(hào)]
sqlType [jdbc type]
name [column name]
isKey [是否為主鍵]
updated [是否發(fā)生過(guò)變更]
isNull [值是否為null]
value [具體的內(nèi)容下愈，注意為文本]
說(shuō)明：
- 可以提供數(shù)據(jù)庫(kù)變更前和變更后的字段內(nèi)容纽绍，針對(duì)binlog中沒(méi)有的name,isKey等信息進(jìn)行補(bǔ)全
- 可以提供ddl的變更語(yǔ)句
HA機(jī)制設(shè)計(jì)
canal的ha分為兩部分，canal server和canal client分別有對(duì)應(yīng)的ha實(shí)現(xiàn)
- canal server: 為了減少對(duì)mysql dump的請(qǐng)求势似，不同server上的instance要求同一時(shí)間只能有一個(gè)處于running拌夏，其他的處于standby狀態(tài).
- canal client: 為了保證有序性，一份instance同一時(shí)間只能由一個(gè)canal client進(jìn)行g(shù)et/ack/rollback操作履因，否則客戶端接收無(wú)法保證有序障簿。
整個(gè)HA機(jī)制的控制主要是依賴了zookeeper的幾個(gè)特性，watcher和EPHEMERAL節(jié)點(diǎn)(和session生命周期綁定)栅迄，可以看下我之前zookeeper的相關(guān)文章站故。
Canal Server:
大致步驟：
- canal server要啟動(dòng)某個(gè)canal instance時(shí)都先向zookeeper進(jìn)行一次嘗試啟動(dòng)判斷 (實(shí)現(xiàn)：創(chuàng)建EPHEMERAL節(jié)點(diǎn)，誰(shuí)創(chuàng)建成功就允許誰(shuí)啟動(dòng))
- 創(chuàng)建zookeeper節(jié)點(diǎn)成功后霞篡，對(duì)應(yīng)的canal server就啟動(dòng)對(duì)應(yīng)的canal instance世蔗，沒(méi)有創(chuàng)建成功的canal instance就會(huì)處于standby狀態(tài)
- 一旦zookeeper發(fā)現(xiàn)canal server A創(chuàng)建的節(jié)點(diǎn)消失后端逼，立即通知其他的canal server再次進(jìn)行步驟1的操作，重新選出一個(gè)canal server啟動(dòng)instance.
- canal client每次進(jìn)行connect時(shí)污淋，會(huì)首先向zookeeper詢問(wèn)當(dāng)前是誰(shuí)啟動(dòng)了canal instance顶滩，然后和其建立鏈接，一旦鏈接不可用寸爆，會(huì)重新嘗試connect.
Canal Client的方式和canal server方式類似礁鲁，也是利用zookeeper的搶占EPHEMERAL節(jié)點(diǎn)的方式進(jìn)行控制.
