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ǔ)的最后一次寫入位置
Get : 數(shù)據(jù)訂閱獲取的最后一次提取位置
Ack : 數(shù)據(jù)消費(fèi)成功的最后一次消費(fèi)位置

借鑒Disruptor的RingBuffer的實(shí)現(xiàn)，將RingBuffer拉直來(lái)看：

image.png

實(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ù)?；趃et獲取的batchId進(jìn)行提交，避免誤操作
void ack(long batchId)，顧命思議，確認(rèn)已經(jīng)消費(fèi)成功，通知server刪除數(shù)據(jù)?；趃et獲取的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:

image.png

大致步驟：

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)行控制.

色偷偷精品伊人,欧洲久久精品,欧美综合婷婷骚逼,国产AV主播,国产最新探花在线,九色在线视频一区,伊人大交九欧美,1769亚洲,黄色成人av

阿里開(kāi)源Canal--①簡(jiǎn)介

阿里開(kāi)源Canal--①簡(jiǎn)介

背景

項(xiàng)目介紹