阿里巴巴開源項目: 基于mysql數(shù)據(jù)庫binlog的增量訂閱&消費

2016.07.05

背景

早期，阿里巴巴B2B公司因為存在杭州和美國雙機房部署，存在跨機房同步的業(yè)務需求。不過早期的數(shù)據(jù)庫同步業(yè)務，主要是基于trigger的方式獲取增量變更，不過從2010年開始，阿里系公司開始逐步的嘗試基于數(shù)據(jù)庫的日志解析，獲取增量變更進行同步，由此衍生出了增量訂閱&消費的業(yè)務，從此開啟了一段新紀元。ps. 目前內(nèi)部使用的同步，已經(jīng)支持mysql5.x和oracle部分版本的日志解析

基于日志增量訂閱&消費支持的業(yè)務：

數(shù)據(jù)庫鏡像
數(shù)據(jù)庫實時備份
多級索引 (賣家和買家各自分庫索引)
search build
業(yè)務cache刷新
價格變化等重要業(yè)務消息

項目介紹

名稱：canal [k?'n?l]

譯意：水道/管道/溝渠

語言：純java開發(fā)

定位：基于數(shù)據(jù)庫增量日志解析，提供增量數(shù)據(jù)訂閱&消費，目前主要支持了mysql

工作原理

mysql主備復制實現(xiàn)

從上層來看，復制分成三步：

master將改變記錄到二進制日志(binary log)中（這些記錄叫做二進制日志事件，binary log events，可以通過show binlog events進行查看）；
slave將master的binary log events拷貝到它的中繼日志(relay log)；
slave重做中繼日志中的事件，將改變反映它自己的數(shù)據(jù)。

canal的工作原理：

原理相對比較簡單：

canal模擬mysql slave的交互協(xié)議，偽裝自己為mysql slave，向mysql master發(fā)送dump協(xié)議
mysql master收到dump請求，開始推送binary log給slave(也就是canal)
canal解析binary log對象(原始為byte流)

架構(gòu)

說明：

server代表一個canal運行實例，對應于一個jvm
instance對應于一個數(shù)據(jù)隊列（1個server對應1..n個instance)

instance模塊：

eventParser (數(shù)據(jù)源接入，模擬slave協(xié)議和master進行交互，協(xié)議解析)
eventSink (Parser和Store鏈接器，進行數(shù)據(jù)過濾，加工，分發(fā)的工作)
eventStore (數(shù)據(jù)存儲)
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數(shù)據(jù)格式，按照生成的方式，主要分為：statement-based、row-based、mixed。
Java代碼
1. mysql> show variables like 'binlog_format';
2. +---------------+-------+
3. | Variable_name | Value |
4. +---------------+-------+
5. | binlog_format | ROW |
6. +---------------+-------+
7. 1 row in set (0.00 sec)

目前canal只能支持row模式的增量訂閱(statement只有sql，沒有數(shù)據(jù)，所以無法獲取原始的變更日志)

EventParser設(shè)計

大致過程：

整個parser過程大致可分為幾步：

Connection獲取上一次解析成功的位置 (如果第一次啟動，則獲取初始指定的位置或者是當前數(shù)據(jù)庫的binlog位點)
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開始推送Binaly Log
接收到的Binaly Log的通過Binlog parser進行協(xié)議解析，補充一些特定信息
// 補充字段名字，字段類型，主鍵信息，unsigned類型處理
傳遞給EventSink模塊進行數(shù)據(jù)存儲，是一個阻塞操作，直到存儲成功
存儲成功后，定時記錄Binaly Log位置

mysql的Binlay Log網(wǎng)絡(luò)協(xié)議：

說明：

圖中的協(xié)議4byte header，主要是描述整個binlog網(wǎng)絡(luò)包的length
binlog event structure，詳細信息請參考： http://dev.mysql.com/doc/internals/en/binary-log.html

EventSink設(shè)計

說明：

數(shù)據(jù)過濾：支持通配符的過濾模式，表名，字段內(nèi)容等
數(shù)據(jù)路由/分發(fā)：解決1:n (1個parser對應多個store的模式)
數(shù)據(jù)歸并：解決n:1 (多個parser對應1個store)
數(shù)據(jù)加工：在進入store之前進行額外的處理，比如join

數(shù)據(jù)1:n業(yè)務

為了合理的利用數(shù)據(jù)庫資源，一般常見的業(yè)務都是按照schema進行隔離，然后在mysql上層或者dao這一層面上，進行一個數(shù)據(jù)源路由，屏蔽數(shù)據(jù)庫物理位置對開發(fā)的影響，阿里系主要是通過cobar/tddl來解決數(shù)據(jù)源路由問題。

所以，一般一個數(shù)據(jù)庫實例上，會部署多個schema，每個schema會有由1個或者多個業(yè)務方關(guān)注

數(shù)據(jù)n:1業(yè)務

同樣，當一個業(yè)務的數(shù)據(jù)規(guī)模達到一定的量級后，必然會涉及到水平拆分和垂直拆分的問題，針對這些拆分的數(shù)據(jù)需要處理時，就需要鏈接多個store進行處理，消費的位點就會變成多份，而且數(shù)據(jù)消費的進度無法得到盡可能有序的保證。

所以，在一定業(yè)務場景下，需要將拆分后的增量數(shù)據(jù)進行歸并處理，比如按照時間戳/全局id進行排序歸并.

EventStore設(shè)計

1. 目前僅實現(xiàn)了Memory內(nèi)存模式，后續(xù)計劃增加本地file存儲，mixed混合模式
2. 借鑒了Disruptor的RingBuffer的實現(xiàn)思路

RingBuffer設(shè)計：

定義了3個cursor

Put : Sink模塊進行數(shù)據(jù)存儲的最后一次寫入位置
Get : 數(shù)據(jù)訂閱獲取的最后一次提取位置
Ack : 數(shù)據(jù)消費成功的最后一次消費位置

借鑒Disruptor的RingBuffer的實現(xiàn)，將RingBuffer拉直來看：

實現(xiàn)說明：

Put/Get/Ack cursor用于遞增，采用long型存儲
buffer的get操作，通過取余或者與操作。(與操作： cusor & (size - 1) , size需要為2的指數(shù)，效率比較高)

Instance設(shè)計

instance代表了一個實際運行的數(shù)據(jù)隊列，包括了EventPaser,EventSink,EventStore等組件。

抽象了CanalInstanceGenerator，主要是考慮配置的管理方式：

manager方式：和你自己的內(nèi)部web console/manager系統(tǒng)進行對接。(目前主要是公司內(nèi)部使用)
spring方式：基于spring xml + properties進行定義，構(gòu)建spring配置.

Server設(shè)計

server代表了一個canal的運行實例，為了方便組件化使用，特意抽象了Embeded(嵌入式) / Netty(網(wǎng)絡(luò)訪問)的兩種實現(xiàn)

Embeded : 對latency和可用性都有比較高的要求，自己又能hold住分布式的相關(guān)技術(shù)(比如failover)
Netty : 基于netty封裝了一層網(wǎng)絡(luò)協(xié)議，由canal server保證其可用性，采用的pull模型，當然latency會稍微打點折扣，不過這個也視情況而定。(阿里系的notify和metaq，典型的push/pull模型，目前也逐步的在向pull模型靠攏，push在數(shù)據(jù)量大的時候會有一些問題)

增量訂閱/消費設(shè)計

具體的協(xié)議格式，可參見：CanalProtocol.proto

get/ack/rollback協(xié)議介紹：

Message getWithoutAck(int batchSize)，允許指定batchSize，一次可以獲取多條，每次返回的對象為Message，包含的內(nèi)容為：
a. batch id 唯一標識
b. entries 具體的數(shù)據(jù)對象，對應的數(shù)據(jù)對象格式：EntryProtocol.proto
void rollback(long batchId)，顧命思議，回滾上次的get請求，重新獲取數(shù)據(jù)?；趃et獲取的batchId進行提交，避免誤操作
void ack(long batchId)，顧命思議，確認已經(jīng)消費成功，通知server刪除數(shù)據(jù)。基于get獲取的batchId進行提交，避免誤操作

canal的get/ack/rollback協(xié)議和常規(guī)的jms協(xié)議有所不同，允許get/ack異步處理，比如可以連續(xù)調(diào)用get多次，后續(xù)異步按順序提交ack/rollback，項目中稱之為流式api.

流式api設(shè)計的好處：

get/ack異步化，減少因ack帶來的網(wǎng)絡(luò)延遲和操作成本 (99%的狀態(tài)都是處于正常狀態(tài)，異常的rollback屬于個別情況，沒必要為個別的case犧牲整個性能)
get獲取數(shù)據(jù)后，業(yè)務消費存在瓶頸或者需要多進程/多線程消費時，可以不停的輪詢get數(shù)據(jù)，不停的往后發(fā)送任務，提高并行化. (作者在實際業(yè)務中的一個case：業(yè)務數(shù)據(jù)消費需要跨中美網(wǎng)絡(luò)，所以一次操作基本在200ms以上，為了減少延遲，所以需要實施并行化)

流式api設(shè)計：

每次get操作都會在meta中產(chǎn)生一個mark，mark標記會遞增，保證運行過程中mark的唯一性
每次的get操作，都會在上一次的mark操作記錄的cursor繼續(xù)往后取，如果mark不存在，則在last ack cursor繼續(xù)往后取
進行ack時，需要按照mark的順序進行數(shù)序ack，不能跳躍ack. ack會刪除當前的mark標記，并將對應的mark位置更新為last ack cusor
一旦出現(xiàn)異常情況，客戶端可發(fā)起rollback情況，重新置位：刪除所有的mark, 清理get請求位置，下次請求會從last ack cursor繼續(xù)往后取

數(shù)據(jù)對象格式：EntryProtocol.proto

Java代碼

Entry
Header
logfileName [binlog文件名]
logfileOffset [binlog position]
executeTime [發(fā)生的變更]
schemaName
tableName
eventType [insert/update/delete類型]
entryType [事務頭BEGIN/事務尾END/數(shù)據(jù)ROWDATA]
storeValue [byte數(shù)據(jù),可展開，對應的類型為RowChange]
RowChange
isDdl [是否是ddl變更操作，比如create table/drop table]
sql [具體的ddl sql]
rowDatas [具體insert/update/delete的變更數(shù)據(jù)，可為多條，1個binlog event事件可對應多條變更，比如批處理]
beforeColumns [Column類型的數(shù)組]
afterColumns [Column類型的數(shù)組]
Column
index
sqlType [jdbc type]
name [column name]
isKey [是否為主鍵]
updated [是否發(fā)生過變更]
isNull [值是否為null]
value [具體的內(nèi)容，注意為文本]

說明：

可以提供數(shù)據(jù)庫變更前和變更后的字段內(nèi)容，針對binlog中沒有的name,isKey等信息進行補全
可以提供ddl的變更語句

HA機制設(shè)計

canal的ha分為兩部分，canal server和canal client分別有對應的ha實現(xiàn)

canal server: 為了減少對mysql dump的請求，不同server上的instance要求同一時間只能有一個處于running，其他的處于standby狀態(tài).
canal client: 為了保證有序性，一份instance同一時間只能由一個canal client進行g(shù)et/ack/rollback操作，否則客戶端接收無法保證有序。

整個HA機制的控制主要是依賴了zookeeper的幾個特性，watcher和EPHEMERAL節(jié)點(和session生命周期綁定)，可以看下我之前zookeeper的相關(guān)文章。

Canal Server:

大致步驟：

canal server要啟動某個canal instance時都先向zookeeper進行一次嘗試啟動判斷 (實現(xiàn)：創(chuàng)建EPHEMERAL節(jié)點，誰創(chuàng)建成功就允許誰啟動)
創(chuàng)建zookeeper節(jié)點成功后，對應的canal server就啟動對應的canal instance，沒有創(chuàng)建成功的canal instance就會處于standby狀態(tài)
一旦zookeeper發(fā)現(xiàn)canal server A創(chuàng)建的節(jié)點消失后，立即通知其他的canal server再次進行步驟1的操作，重新選出一個canal server啟動instance.
canal client每次進行connect時，會首先向zookeeper詢問當前是誰啟動了canal instance，然后和其建立鏈接，一旦鏈接不可用，會重新嘗試connect.

Canal Client的方式和canal server方式類似，也是利用zokeeper的搶占EPHEMERAL節(jié)點的方式進行控制.