Canal是阿里巴巴旗下的一款開源項目,利用Java開發(fā)。主要用途是基于MySQL數(shù)據(jù)庫增量日志解析,提供增量數(shù)據(jù)訂閱和消費(fèi),目前主要支持MySQL。
GitHub地址:https://github.com/alibaba/canal
在介紹Canal內(nèi)部原理之前,首先來了解一下MySQL Master/Slave同步原理:
MySQL master啟動binlog機(jī)制,將數(shù)據(jù)變更寫入二進(jìn)制日志(binary log, 其中記錄叫做二進(jìn)制日志事件binary log events,可以通過show binlog events進(jìn)行查看)MySQL slave(I/O thread)將master的binary log events拷貝到它的中繼日志(relay log)MySQL slave(SQL thread)重放relay log中事件,將數(shù)據(jù)變更反映它自己的數(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流)
簡而言之,Canal是通過模擬成為MySQL的slave,監(jiān)聽MySQL的binlog日志來獲取數(shù)據(jù)。當(dāng)把MySQL的binlog設(shè)置為row模式以后,可以獲取到執(zhí)行的每一個Insert/Update/Delete的腳本,以及修改前和修改后的數(shù)據(jù),基于這個特性,Canal就能高效的獲取到MySQL數(shù)據(jù)的變更。 Canal架構(gòu):
說明: server代表一個Canal運(yùn)行實(shí)例,對應(yīng)于一個jvm instance對應(yīng)于一個數(shù)據(jù)隊列(1個server對應(yīng)1..n個instance)
EventParser:數(shù)據(jù)源接入,模擬slave協(xié)議和master進(jìn)行交互,協(xié)議解析
EventSink:Parser和Store連接器,主要進(jìn)行數(shù)據(jù)過濾,加工,分發(fā)的工作
EventStore:負(fù)責(zé)存儲
MemoryMetaManager:增量訂閱和消費(fèi)信息管理器
Event Parser設(shè)計:
整個parser過程大致可分為以下幾步:
Connection獲取上一次解析成功的log position(如果是第一次啟動,則獲取初始指定的位置或者是當(dāng)前數(shù)據(jù)庫的binlog log position)Connection建立連接,向MySQL master發(fā)送BINLOG_DUMP請求MySQL開始推送binary Log接收到的binary Log通過BinlogParser進(jìn)行協(xié)議解析,補(bǔ)充一些特定信息。如補(bǔ)充字段名字、字段類型、主鍵信息、unsigned類型處理等將解析后的數(shù)據(jù)傳入到EventSink組件進(jìn)行數(shù)據(jù)存儲(這是一個阻塞操作,直到存儲成功)定時記錄binary Log位置,以便重啟后繼續(xù)進(jìn)行增量訂閱
如果需要同步的master宕機(jī),可以從它的其他slave節(jié)點(diǎn)繼續(xù)同步binlog日志,避免單點(diǎn)故障。 Event Sink設(shè)計:
EventSink主要作用如下:
數(shù)據(jù)過濾:支持通配符的過濾模式,表名,字段內(nèi)容等
數(shù)據(jù)路由/分發(fā):解決1:n(1個parser對應(yīng)多個store的模式)
數(shù)據(jù)歸并:解決n:1(多個parser對應(yīng)1個store)
數(shù)據(jù)加工:在進(jìn)入store之前進(jìn)行額外的處理,比如join 數(shù)據(jù)1:n業(yè)務(wù)
為了合理的利用數(shù)據(jù)庫資源, 一般常見的業(yè)務(wù)都是按照schema進(jìn)行隔離,然后在MySQL上層或者dao這一層面上,進(jìn)行一個數(shù)據(jù)源路由,屏蔽數(shù)據(jù)庫物理位置對開發(fā)的影響,阿里系主要是通過cobar/tddl來解決數(shù)據(jù)源路由問題。所以,一般一個數(shù)據(jù)庫實(shí)例上,會部署多個schema,每個schema會有由1個或者多個業(yè)務(wù)方關(guān)注。
數(shù)據(jù)n:1業(yè)務(wù)
同樣,當(dāng)一個業(yè)務(wù)的數(shù)據(jù)規(guī)模達(dá)到一定的量級后,必然會涉及到水平拆分和垂直拆分的問題,針對這些拆分的數(shù)據(jù)需要處理時,就需要鏈接多個store進(jìn)行處理,消費(fèi)的位點(diǎn)就會變成多份,而且數(shù)據(jù)消費(fèi)的進(jìn)度無法得到盡可能有序的保證。所以,在一定業(yè)務(wù)場景下,需要將拆分后的增量數(shù)據(jù)進(jìn)行歸并處理,比如按照時間戳/全局id進(jìn)行排序歸并。 Event Store設(shè)計:
支持多種存儲模式,比如Memory內(nèi)存模式。采用內(nèi)存環(huán)裝的設(shè)計來保存消息,借鑒了Disruptor的RingBuffer的實(shí)現(xiàn)思路。 RingBuffer設(shè)計:
定義了3個cursor:
put:Sink模塊進(jìn)行數(shù)據(jù)存儲的最后一次寫入位置(同步寫入數(shù)據(jù)的cursor)
get:數(shù)據(jù)訂閱獲取的最后一次提取位置(同步獲取的數(shù)據(jù)的cursor)
ack:數(shù)據(jù)消費(fèi)成功的最后一次消費(fèi)位置
借鑒Disruptor的RingBuffer的實(shí)現(xiàn),將RingBuffer拉直來看:
實(shí)現(xiàn)說明:
put/get/ack cursor用于遞增,采用long型存儲。三者之間的關(guān)系為put>=get>=ackbuffer的get操作,通過取余或者&操作。(&操作:cusor & (size - 1) , size需要為2的指數(shù),效率比較高)
Instance設(shè)計:
instance代表了一個實(shí)際運(yùn)行的數(shù)據(jù)隊列,包括了EventPaser、EventSink、EventStore等組件。抽象了CanalInstanceGenerator,主要是考慮配置的管理方式:
manager方式:和你自己的內(nèi)部web console/manager系統(tǒng)進(jìn)行對接。(目前主要是公司內(nèi)部使用)
spring方式:基于spring xml + properties進(jìn)行定義,構(gòu)建spring配置。 Server設(shè)計:
server代表了一個Canal運(yùn)行實(shí)例,為了方便組件化使用,特意抽象了Embeded(嵌入式)/Netty(網(wǎng)絡(luò)訪問)的兩種實(shí)現(xiàn)。
增量訂閱/消費(fèi)設(shè)計:
具體的協(xié)議格式,可參見:CanalProtocol.proto。數(shù)據(jù)對象格式:EntryProtocol.proto
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
Entry Header logfileName [binlog文件名] logfileOffset [binlog position] executeTime [binlog里記錄變更發(fā)生的時間戳] schemaName [數(shù)據(jù)庫實(shí)例] tableName [表名] eventType [ insert / update / delete 類型] entryType [事務(wù)頭 BEGIN /事務(wù)尾 END /數(shù)據(jù)ROWDATA] storeValue [byte數(shù)據(jù),可展開,對應(yīng)的類型為RowChange] RowChange isDdl [是否是ddl變更操作,比如 create table / drop table ] sql [具體的ddl sql] rowDatas [具體 insert / update / delete 的變更數(shù)據(jù),可為多條,1個binlog event事件可對應(yīng)多條變更,比如批處理] beforeColumns [ Column 類型的數(shù)組] afterColumns [ Column 類型的數(shù)組] Column index [ column 序號] sqlType [jdbc type] name [ column name ] isKey [是否為主鍵] updated [是否發(fā)生過變更] isNull [值是否為 null ] value [具體的內(nèi)容,注意為文本] |
針對上述的補(bǔ)充說明:
1.可以提供數(shù)據(jù)庫變更前和變更后的字段內(nèi)容,針對binlog中沒有的name、isKey等信息進(jìn)行補(bǔ)全
2.可以提供ddl的變更語句
Canal HA機(jī)制:
Canal的HA實(shí)現(xiàn)機(jī)制是依賴zookeeper實(shí)現(xiàn)的,主要分為Canal server和Canal client的HA。 Canal server:為了減少對MySQL dump的請求,不同server上的instance要求同一時間只能有一個處于running狀態(tài),其他的處于standby狀態(tài)。
Canal client:為了保證有序性,一份instance同一時間只能由一個Canal client進(jìn)行g(shù)et/ack/rollback操作,否則客戶端接收無法保證有序。 Canal Server HA架構(gòu)圖:
大致步驟:
- Canal server要啟動某個Canal instance時都先向Zookeeper進(jìn)行一次嘗試啟動判斷 (實(shí)現(xiàn):創(chuàng)建EPHEMERAL節(jié)點(diǎn),誰創(chuàng)建成功就允許誰啟動)
- 創(chuàng)建Zookeeper節(jié)點(diǎn)成功后,對應(yīng)的Canal server就啟動對應(yīng)的Canal instance,沒有創(chuàng)建成功的Canal instance就會處于standby狀態(tài)
- 一旦Zookeeper發(fā)現(xiàn)Canal server A創(chuàng)建的節(jié)點(diǎn)消失后,立即通知其他的Canal server再次進(jìn)行步驟1的操作,重新選出一個Canal server啟動instance
- Canal client每次進(jìn)行connect時,會首先向Zookeeper詢問當(dāng)前是誰啟動了Canal instance,然后和其建立鏈接,一旦鏈接不可用,會重新嘗試connect
Canal Client的方式和Canal server方式類似,也是利用Zookeeper的搶占EPHEMERAL節(jié)點(diǎn)的方式進(jìn)行控制。
到此這篇關(guān)于詳解監(jiān)聽MySQL的binlog日志工具分析:Canal的文章就介紹到這了,更多相關(guān)MySQL的binlog日志內(nèi)容請搜索服務(wù)器之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持服務(wù)器之家!
原文鏈接:https://www.cnblogs.com/bigdatalearnshare/p/13832709.html