史上最全的分布式數(shù)據(jù)同步中間間canal 之入門篇

什么是canal

canal是純Java開發(fā)斯碌。基于數(shù)據(jù)庫增量日志解析，提供增量數(shù)據(jù)訂閱&消費(fèi)汗唱，目前主要支持了MySQL

![史上最全的分布式數(shù)據(jù)同步中間間canal 之入門篇]

v2-da6d148ce9ad5297c4a8317ee8c492cf_hd_看圖王.jpg

如上圖：canal 模擬 MySQL slave 的交互協(xié)議，偽裝自己為 MySQL slave 丈攒，向 MySQL master 發(fā)送dump 協(xié)議

canal 搭建

搭建mysql環(huán)境

對于自建 MySQL , 需要先開啟 Binlog 寫入功能渡嚣，配置 binlog-format 為 ROW 模式，my.cnf 中配置如下

授權(quán) canal 鏈接 MySQL 賬號具有作為 MySQL slave 的權(quán)限, 如果已有賬戶可直接 grant

這個(gè)第一步還是蠻簡單的识椰，就是要自己搭建一個(gè)mysql绝葡，修改一下mysql的配置,這個(gè)配置一般是再/etc/my.cnf中，還是得要點(diǎn)小基礎(chǔ)的哈腹鹉，至少mysql得會搭

搭建canal環(huán)境

下載 canal, 訪問 release 頁面 , 選擇需要的包下載, 如以 1.0.17 版本為例

解壓縮

解壓完成后藏畅，進(jìn)入 /tmp/canal 目錄，可以看到如下結(jié)構(gòu)

v2-c1756d30c8acd6f7bcf12f0e76ab53da_hd_看圖王.jpg

配置修改

mysql serverId , v1.0.26+ will autoGen

canal.instance.mysql.slaveId=8

enable gtid use true/false

canal.instance.gtidon=false

position info 需要改成自己的數(shù)據(jù)庫信息

canal.instance.master.address=10.0.98.186:3306
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=

rds oss binlog

canal.instance.rds.accesskey=
canal.instance.rds.secretkey=
canal.instance.rds.instanceId=

table meta tsdb info

canal.instance.tsdb.enable=true

canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb

canal.instance.tsdb.dbUsername=canal

canal.instance.tsdb.dbPassword=canal

canal.instance.standby.address =

canal.instance.standby.journal.name =

canal.instance.standby.position =

canal.instance.standby.timestamp =

canal.instance.standby.gtid=

username/password 需要改成自己的數(shù)據(jù)庫信息

canal.instance.dbUsername=root
canal.instance.dbPassword=root
canal.instance.connectionCharset=UTF-8
canal.instance.defaultDatabaseName=expert-online-school

table regex

canal.instance.filter.regex=.\..

table black regex

canal.instance.filter.black.regex=
#################################################
</pre>

注意： canal.instance.connectionCharset 代表數(shù)據(jù)庫的編碼方式對應(yīng)到 java 中的編碼類型功咒，比如 UTF-8愉阎，GBK , ISO-8859-1 如果系統(tǒng)是1個(gè) cpu，需要將 canal.instance.parser.parallel 設(shè)置為 false

啟動

到目前為止 canal的服務(wù)端我們已經(jīng)搭建好了但是到目前我們只是把數(shù)據(jù)庫的binlog 拉到canal中力奋，我們還得把數(shù)據(jù)用otter去消費(fèi)

寫個(gè)簡單的Demo 去監(jiān)聽mysql 數(shù)據(jù)的變動

Jar包

測試代碼

import java.net.InetSocketAddress;
import java.util.List;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.protocol.Message;

public class CanalTest {

public static void main(String[] args) throws Exception {

CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("10.0.98.186", 11111), "expert", "root", "root");
connector.connect();
connector.subscribe(".\..");
connector.rollback();

while (true) {
Message message = connector.getWithoutAck(100); // 獲取指定數(shù)量的數(shù)據(jù)
long batchId = message.getId();
if (batchId == -1 || message.getEntries().isEmpty()) {
Thread.sleep(1000);
continue;
}
// System.out.println(message.getEntries());
printEntries(message.getEntries());
connector.ack(batchId);// 提交確認(rèn)榜旦，消費(fèi)成功，通知server刪除數(shù)據(jù)
// connector.rollback(batchId);// 處理失敗, 回滾數(shù)據(jù)景殷，后續(xù)重新獲取數(shù)據(jù)
}
}

private static void printEntries(List<Entry> entries) throws Exception {
for (Entry entry : entries) {
if (entry.getEntryType() != EntryType.ROWDATA) {
continue;
}

RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());

EventType eventType = rowChange.getEventType();
System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType));

for (RowData rowData : rowChange.getRowDatasList()) {
switch (rowChange.getEventType()) {
case INSERT:
System.out.println("INSERT ");
printColumns(rowData.getAfterColumnsList());
break;
case UPDATE:
System.out.println("UPDATE ");
printColumns(rowData.getAfterColumnsList());
break;
case DELETE:
System.out.println("DELETE ");
printColumns(rowData.getBeforeColumnsList());
break;

default:
break;
}
}
}
}

private static void printColumns(List<Column> columns) {
for(Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
}
}
}</pre>

測試結(jié)果

<pre style="-webkit-tap-highlight-color: transparent; box-sizing: border-box; font-family: Consolas, Menlo, Courier, monospace; font-size: 16px; white-space: pre-wrap; position: relative; line-height: 1.5; color: rgb(153, 153, 153); margin: 1em 0px; padding: 12px 10px; background: rgb(244, 245, 246); border: 1px solid rgb(232, 232, 232); font-style: normal; font-variant-ligatures: normal; font-variant-caps: normal; font-weight: 400; letter-spacing: normal; orphans: 2; text-align: start; text-indent: 0px; text-transform: none; widows: 2; word-spacing: 0px; -webkit-text-stroke-width: 0px; text-decoration-style: initial; text-decoration-color: initial;">================> binlog[mysql-bin.000017:240485980] , name[xxl_job,xxl_job_registry] , eventType : UPDATE
UPDATE
id : 402 update=false
registry_group : EXECUTOR update=false
registry_key : hq-eos-crawler update=false
registry_value : 172.27.0.1:15674 update=false
update_time : 2019-12-03 17:54:42 update=true
================> binlog[mysql-bin.000017:240486374] , name[xxl_job,xxl_job_registry] , eventType : UPDATE
UPDATE
id : 82 update=false
registry_group : EXECUTOR update=false
registry_key : hq-eos-inf-config update=false
registry_value : 172.18.0.1:15672 update=false
update_time : 2019-12-03 17:54:42 update=true
================> binlog[mysql-bin.000017:240486774] , name[xxl_job,xxl_job_registry] , eventType : UPDATE</pre>

注意一下

這里的配置來自于 canal.properties 我把這個(gè)配置也貼出來吧

flush data to zk

canal.zookeeper.flush.period = 1000
canal.withoutNetty = false

flush meta cursor/parse position to file

canal.file.data.dir = ${canal.conf.dir}
canal.file.flush.period = 1000

memory store RingBuffer size, should be Math.pow(2,n)

canal.instance.memory.buffer.size = 16384

memory store RingBuffer used memory unit size , default 1kb

canal.instance.memory.buffer.memunit = 1024

meory store gets mode used MEMSIZE or ITEMSIZE

canal.instance.memory.batch.mode = MEMSIZE

detecing config

canal.instance.detecting.enable = false

canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()

canal.instance.detecting.sql = select 1
canal.instance.detecting.interval.time = 3
canal.instance.detecting.retry.threshold = 3
canal.instance.detecting.heartbeatHaEnable = false

support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery

canal.instance.transaction.size = 1024

mysql fallback connected to new master should fallback times

canal.instance.fallbackIntervalInSeconds = 60

network config

canal.instance.network.receiveBufferSize = 16384
canal.instance.network.sendBufferSize = 16384
canal.instance.network.soTimeout = 30

binlog filter config

canal.instance.filter.druid.ddl = true
canal.instance.filter.query.dcl = false
canal.instance.filter.query.dml = false
canal.instance.filter.query.ddl = false
canal.instance.filter.table.error = false
canal.instance.filter.rows = false
canal.instance.filter.transaction.entry = false

binlog format/image check

canal.instance.binlog.format = ROW,STATEMENT,MIXED
canal.instance.binlog.image = FULL,MINIMAL,NOBLOB

binlog ddl isolation

canal.instance.get.ddl.isolation = false

parallel parser config

canal.instance.parser.parallel = true

concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()

canal.instance.parser.parallelThreadSize = 16

disruptor ringbuffer size, must be power of 2

canal.instance.parser.parallelBufferSize = 256

table meta tsdb info

canal.instance.tsdb.enable=true
canal.instance.tsdb.dir= ${canal.file.data.dir:../conf}/$ {canal.instance.destination:}
canal.instance.tsdb.url=jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
canal.instance.tsdb.dbUsername=root
canal.instance.tsdb.dbPassword=root

rds oss binlog account

canal.instance.rds.accesskey =
canal.instance.rds.secretkey =

#################################################
######### destinations #############
#################################################
canal.destinations= expert

conf root dir

canal.conf.dir = ../conf

auto scan instance dir add/remove and start/stop instance

canal.auto.scan = true
canal.auto.scan.interval = 5

canal.instance.tsdb.spring.xml=classpath:spring/tsdb/h2-tsdb.xml

canal.instance.tsdb.spring.xml=classpath:spring/tsdb/mysql-tsdb.xml

canal.instance.global.mode = spring
canal.instance.global.lazy = false

canal.instance.global.manager.address = 127.0.0.1:1099

canal.instance.global.spring.xml = classpath:spring/memory-instance.xml

canal.instance.global.spring.xml = classpath:spring/file-instance.xml

canal.instance.global.spring.xml = classpath:spring/default-instance.xml</pre>

這上面很多配置這邊就不一一說明了 zookeeper的話是做instance 集群的時(shí)候用的

結(jié)尾

canal入門寫完溅呢，謝謝大家的支持，大家如果想學(xué)后面一篇文章猿挚，一定要把這個(gè)照著做一遍咐旧。
因?yàn)椴┲饕彩且粋€(gè)開發(fā)萌新我也是一邊學(xué)一邊寫我有個(gè)目標(biāo)就是一周二到三篇希望能堅(jiān)持個(gè)一年吧希望各位大佬多提意見，讓我多學(xué)習(xí)绩蜻，一起進(jìn)步铣墨。

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者

人面猴
序言：七十年代末，一起剝皮案震驚了整個(gè)濱河市办绝，隨后出現(xiàn)的幾起案子伊约，更是在濱河造成了極大的恐慌，老刑警劉巖孕蝉，帶你破解...
沈念sama閱讀 216,372評論 6贊 498
死咒
序言：濱河連續(xù)發(fā)生了三起死亡事件屡律，死亡現(xiàn)場離奇詭異，居然都是意外死亡昔驱，警方通過查閱死者的電腦和手機(jī)疹尾，發(fā)現(xiàn)死者居然都...
沈念sama閱讀 92,368評論 3贊 392
救了他兩次的神仙讓他今天三更去死
文/潘曉璐我一進(jìn)店門，熙熙樓的掌柜王于貴愁眉苦臉地迎上來骤肛，“玉大人纳本，你說我怎么就攤上這事∫傅撸” “怎么了繁成？”我有些...
開封第一講書人閱讀 162,415評論 0贊 353
道士緝兇錄：失蹤的賣姜人
文/不壞的土叔我叫張陵，是天一觀的道長淑玫。經(jīng)常有香客問我巾腕，道長面睛，這世上最難降的妖魔是什么？我笑而不...
開封第一講書人閱讀 58,157評論 1贊 292
?港島之戀（遺憾婚禮）
正文為了忘掉前任尊搬，我火速辦了婚禮叁鉴，結(jié)果婚禮上，老公的妹妹穿的比我還像新娘佛寿。我一直安慰自己幌墓，他們只是感情好，可當(dāng)我...
茶點(diǎn)故事閱讀 67,171評論 6贊 388
惡毒庶女頂嫁案：這布局不是一般人想出來的
文/花漫我一把揭開白布冀泻。她就那樣靜靜地躺著常侣，像睡著了一般。火紅的嫁衣襯著肌膚如雪弹渔。梳的紋絲不亂的頭發(fā)上胳施，一...
開封第一講書人閱讀 51,125評論 1贊 297
城市分裂傳說
那天，我揣著相機(jī)與錄音肢专，去河邊找鬼舞肆。笑死，一個(gè)胖子當(dāng)著我的面吹牛鸟召，可吹牛的內(nèi)容都是我干的胆绊。我是一名探鬼主播氨鹏，決...
沈念sama閱讀 40,028評論 3贊 417
雙鴛鴦連環(huán)套：你想象不到人心有多黑
文/蒼蘭香墨我猛地睜開眼欧募，長吁一口氣：“原來是場噩夢啊……” “哼！你這毒婦竟也來了仆抵？” 一聲冷哼從身側(cè)響起跟继，我...
開封第一講書人閱讀 38,887評論 0贊 274
萬榮殺人案實(shí)錄
序言：老撾萬榮一對情侶失蹤，失蹤者是張志新（化名）和其女友劉穎镣丑，沒想到半個(gè)月后舔糖，有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體，經(jīng)...
沈念sama閱讀 45,310評論 1贊 310
?護(hù)林員之死
正文獨(dú)居荒郊野嶺守林人離奇死亡莺匠，尸身上長有42處帶血的膿包…… 初始之章·張勛以下內(nèi)容為張勛視角年9月15日...
茶點(diǎn)故事閱讀 37,533評論 2贊 332
?白月光啟示錄
正文我和宋清朗相戀三年金吗，在試婚紗的時(shí)候發(fā)現(xiàn)自己被綠了。大學(xué)時(shí)的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片趣竣。...
茶點(diǎn)故事閱讀 39,690評論 1贊 348
活死人
序言：一個(gè)原本活蹦亂跳的男人離奇死亡摇庙，死狀恐怖，靈堂內(nèi)的尸體忽然破棺而出遥缕，到底是詐尸還是另有隱情卫袒，我是刑警寧澤，帶...
沈念sama閱讀 35,411評論 5贊 343
?日本核電站爆炸內(nèi)幕
正文年R本政府宣布单匣，位于F島的核電站夕凝，受9級特大地震影響宝穗，放射性物質(zhì)發(fā)生泄漏。R本人自食惡果不足惜码秉，卻給世界環(huán)境...
茶點(diǎn)故事閱讀 41,004評論 3贊 325
男人毒藥：我在死后第九天來索命
文/蒙蒙一慈俯、第九天我趴在偏房一處隱蔽的房頂上張望兔朦。院中可真熱鬧，春花似錦、人聲如沸雇盖。這莊子的主人今日做“春日...
開封第一講書人閱讀 31,659評論 0贊 22
一樁弒父案滋恬，背后竟有這般陰謀
文/蒼蘭香墨我抬頭看了看天上的太陽浅乔。三九已至，卻和暖如春礁竞，著一層夾襖步出監(jiān)牢的瞬間糖荒，已是汗流浹背。一陣腳步聲響...
開封第一講書人閱讀 32,812評論 1贊 268
情欲美人皮
我被黑心中介騙來泰國打工模捂，沒想到剛下飛機(jī)就差點(diǎn)兒被人妖公主榨干…… 1. 我叫王不留捶朵，地道東北人。一個(gè)月前我還...
沈念sama閱讀 47,693評論 2贊 368
代替公主和親
正文我出身青樓狂男，卻偏偏與公主長得像综看，于是被迫代替她去往敵國和親。傳聞我的和親對象是個(gè)殘疾皇子岖食，可洞房花燭夜當(dāng)晚...
茶點(diǎn)故事閱讀 44,577評論 2贊 353

史上最全的分布式數(shù)據(jù)同步中間間canal 之入門篇

什么是canal

canal 搭建

搭建mysql環(huán)境

搭建canal環(huán)境

mysql serverId , v1.0.26+ will autoGen

enable gtid use true/false

position info 需要改成自己的數(shù)據(jù)庫信息

rds oss binlog

table meta tsdb info

canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb

canal.instance.tsdb.dbUsername=canal

canal.instance.tsdb.dbPassword=canal

canal.instance.standby.address =

canal.instance.standby.journal.name =

canal.instance.standby.position =

canal.instance.standby.timestamp =

canal.instance.standby.gtid=

username/password 需要改成自己的數(shù)據(jù)庫信息

table regex

table black regex

寫個(gè)簡單的Demo 去監(jiān)聽mysql 數(shù)據(jù)的變動

Jar包

測試代碼

flush data to zk

flush meta cursor/parse position to file

memory store RingBuffer size, should be Math.pow(2,n)

memory store RingBuffer used memory unit size , default 1kb

meory store gets mode used MEMSIZE or ITEMSIZE

detecing config

canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()

support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery

mysql fallback connected to new master should fallback times

network config

binlog filter config

binlog format/image check

binlog ddl isolation

parallel parser config

concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()

canal.instance.parser.parallelThreadSize = 16

disruptor ringbuffer size, must be power of 2

table meta tsdb info

rds oss binlog account

conf root dir

auto scan instance dir add/remove and start/stop instance

canal.instance.tsdb.spring.xml=classpath:spring/tsdb/mysql-tsdb.xml

canal.instance.global.manager.address = 127.0.0.1:1099

canal.instance.global.spring.xml = classpath:spring/memory-instance.xml

canal.instance.global.spring.xml = classpath:spring/default-instance.xml</pre>

結(jié)尾

推薦閱讀更多精彩內(nèi)容