戰(zhàn)：基于MySQL binlog的增量數(shù)據(jù)同步全解析)
1. 先搞清楚canal到底在解決什么問(wèn)題如果說(shuō)這兩年數(shù)據(jù)同步領(lǐng)域哪個(gè)組件最值得花一個(gè)星期認(rèn)真學(xué)一遍我的答案一定是canal。這名字你可能在簡(jiǎn)歷上見(jiàn)過(guò)在招聘JD里見(jiàn)過(guò)甚至在同事的代碼評(píng)審里被吐槽過(guò)但真要問(wèn)你它到底是干嘛的很多人的回答其實(shí)是模糊的嗯就是同步MySQL數(shù)據(jù)的——差不多對(duì)但不夠準(zhǔn)確。我換個(gè)說(shuō)法你就明白了canal做的是MySQL增量數(shù)據(jù)訂閱也就是監(jiān)聽(tīng)MySQL的binlog日志每當(dāng)數(shù)據(jù)庫(kù)里發(fā)生insert、update、delete操作canal就能像裝了竊聽(tīng)器一樣在毫秒級(jí)把這個(gè)變更事件推給你。注意關(guān)鍵詞是增量——它不是全量復(fù)制不是每天導(dǎo)一次數(shù)據(jù)而是從你部署那一刻開(kāi)始實(shí)時(shí)關(guān)注每一條新變化。那它解決的到底是什么問(wèn)題我舉個(gè)最常見(jiàn)的場(chǎng)景。你們系統(tǒng)里肯定有一張熱點(diǎn)表比如商品信息、用戶資料、配置數(shù)據(jù)業(yè)務(wù)方讀得很頻繁。為了扛住流量你引入了Redis做緩存查詢走緩存寫(xiě)操作更新數(shù)據(jù)庫(kù)。聽(tīng)起來(lái)沒(méi)問(wèn)題但更新數(shù)據(jù)庫(kù)的同時(shí)同步緩存這一步寫(xiě)法稍微不嚴(yán)謹(jǐn)就會(huì)出現(xiàn)緩存和數(shù)據(jù)庫(kù)不一致的臟數(shù)據(jù)。常見(jiàn)的做法有幾種雙寫(xiě)、定時(shí)任務(wù)全量刷新、消息隊(duì)列異步通知。雙寫(xiě)邏輯侵入業(yè)務(wù)代碼定時(shí)任務(wù)有延遲且浪費(fèi)資源消息隊(duì)列異步通知倒是好但你需要手動(dòng)把每一次數(shù)據(jù)變更封裝成消息發(fā)出去代碼量不小。canal的思路是我不碰你的業(yè)務(wù)代碼我在數(shù)據(jù)庫(kù)這一側(cè)偷聽(tīng)binlog把變更事件統(tǒng)一抓出來(lái)再推給你的下游。無(wú)論你是想去同步緩存、同步到ES、同步到數(shù)據(jù)倉(cāng)庫(kù)還是同步到另一個(gè)MySQL都不需要改業(yè)務(wù)代碼。這個(gè)解耦的價(jià)值用過(guò)的人都懂——只需要一次配置全公司的數(shù)據(jù)消費(fèi)方都能受益。對(duì)小白來(lái)說(shuō)canal最友好的一點(diǎn)是它自帶一個(gè)客戶端訂閱模型你只要能寫(xiě)Java代碼幾行就能跑通一個(gè)Demo完全不需要一開(kāi)始就理解它的存儲(chǔ)原理。但如果你想用好它光是會(huì)跑肯定不夠因?yàn)檎嬲目尤诩?xì)節(jié)里binlog格式?jīng)]配對(duì)、位點(diǎn)丟失導(dǎo)致重復(fù)消費(fèi)、和多個(gè)訂閱端搶數(shù)據(jù)……這些問(wèn)題我用后面幾個(gè)章節(jié)慢慢展開(kāi)。順便說(shuō)一句canal項(xiàng)目的名稱(chēng)來(lái)自canal這個(gè)英文單詞中文意思是運(yùn)河、水道寓意很直白數(shù)據(jù)從MySQL流向下游的通道。這個(gè)比喻特別貼切你越往后用越會(huì)覺(jué)得它真的就像一條管道把數(shù)據(jù)庫(kù)的血液輸送到各個(gè)需要的地方。2. 安裝前的三個(gè)關(guān)鍵開(kāi)關(guān)JDK、binlog和權(quán)限安裝canal本身不難難的是準(zhǔn)備工作。很多小白按照網(wǎng)上的教程一步步來(lái)結(jié)果客戶端連不上、收不到消息回頭一看全是配套環(huán)境沒(méi)弄對(duì)。所以這一章我把安裝前置條件拆成三個(gè)開(kāi)關(guān)每個(gè)都要提前確認(rèn)壞了哪一個(gè)都會(huì)讓你后面抓狂。2.1 第一個(gè)開(kāi)關(guān)JDK版本canal的server端是Java寫(xiě)的所以你機(jī)器上得有JDK。這里有個(gè)坑需要你避一下canal 1.1.x系列用JDK 8沒(méi)問(wèn)題但如果你用的是比較新的canal 1.1.7版本或者想跑源碼調(diào)試建議直接用JDK 11。我遇到過(guò)用JDK 8跑新版本canal啟動(dòng)的時(shí)候報(bào)了一些奇怪的并發(fā)類(lèi)加載錯(cuò)誤看起來(lái)像是代碼問(wèn)題其實(shí)是編譯目標(biāo)版本太高導(dǎo)致的。裝好之后確認(rèn)一下版本java -version正常情況下你會(huì)看到類(lèi)似openjdk version 1.8.0_***或openjdk version 11.0.***的輸出。如果你的機(jī)器上存在多個(gè)JDK建議用export JAVA_HOME/path/to/jdk指定清楚避免canal的啟動(dòng)腳本拿到錯(cuò)誤的版本。2.2 第二個(gè)開(kāi)關(guān)MySQL必須開(kāi)啟binlogcanal的工作原理是偽裝成MySQL的從節(jié)點(diǎn)向主庫(kù)請(qǐng)求binlog日志。所以你的MySQL必須開(kāi)啟binlog這是整個(gè)鏈條的源頭。修改MySQL配置文件Linux下一般是/etc/my.cnf或/etc/mysql/mysql.conf.d/mysqld.cnf[mysqld] log-binmysql-bin binlog-formatROW server-id1這三行缺一不可尤其第二行binlog-formatROW。我見(jiàn)過(guò)有人在測(cè)試環(huán)境用了默認(rèn)的STATEMENT格式結(jié)果canal解析出來(lái)一堆SQL語(yǔ)句根本拿不到精確的行級(jí)變更數(shù)據(jù)。因?yàn)閏anal的核心輸出是某行某個(gè)字段從舊值變成了新值而這種結(jié)構(gòu)化信息只有ROW格式才完整保留。改完之后記得重啟MySQLservice mysqld restart或者systemctl restart mysqld。然后進(jìn)入MySQL命令行驗(yàn)證SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE binlog_format;兩個(gè)結(jié)果都符合預(yù)期這個(gè)開(kāi)關(guān)才算真正打開(kāi)。還有一個(gè)細(xì)節(jié)容易被忽略server-id不能和canal實(shí)例配置的server-id重復(fù)。我用一個(gè)真實(shí)的踩坑經(jīng)歷說(shuō)明第一次部署的時(shí)候我隨手給canal設(shè)置了server-id1和MySQL自己的server-id沖突結(jié)果canal連接主庫(kù)時(shí)直接被拒日志里報(bào)ERROR: Got error 1236。這個(gè)問(wèn)題排查了很久才定位到因?yàn)閳?bào)錯(cuò)信息很不直觀。所以提前規(guī)劃好MySQL給一個(gè)server-idcanal給另一個(gè)比如101、102這種互不干擾。2.3 第三個(gè)開(kāi)關(guān)MySQL授權(quán)賬號(hào)canal需要用一個(gè)專(zhuān)門(mén)的MySQL賬號(hào)連接主庫(kù)這個(gè)賬號(hào)要有REPLICATION SLAVE和REPLICATION CLIENT權(quán)限。很多人在這步直接用root賬號(hào)倒也能跑通但從規(guī)范角度不推薦尤其到了生產(chǎn)環(huán)境最少權(quán)限原則是底線。創(chuàng)建賬號(hào)的SQL如下CREATE USER canal% IDENTIFIED BY canal123; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; FLUSH PRIVILEGES;這里給%是允許任意主機(jī)連接實(shí)際生產(chǎn)建議限定為運(yùn)行canal的那臺(tái)機(jī)器IP安全第一。另外注意如果數(shù)據(jù)庫(kù)里已有的表結(jié)構(gòu)比較復(fù)雜canal在解析DDL事件時(shí)可能還需要訪問(wèn)表定義所以順手把SELECT權(quán)限也加上。這三件事做完環(huán)境準(zhǔn)備基本就緒。老實(shí)說(shuō)這個(gè)過(guò)程十分鐘都用不了但價(jià)值遠(yuǎn)超十分鐘——它決定了你后面所有排查動(dòng)作能不能有一個(gè)干凈的基礎(chǔ)。3. 一條數(shù)據(jù)從MySQL到客戶端的完整路徑很多教程上來(lái)就讓你裝canal、跑Demo等你跑通了還是一頭霧水它到底是怎么拿到數(shù)據(jù)的這一章我從原理層面完整講一遍大概需要你花五分鐘耐心讀讀完之后你就擁有了排查問(wèn)題的底層地圖遇到任何詭異現(xiàn)象都能知道往哪個(gè)環(huán)節(jié)去查。3.1 先理解MySQL主從復(fù)制是怎么回事canal的核心機(jī)制其實(shí)是在模仿MySQL的從庫(kù)。MySQL的主從復(fù)制過(guò)程很簡(jiǎn)單主庫(kù)把每一次變更寫(xiě)入binlog從庫(kù)連接到主庫(kù)請(qǐng)求從某個(gè)binlog位點(diǎn)開(kāi)始的數(shù)據(jù)拿到之后在本地重放。canal做的事情和從庫(kù)一模一樣只不過(guò)它拿到這些日志之后重放的不是SQL而是把日志解析成結(jié)構(gòu)化的數(shù)據(jù)變更事件交給下游程序處理。這個(gè)過(guò)程涉及binlog里的幾個(gè)概念我用大白話解釋一下binlog文件MySQL落盤(pán)生成的日志文件比如mysql-bin.000001達(dá)到一定大小后自動(dòng)切換下一個(gè)文件。binlog位點(diǎn)position一條日志在文件中的偏移量它是唯一的?？蛻舳嘶謴?fù)消費(fèi)時(shí)只要記住上次消費(fèi)到的position就能從斷點(diǎn)繼續(xù)。GTID全局事務(wù)ID是比position更高級(jí)的定位方式不依賴(lài)具體文件名和偏移量。新版本canal推薦用GTID模式主從切換時(shí)能自動(dòng)適應(yīng)。3.2 canal內(nèi)部的三段式流水線canal從拿到binlog到吐出數(shù)據(jù)內(nèi)部經(jīng)歷了三個(gè)核心階段第一階段抓取Parser。canal偽裝成從庫(kù)向MySQL發(fā)送dump協(xié)議請(qǐng)求。MySQL就會(huì)從指定的binlog位點(diǎn)開(kāi)始把日志一條條推送給canal。這個(gè)階段最純粹只負(fù)責(zé)拿。第二階段解析Sink。拿到的binlog是事件流需要解析。比如一個(gè)UPDATE事件會(huì)解析出舊值before和新值after這個(gè)是canal最有價(jià)值的輸出——下游可以直接拿到{id: 1, name: 舊值}變成{id: 1, name: 新值}這樣的JSON數(shù)據(jù)而不需要自己拼SQL去反查。第三階段存儲(chǔ)Store。解析完成后事件會(huì)寫(xiě)入一個(gè)內(nèi)存環(huán)形隊(duì)列等待客戶端來(lái)拉取。canal內(nèi)置了一個(gè)基于內(nèi)存的存儲(chǔ)組件默認(rèn)容量是16384個(gè)事件客戶端消費(fèi)速度跟不上時(shí)事件會(huì)在內(nèi)存里排隊(duì)排滿了之后才會(huì)出現(xiàn)丟數(shù)據(jù)或者阻塞的情況。整個(gè)鏈條如下圖所示文字版MySQL binlog → canal Server 抓取 → 解析 → 內(nèi)存隊(duì)列 → 客戶端(Client) 拉取 → 你的業(yè)務(wù)代碼/ES/Redis...3.3 你眼里的canal組件Server、Instance和Client實(shí)際部署時(shí)你會(huì)接觸到幾個(gè)概念這里一次性講清楚避免后續(xù)被配置項(xiàng)繞暈canal Server一個(gè)獨(dú)立運(yùn)行的Java進(jìn)程核心服務(wù)。canal InstanceServer內(nèi)部的一個(gè)邏輯實(shí)例每個(gè)Instance對(duì)應(yīng)一個(gè)MySQL數(shù)據(jù)源。一個(gè)Server可以配置多個(gè)Instance每個(gè)Instance有獨(dú)立的配置、獨(dú)立的訂閱位點(diǎn)、獨(dú)立的隊(duì)列。canal Client你寫(xiě)的消費(fèi)端程序通過(guò)TCP協(xié)議從Server拉取數(shù)據(jù)。打個(gè)比方Server是餐廳Instance是廚房里的一個(gè)灶臺(tái)——每個(gè)灶臺(tái)對(duì)應(yīng)一個(gè)供應(yīng)商MySQLClient是來(lái)吃飯的顧客。一個(gè)餐廳有多個(gè)灶臺(tái)很正常每個(gè)灶臺(tái)負(fù)責(zé)炒不同的菜互不干擾。從部署角度小白剛開(kāi)始只需要一個(gè)Server、一個(gè)Instance、一個(gè)Client后面需要訂閱多個(gè)庫(kù)或者對(duì)接多套下游時(shí)再考慮多Instance。4. 跑通第一個(gè)Demo用Java代碼訂閱一條insert前面講了那么多理論和準(zhǔn)備這一章我們真正動(dòng)手。我用一臺(tái)Linux機(jī)器做演示canal部署包從官方GitHub Release頁(yè)面下載最新穩(wěn)定版即可解壓后目錄結(jié)構(gòu)大致如下/canal ├── bin ├── conf │ ├── canal.properties │ ├── example │ │ └── instance.propertiesconf/canal.properties是全局配置conf/example是默認(rèn)實(shí)例的配置目錄instance.properties是實(shí)例級(jí)配置。4.1 修改實(shí)例配置先改instance.properties這是核心# MySQL主庫(kù)地址 canal.instance.master.address127.0.0.1:3306 # 連接MySQL的賬號(hào)密碼 canal.instance.dbUsernamecanal canal.instance.dbPasswordcanal123 # 訂閱的數(shù)據(jù)庫(kù)和表*表示所有 canal.instance.filter.regextest\\..* # 位點(diǎn)定位方式 canal.instance.gtidfalse重點(diǎn)說(shuō)明一下filter.regex這個(gè)配置。它的格式是數(shù)據(jù)庫(kù)名\\..*比如我只想訂閱test庫(kù)的所有表就寫(xiě)test\\..*想訂閱test庫(kù)下的user表就寫(xiě)test\\.user想訂閱多個(gè)庫(kù)用|分隔。這個(gè)正則很容易寫(xiě)錯(cuò)因?yàn)镴ava字符串本身也要轉(zhuǎn)義所以\\.寫(xiě)成\\.而不是\.。我見(jiàn)過(guò)有人照抄配置把轉(zhuǎn)義符弄丟了結(jié)果canal直接不輸出任何數(shù)據(jù)查了半天才發(fā)現(xiàn)是正則匹配問(wèn)題。4.2 修改全局配置再看canal.properties重點(diǎn)確認(rèn)以下幾項(xiàng)canal.port11111 canal.destinationsexample canal.instance.global.modetcpcanal.port是客戶端連接的TCP端口默認(rèn)11111不用改。canal.destinations指定啟動(dòng)哪些實(shí)例默認(rèn)example對(duì)應(yīng)conf目錄下的example文件夾。canal.instance.global.modetcp表示消費(fèi)端通過(guò)TCP直連方式拉取數(shù)據(jù)——這是最簡(jiǎn)單的方式后面聊MQ模式時(shí)再說(shuō)。4.3 啟動(dòng)Servercd /canal/bin ./startup.sh啟動(dòng)后看日志確認(rèn)沒(méi)有異常tail -f /canal/logs/canal/canal.log如果看到canal server is running now之類(lèi)的字樣并且沒(méi)有報(bào)錯(cuò)說(shuō)明Server起來(lái)了。再檢查實(shí)例狀態(tài)tail -f /canal/logs/example/example.log正常情況下能看到start successful。這里有個(gè)新手最容易遇到的情況日志里顯示canal is starting然后卡住不動(dòng)多半是MySQL連接沒(méi)通去查example.log里有沒(méi)有connect timeout或者Access denied字樣。4.4 寫(xiě)一個(gè)最小客戶端canal官方提供了一個(gè)客戶端庫(kù)canal.client用Maven引入dependency groupIdcom.alibaba.otter/groupId artifactIdcanal.client/artifactId version1.1.7/version /dependency然后寫(xiě)一個(gè)最簡(jiǎn)單的Java類(lèi)核心邏輯是建立連接 → 拉取消息 → 解析事件 → 打印變更內(nèi)容。import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.CanalEntry.EntryType; import java.net.InetSocketAddress; import java.util.List; public class CanalDemo { public static void main(String[] args) { // 連接canal server CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(127.0.0.1, 11111), example, // instance名稱(chēng) , // 無(wú)認(rèn)證時(shí)填空 ); connector.connect(); // 訂閱所有表這里要和instance的filter.regex對(duì)應(yīng) connector.subscribe(test\\..*); while (true) { // 一次拉取100條不設(shè)置超時(shí) Message message connector.getWithoutAck(100); long batchId message.getId(); if (batchId -1 || message.getEntries().isEmpty()) { // 沒(méi)有數(shù)據(jù)睡1秒再試 try { Thread.sleep(1000); } catch (InterruptedException e) {} continue; } for (CanalEntry.Entry entry : message.getEntries()) { if (entry.getEntryType() ! EntryType.ROWDATA) { continue; } CanalEntry.RowChange rowChange CanalEntry.RowChange.parseFrom(entry.getStoreValue()); CanalEntry.EventType eventType rowChange.getEventType(); for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { if (eventType CanalEntry.EventType.INSERT) { System.out.println(INSERT: rowData.getAfterColumnsList()); } else if (eventType CanalEntry.EventType.UPDATE) { System.out.println(UPDATE before: rowData.getBeforeColumnsList()); System.out.println(UPDATE after: rowData.getAfterColumnsList()); } else if (eventType CanalEntry.EventType.DELETE) { System.out.println(DELETE: rowData.getBeforeColumnsList()); } } } // 確認(rèn)消費(fèi)成功 connector.ack(batchId); } } }這段代碼看著有點(diǎn)長(zhǎng)但核心就三件事連接、拉取、確認(rèn)。所有canal客戶端的骨架都是這個(gè)模式你以后無(wú)論用Spring Boot封裝還是寫(xiě)其他語(yǔ)言的客戶端都逃不開(kāi)這個(gè)套路。4.5 實(shí)測(cè)驗(yàn)證運(yùn)行上面的Java類(lèi)然后在MySQL里執(zhí)行INSERT INTO test.user (name, age) VALUES (張三, 25);立刻就能在控制臺(tái)看到類(lèi)似下面的輸出INSERT: [name: 張三, age: 25]看到這一行你的canal入門(mén)第一個(gè)Demo就算真正跑通了。從這一刻起任何數(shù)據(jù)庫(kù)變更都會(huì)實(shí)時(shí)出現(xiàn)在你的消費(fèi)端面前剩下的就是你怎么利用這些數(shù)據(jù)了。5. 小白最容易踩的五個(gè)坑含完整排查思路說(shuō)實(shí)話canal的部署和Demo真的很簡(jiǎn)單但越是這種看起來(lái)簡(jiǎn)單的組件越容易在細(xì)節(jié)上翻車(chē)。這一章我把自己和周?chē)瞬冗^(guò)的坑整理出來(lái)每一項(xiàng)都附上排查鏈路而不是直接給結(jié)論——因?yàn)橄麓文阌龅降目赡苤皇莻€(gè)相似但不完全一樣的問(wèn)題思路比答案更重要。5.1 坑一binlog格式不是ROW解析結(jié)果全是SQL現(xiàn)象客戶端能連上能收到消息但解析出來(lái)的RowData是空的只有SQL語(yǔ)句。排查鏈路第一步回到MySQL執(zhí)行SHOW VARIABLES LIKE binlog_format;如果結(jié)果是STATEMENT或MIXED基本就是問(wèn)題所在。第二步修改配置為ROW重啟MySQL。第三步重新執(zhí)行一條INSERT看canal是否輸出結(jié)構(gòu)化的列值。根因只有ROW格式才會(huì)記錄每一行變更的具體列值STATEMENT格式記錄的是原始SQL文本canal無(wú)法從中還原出行級(jí)別數(shù)據(jù)。5.2 坑二server-id沖突導(dǎo)致連接被拒現(xiàn)象canal日志里出現(xiàn)ERROR: Got error 1236或者直接連接不上MySQL。排查鏈路第一步檢查MySQL的server-idSELECT server_id;第二步檢查canal的canal.instance.properties里有沒(méi)有canal.instance.mysql.slaveId配置項(xiàng)這個(gè)值默認(rèn)是123456。如果和MySQL的server-id一樣就改成一個(gè)不同的值比如654321。第三步重啟canal實(shí)例觀察日志。根因MySQL主從復(fù)制機(jī)制要求每個(gè)從節(jié)點(diǎn)的server-id必須全局唯一。canal偽裝成一個(gè)從節(jié)點(diǎn)如果它的server-id和MySQL主庫(kù)相同主庫(kù)會(huì)認(rèn)為這是我自己直接拒絕復(fù)制請(qǐng)求。這里插一句生產(chǎn)中很多人會(huì)忽略這個(gè)參數(shù)因?yàn)樗哪J(rèn)值看起來(lái)沒(méi)存在感。建議部署清單里專(zhuān)門(mén)加一行確認(rèn)slaveId與所有MySQL節(jié)點(diǎn)均不重復(fù)。5.3 坑三位點(diǎn)丟失導(dǎo)致重復(fù)消費(fèi)或漏消費(fèi)現(xiàn)象canal重啟后客戶端收到了之前已經(jīng)處理過(guò)的舊數(shù)據(jù)或者直接跳過(guò)了部分?jǐn)?shù)據(jù)。排查鏈路第一步確認(rèn)canal的meta信息存儲(chǔ)在哪里。默認(rèn)情況下位點(diǎn)存在內(nèi)存中Server重啟就會(huì)丟失從當(dāng)前最新位置開(kāi)始消費(fèi)。如果你需要持久化要在canal.properties里配置canal.instance.global.spring.xmlclasspath:spring/memory-instance.xml改為file-instance.xml可以把meta寫(xiě)到本地文件。第二步檢查客戶端代碼是否正確ack。只get不ack重啟后會(huì)從上次ack的位置重新拉取產(chǎn)生重復(fù)消費(fèi)ack太快也可能出問(wèn)題如果先ack再處理業(yè)務(wù)邏輯處理失敗時(shí)數(shù)據(jù)就再也拉不回來(lái)了這是至少一次語(yǔ)義下最常見(jiàn)的取舍。第三步設(shè)計(jì)消費(fèi)冪等。無(wú)論位點(diǎn)多可靠分布式環(huán)境下重復(fù)消費(fèi)幾乎不可避免所以消費(fèi)端的冪等才是最終兜底。根因canal的位點(diǎn)機(jī)制是經(jīng)典的有且至少一次消費(fèi)語(yǔ)義重復(fù)消費(fèi)是常態(tài)漏消費(fèi)是因?yàn)槲稽c(diǎn)推進(jìn)到了業(yè)務(wù)成功之前。5.4 坑四多個(gè)客戶端實(shí)例同時(shí)消費(fèi)同一個(gè)instance現(xiàn)象多個(gè)服務(wù)實(shí)例都連了同一個(gè)canal destination結(jié)果每條數(shù)據(jù)只被其中一個(gè)實(shí)例收到而不是每個(gè)實(shí)例都收到。排查鏈路第一步確認(rèn)你的業(yè)務(wù)訴求是廣播還是負(fù)載均衡。canal不保證多個(gè)client同時(shí)消費(fèi)時(shí)每個(gè)client都能拿到全量數(shù)據(jù)默認(rèn)是競(jìng)爭(zhēng)消費(fèi)——一條數(shù)據(jù)只會(huì)給一個(gè)client。第二步如果確實(shí)需要多個(gè)下游分別消費(fèi)同一份數(shù)據(jù)可以有三種方案一是每個(gè)下游對(duì)應(yīng)一個(gè)獨(dú)立的instance二是通過(guò)MQ模式發(fā)布訂閱三是使用canal的CanalMQStarter配多個(gè)不同的group。根因canal的tcp模式的定位是一個(gè)destination對(duì)應(yīng)一個(gè)消費(fèi)組多個(gè)client是在消費(fèi)組內(nèi)競(jìng)爭(zhēng)拉取不是復(fù)制分發(fā)。5.5 坑五DDL事件和事務(wù)邊界導(dǎo)致的數(shù)據(jù)異常現(xiàn)象執(zhí)行ALTER TABLE之后消費(fèi)端收到的數(shù)據(jù)突然解析失敗或者一個(gè)事務(wù)內(nèi)的多變更變得不完整。排查鏈路第一步檢查消費(fèi)端是否處理了EntryType.ROWDATA之外的事件。DDL事件屬于ENTRY_TYPE的其他類(lèi)型如果你在消費(fèi)端直接跳過(guò)所有非ROWDATA事件DDL對(duì)下游可能造成結(jié)構(gòu)不一致。第二步開(kāi)啟canal的并行消費(fèi)時(shí)要注意事務(wù)一致性。如果下游是數(shù)據(jù)庫(kù)或緩存建議按事務(wù)邊界整體提交避免同一個(gè)事務(wù)的變更被拆到不同地方執(zhí)行。根因canal按事務(wù)為單位讀取binlog但消費(fèi)端如果并行處理事務(wù)邊界會(huì)被人為拆散。這五個(gè)坑基本覆蓋了我見(jiàn)過(guò)的90%小白問(wèn)題。說(shuō)實(shí)話每個(gè)坑背后都對(duì)應(yīng)一個(gè)理解盲區(qū)不是性能問(wèn)題不是bug而是對(duì)canal是怎么工作的理解不到位。所以每次出問(wèn)題先回到第3章的原理圖里定位是在哪個(gè)環(huán)節(jié)排查速度會(huì)快很多。6. 進(jìn)階思路從單機(jī)Demo走向生產(chǎn)環(huán)境Demo跑通、坑也避開(kāi)之后你可能會(huì)問(wèn)那真實(shí)項(xiàng)目里一般怎么用這一章我講三個(gè)最常用的落地方向以及小白從單機(jī)走向生產(chǎn)時(shí)需要補(bǔ)的功課。6.1 實(shí)戰(zhàn)方向一數(shù)據(jù)庫(kù)變更同步到Redis這是canal最常見(jiàn)的用法。假設(shè)你的商品表product里有價(jià)格字段價(jià)格變了緩存也得變。傳統(tǒng)做法是改業(yè)務(wù)代碼在更新數(shù)據(jù)庫(kù)后刪除緩存用canal之后消費(fèi)端收到UPDATE事件直接從JSON里提取最新值執(zhí)行SET key value或者刪除緩存。業(yè)務(wù)代碼零侵入。要注意兩點(diǎn)第一canal發(fā)來(lái)的是列名值的結(jié)構(gòu)不是SQL所以你需要自己把變更映射成業(yè)務(wù)對(duì)象這個(gè)映射邏輯一般寫(xiě)在消費(fèi)端一個(gè)獨(dú)立模塊里第二如果變更頻繁要注意給緩存加防抖比如1秒內(nèi)同一行的多次變更合并成一次刷緩存避免抖動(dòng)。6.2 實(shí)戰(zhàn)方向二異構(gòu)數(shù)據(jù)同步到Elasticsearch把MySQL的數(shù)據(jù)同步到ES做全文檢索也是高頻場(chǎng)景。流程大致是canal監(jiān)聽(tīng)MySQL變更 → 消費(fèi)端把變更拼成ES文檔 → 調(diào)用ES的index/update接口。這里有個(gè)常見(jiàn)的設(shè)計(jì)問(wèn)題ES里可能需要寬表比如訂單文檔里要包含用戶名、商品名而這些字段來(lái)自其他表。如果只訂閱訂單表拿不到關(guān)聯(lián)信息。解決思路有兩個(gè)一是訂閱多張表在消費(fèi)端做內(nèi)存關(guān)聯(lián)二是讓業(yè)務(wù)方寫(xiě)出包含所有所需字段的預(yù)聚合表canal只訂閱這一張。第二個(gè)方案更簡(jiǎn)單可靠推薦小白先這么做。6.3 從單機(jī)到集群三個(gè)生產(chǎn)級(jí)配置生產(chǎn)環(huán)境不可能只跑一個(gè)單機(jī)canal但我也要提醒你一開(kāi)始別急著上集群先把下面三件事做好比盲目加機(jī)器更有價(jià)值。第一個(gè)是位點(diǎn)持久化。前面說(shuō)過(guò)默認(rèn)位點(diǎn)在內(nèi)存在Server重啟就丟。生產(chǎn)環(huán)境務(wù)必配置成file模式或zookeeper模式。我用的是file模式最簡(jiǎn)單但要注意磁盤(pán)空間和文件權(quán)限。第二個(gè)是訂閱粒度收斂。開(kāi)發(fā)環(huán)境訂閱所有庫(kù)所有表沒(méi)問(wèn)題生產(chǎn)環(huán)境一定要用filter.regex收窄到具體業(yè)務(wù)庫(kù)和表不然會(huì)白白消耗大量性能還可能把其他團(tuán)隊(duì)的數(shù)據(jù)變更也帶進(jìn)你的消費(fèi)邏輯。第三個(gè)是監(jiān)控告警。canal是實(shí)時(shí)鏈路一旦它停了下游全部數(shù)據(jù)滯后。至少要做到進(jìn)程存活監(jiān)控、消費(fèi)延遲監(jiān)控canal有delayed指標(biāo)、客戶端積壓消息數(shù)監(jiān)控。這些指標(biāo)都可以通過(guò)JMX或者集成到現(xiàn)有監(jiān)控系統(tǒng)里別等業(yè)務(wù)方來(lái)問(wèn)為什么數(shù)據(jù)還沒(méi)同步過(guò)來(lái)你才發(fā)現(xiàn)問(wèn)題。至于集群模式canal支持多Server ZooKeeper的HA方案多個(gè)instance由ZK協(xié)調(diào)一臺(tái)掛了自動(dòng)切換到另一臺(tái)。這個(gè)配置本身不復(fù)雜但我建議在你的業(yè)務(wù)真正需要高可用之前先把單機(jī)的監(jiān)控和運(yùn)維做扎實(shí)因?yàn)榧阂步鉀Q不了消費(fèi)端邏輯的bug。7. 最后的實(shí)用建議這一章不聊原理、不聊配置就說(shuō)一些零散但真實(shí)的經(jīng)驗(yàn)。第一開(kāi)發(fā)環(huán)境一定要把binlog打開(kāi)。我見(jiàn)過(guò)不少團(tuán)隊(duì)本地MySQL默認(rèn)沒(méi)開(kāi)binlog結(jié)果canal跑不起來(lái)排查半天發(fā)現(xiàn)是本地環(huán)境的鍋。你可以在MySQL配置里直接默認(rèn)開(kāi)啟對(duì)開(kāi)發(fā)環(huán)境性能影響微乎其微。第二消費(fèi)端一定要做冪等。不管你用哪種位點(diǎn)配置重復(fù)消費(fèi)都是可能發(fā)生的所以消費(fèi)邏輯必須能反復(fù)執(zhí)行而結(jié)果一致。比如同步到ES用文檔ID做upsert天然冪等同步到Redis覆蓋寫(xiě)本來(lái)就冪等同步到另一個(gè)MySQL用主鍵做INSERT ... ON DUPLICATE KEY UPDATE就能保證不炸。第三binlog里的時(shí)間字段要小心。canal解析出的時(shí)間字段默認(rèn)是字符串格式時(shí)區(qū)可能有偏移。如果你發(fā)現(xiàn)同步的數(shù)據(jù)在時(shí)間上差了8個(gè)小時(shí)去檢查canal的時(shí)區(qū)配置。第四測(cè)試的時(shí)候別只測(cè)INSERT。我見(jiàn)過(guò)有人Demo只驗(yàn)證了insert就以為一切正常結(jié)果上線后update和delete全沒(méi)處理數(shù)據(jù)越差越多。測(cè)試至少覆蓋增刪改三種事件最好再測(cè)一個(gè)事務(wù)里多條變更的順序性。最后說(shuō)一句掏心窩的話canal入門(mén)真的不難難的是你要理解它和業(yè)務(wù)系統(tǒng)的邊界在哪里。它能幫你解決數(shù)據(jù)同步但它不會(huì)幫你把下游的消費(fèi)邏輯寫(xiě)好也不會(huì)幫你兜住消費(fèi)端拋出的異常。把它當(dāng)成一個(gè)可靠的數(shù)據(jù)搬運(yùn)工真正干活的內(nèi)容還是要靠你精心設(shè)計(jì)消費(fèi)端來(lái)處理。我用canal做了好幾個(gè)項(xiàng)目的增量同步從最開(kāi)始照貓畫(huà)虎跑通Demo到現(xiàn)在能快速定位各種奇怪問(wèn)題最大的感受就是工具本身只是管道關(guān)鍵還是對(duì)數(shù)據(jù)變化鏈路有一顆敬畏之心。希望你也能順著這篇文章跑通自己的第一個(gè)Demo然后在這個(gè)基礎(chǔ)上一步步把數(shù)據(jù)流動(dòng)做得更穩(wěn)、更優(yōu)雅。 Canal實(shí)戰(zhàn)：基于MySQL binlog的增量數(shù)據(jù)同步全解析
1. 先搞清楚canal到底在解決什么問(wèn)題如果說(shuō)這兩年數(shù)據(jù)同步領(lǐng)域哪個(gè)組件最值得花一個(gè)星期認(rèn)真學(xué)一遍我的答案一定是canal。這名字你可能在簡(jiǎn)歷上見(jiàn)過(guò)在招聘JD里見(jiàn)過(guò)甚至在同事的代碼評(píng)審里被吐槽過(guò)但真要問(wèn)你它到底是干嘛的很多人的回答其實(shí)是模糊的嗯就是同步MySQL數(shù)據(jù)的——差不多對(duì)但不夠準(zhǔn)確。我換個(gè)說(shuō)法你就明白了canal做的是MySQL增量數(shù)據(jù)訂閱也就是監(jiān)聽(tīng)MySQL的binlog日志每當(dāng)數(shù)據(jù)庫(kù)里發(fā)生insert、update、delete操作canal就能像裝了竊聽(tīng)器一樣在毫秒級(jí)把這個(gè)變更事件推給你。注意關(guān)鍵詞是增量——它不是全量復(fù)制不是每天導(dǎo)一次數(shù)據(jù)而是從你部署那一刻開(kāi)始實(shí)時(shí)關(guān)注每一條新變化。那它解決的到底是什么問(wèn)題我舉個(gè)最常見(jiàn)的場(chǎng)景。你們系統(tǒng)里肯定有一張熱點(diǎn)表比如商品信息、用戶資料、配置數(shù)據(jù)業(yè)務(wù)方讀得很頻繁。為了扛住流量你引入了Redis做緩存查詢走緩存寫(xiě)操作更新數(shù)據(jù)庫(kù)。聽(tīng)起來(lái)沒(méi)問(wèn)題但更新數(shù)據(jù)庫(kù)的同時(shí)同步緩存這一步寫(xiě)法稍微不嚴(yán)謹(jǐn)就會(huì)出現(xiàn)緩存和數(shù)據(jù)庫(kù)不一致的臟數(shù)據(jù)。常見(jiàn)的做法有幾種雙寫(xiě)、定時(shí)任務(wù)全量刷新、消息隊(duì)列異步通知。雙寫(xiě)邏輯侵入業(yè)務(wù)代碼定時(shí)任務(wù)有延遲且浪費(fèi)資源消息隊(duì)列異步通知倒是好但你需要手動(dòng)把每一次數(shù)據(jù)變更封裝成消息發(fā)出去代碼量不小。canal的思路是我不碰你的業(yè)務(wù)代碼我在數(shù)據(jù)庫(kù)這一側(cè)偷聽(tīng)binlog把變更事件統(tǒng)一抓出來(lái)再推給你的下游。無(wú)論你是想去同步緩存、同步到ES、同步到數(shù)據(jù)倉(cāng)庫(kù)還是同步到另一個(gè)MySQL都不需要改業(yè)務(wù)代碼。這個(gè)解耦的價(jià)值用過(guò)的人都懂——只需要一次配置全公司的數(shù)據(jù)消費(fèi)方都能受益。對(duì)小白來(lái)說(shuō)canal最友好的一點(diǎn)是它自帶一個(gè)客戶端訂閱模型你只要能寫(xiě)Java代碼幾行就能跑通一個(gè)Demo完全不需要一開(kāi)始就理解它的存儲(chǔ)原理。但如果你想用好它光是會(huì)跑肯定不夠因?yàn)檎嬲目尤诩?xì)節(jié)里binlog格式?jīng)]配對(duì)、位點(diǎn)丟失導(dǎo)致重復(fù)消費(fèi)、和多個(gè)訂閱端搶數(shù)據(jù)……這些問(wèn)題我用后面幾個(gè)章節(jié)慢慢展開(kāi)。順便說(shuō)一句canal項(xiàng)目的名稱(chēng)來(lái)自canal這個(gè)英文單詞中文意思是運(yùn)河、水道寓意很直白數(shù)據(jù)從MySQL流向下游的通道。這個(gè)比喻特別貼切你越往后用越會(huì)覺(jué)得它真的就像一條管道把數(shù)據(jù)庫(kù)的血液輸送到各個(gè)需要的地方。2. 安裝前的三個(gè)關(guān)鍵開(kāi)關(guān)JDK、binlog和權(quán)限安裝canal本身不難難的是準(zhǔn)備工作。很多小白按照網(wǎng)上的教程一步步來(lái)結(jié)果客戶端連不上、收不到消息回頭一看全是配套環(huán)境沒(méi)弄對(duì)。所以這一章我把安裝前置條件拆成三個(gè)開(kāi)關(guān)每個(gè)都要提前確認(rèn)壞了哪一個(gè)都會(huì)讓你后面抓狂。2.1 第一個(gè)開(kāi)關(guān)JDK版本canal的server端是Java寫(xiě)的所以你機(jī)器上得有JDK。這里有個(gè)坑需要你避一下canal 1.1.x系列用JDK 8沒(méi)問(wèn)題但如果你用的是比較新的canal 1.1.7版本或者想跑源碼調(diào)試建議直接用JDK 11。我遇到過(guò)用JDK 8跑新版本canal啟動(dòng)的時(shí)候報(bào)了一些奇怪的并發(fā)類(lèi)加載錯(cuò)誤看起來(lái)像是代碼問(wèn)題其實(shí)是編譯目標(biāo)版本太高導(dǎo)致的。裝好之后確認(rèn)一下版本java -version正常情況下你會(huì)看到類(lèi)似openjdk version 1.8.0_***或openjdk version 11.0.***的輸出。如果你的機(jī)器上存在多個(gè)JDK建議用export JAVA_HOME/path/to/jdk指定清楚避免canal的啟動(dòng)腳本拿到錯(cuò)誤的版本。2.2 第二個(gè)開(kāi)關(guān)MySQL必須開(kāi)啟binlogcanal的工作原理是偽裝成MySQL的從節(jié)點(diǎn)向主庫(kù)請(qǐng)求binlog日志。所以你的MySQL必須開(kāi)啟binlog這是整個(gè)鏈條的源頭。修改MySQL配置文件Linux下一般是/etc/my.cnf或/etc/mysql/mysql.conf.d/mysqld.cnf[mysqld] log-binmysql-bin binlog-formatROW server-id1這三行缺一不可尤其第二行binlog-formatROW。我見(jiàn)過(guò)有人在測(cè)試環(huán)境用了默認(rèn)的STATEMENT格式結(jié)果canal解析出來(lái)一堆SQL語(yǔ)句根本拿不到精確的行級(jí)變更數(shù)據(jù)。因?yàn)閏anal的核心輸出是某行某個(gè)字段從舊值變成了新值而這種結(jié)構(gòu)化信息只有ROW格式才完整保留。改完之后記得重啟MySQLservice mysqld restart或者systemctl restart mysqld。然后進(jìn)入MySQL命令行驗(yàn)證SHOW VARIABLES LIKE log_bin; SHOW VARIABLES LIKE binlog_format;兩個(gè)結(jié)果都符合預(yù)期這個(gè)開(kāi)關(guān)才算真正打開(kāi)。還有一個(gè)細(xì)節(jié)容易被忽略server-id不能和canal實(shí)例配置的server-id重復(fù)。我用一個(gè)真實(shí)的踩坑經(jīng)歷說(shuō)明第一次部署的時(shí)候我隨手給canal設(shè)置了server-id1和MySQL自己的server-id沖突結(jié)果canal連接主庫(kù)時(shí)直接被拒日志里報(bào)ERROR: Got error 1236。這個(gè)問(wèn)題排查了很久才定位到因?yàn)閳?bào)錯(cuò)信息很不直觀。所以提前規(guī)劃好MySQL給一個(gè)server-idcanal給另一個(gè)比如101、102這種互不干擾。2.3 第三個(gè)開(kāi)關(guān)MySQL授權(quán)賬號(hào)canal需要用一個(gè)專(zhuān)門(mén)的MySQL賬號(hào)連接主庫(kù)這個(gè)賬號(hào)要有REPLICATION SLAVE和REPLICATION CLIENT權(quán)限。很多人在這步直接用root賬號(hào)倒也能跑通但從規(guī)范角度不推薦尤其到了生產(chǎn)環(huán)境最少權(quán)限原則是底線。創(chuàng)建賬號(hào)的SQL如下CREATE USER canal% IDENTIFIED BY canal123; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; FLUSH PRIVILEGES;這里給%是允許任意主機(jī)連接實(shí)際生產(chǎn)建議限定為運(yùn)行canal的那臺(tái)機(jī)器IP安全第一。另外注意如果數(shù)據(jù)庫(kù)里已有的表結(jié)構(gòu)比較復(fù)雜canal在解析DDL事件時(shí)可能還需要訪問(wèn)表定義所以順手把SELECT權(quán)限也加上。這三件事做完環(huán)境準(zhǔn)備基本就緒。老實(shí)說(shuō)這個(gè)過(guò)程十分鐘都用不了但價(jià)值遠(yuǎn)超十分鐘——它決定了你后面所有排查動(dòng)作能不能有一個(gè)干凈的基礎(chǔ)。3. 一條數(shù)據(jù)從MySQL到客戶端的完整路徑很多教程上來(lái)就讓你裝canal、跑Demo等你跑通了還是一頭霧水它到底是怎么拿到數(shù)據(jù)的這一章我從原理層面完整講一遍大概需要你花五分鐘耐心讀讀完之后你就擁有了排查問(wèn)題的底層地圖遇到任何詭異現(xiàn)象都能知道往哪個(gè)環(huán)節(jié)去查。3.1 先理解MySQL主從復(fù)制是怎么回事canal的核心機(jī)制其實(shí)是在模仿MySQL的從庫(kù)。MySQL的主從復(fù)制過(guò)程很簡(jiǎn)單主庫(kù)把每一次變更寫(xiě)入binlog從庫(kù)連接到主庫(kù)請(qǐng)求從某個(gè)binlog位點(diǎn)開(kāi)始的數(shù)據(jù)拿到之后在本地重放。canal做的事情和從庫(kù)一模一樣只不過(guò)它拿到這些日志之后重放的不是SQL而是把日志解析成結(jié)構(gòu)化的數(shù)據(jù)變更事件交給下游程序處理。這個(gè)過(guò)程涉及binlog里的幾個(gè)概念我用大白話解釋一下binlog文件MySQL落盤(pán)生成的日志文件比如mysql-bin.000001達(dá)到一定大小后自動(dòng)切換下一個(gè)文件。binlog位點(diǎn)position一條日志在文件中的偏移量它是唯一的?？蛻舳嘶謴?fù)消費(fèi)時(shí)只要記住上次消費(fèi)到的position就能從斷點(diǎn)繼續(xù)。GTID全局事務(wù)ID是比position更高級(jí)的定位方式不依賴(lài)具體文件名和偏移量。新版本canal推薦用GTID模式主從切換時(shí)能自動(dòng)適應(yīng)。3.2 canal內(nèi)部的三段式流水線canal從拿到binlog到吐出數(shù)據(jù)內(nèi)部經(jīng)歷了三個(gè)核心階段第一階段抓取Parser。canal偽裝成從庫(kù)向MySQL發(fā)送dump協(xié)議請(qǐng)求。MySQL就會(huì)從指定的binlog位點(diǎn)開(kāi)始把日志一條條推送給canal。這個(gè)階段最純粹只負(fù)責(zé)拿。第二階段解析Sink。拿到的binlog是事件流需要解析。比如一個(gè)UPDATE事件會(huì)解析出舊值before和新值after這個(gè)是canal最有價(jià)值的輸出——下游可以直接拿到{id: 1, name: 舊值}變成{id: 1, name: 新值}這樣的JSON數(shù)據(jù)而不需要自己拼SQL去反查。第三階段存儲(chǔ)Store。解析完成后事件會(huì)寫(xiě)入一個(gè)內(nèi)存環(huán)形隊(duì)列等待客戶端來(lái)拉取。canal內(nèi)置了一個(gè)基于內(nèi)存的存儲(chǔ)組件默認(rèn)容量是16384個(gè)事件客戶端消費(fèi)速度跟不上時(shí)事件會(huì)在內(nèi)存里排隊(duì)排滿了之后才會(huì)出現(xiàn)丟數(shù)據(jù)或者阻塞的情況。整個(gè)鏈條如下圖所示文字版MySQL binlog → canal Server 抓取 → 解析 → 內(nèi)存隊(duì)列 → 客戶端(Client) 拉取 → 你的業(yè)務(wù)代碼/ES/Redis...3.3 你眼里的canal組件Server、Instance和Client實(shí)際部署時(shí)你會(huì)接觸到幾個(gè)概念這里一次性講清楚避免后續(xù)被配置項(xiàng)繞暈canal Server一個(gè)獨(dú)立運(yùn)行的Java進(jìn)程核心服務(wù)。canal InstanceServer內(nèi)部的一個(gè)邏輯實(shí)例每個(gè)Instance對(duì)應(yīng)一個(gè)MySQL數(shù)據(jù)源。一個(gè)Server可以配置多個(gè)Instance每個(gè)Instance有獨(dú)立的配置、獨(dú)立的訂閱位點(diǎn)、獨(dú)立的隊(duì)列。canal Client你寫(xiě)的消費(fèi)端程序通過(guò)TCP協(xié)議從Server拉取數(shù)據(jù)。打個(gè)比方Server是餐廳Instance是廚房里的一個(gè)灶臺(tái)——每個(gè)灶臺(tái)對(duì)應(yīng)一個(gè)供應(yīng)商MySQLClient是來(lái)吃飯的顧客。一個(gè)餐廳有多個(gè)灶臺(tái)很正常每個(gè)灶臺(tái)負(fù)責(zé)炒不同的菜互不干擾。從部署角度小白剛開(kāi)始只需要一個(gè)Server、一個(gè)Instance、一個(gè)Client后面需要訂閱多個(gè)庫(kù)或者對(duì)接多套下游時(shí)再考慮多Instance。4. 跑通第一個(gè)Demo用Java代碼訂閱一條insert前面講了那么多理論和準(zhǔn)備這一章我們真正動(dòng)手。我用一臺(tái)Linux機(jī)器做演示canal部署包從官方GitHub Release頁(yè)面下載最新穩(wěn)定版即可解壓后目錄結(jié)構(gòu)大致如下/canal ├── bin ├── conf │ ├── canal.properties │ ├── example │ │ └── instance.propertiesconf/canal.properties是全局配置conf/example是默認(rèn)實(shí)例的配置目錄instance.properties是實(shí)例級(jí)配置。4.1 修改實(shí)例配置先改instance.properties這是核心# MySQL主庫(kù)地址 canal.instance.master.address127.0.0.1:3306 # 連接MySQL的賬號(hào)密碼 canal.instance.dbUsernamecanal canal.instance.dbPasswordcanal123 # 訂閱的數(shù)據(jù)庫(kù)和表*表示所有 canal.instance.filter.regextest\\..* # 位點(diǎn)定位方式 canal.instance.gtidfalse重點(diǎn)說(shuō)明一下filter.regex這個(gè)配置。它的格式是數(shù)據(jù)庫(kù)名\\..*比如我只想訂閱test庫(kù)的所有表就寫(xiě)test\\..*想訂閱test庫(kù)下的user表就寫(xiě)test\\.user想訂閱多個(gè)庫(kù)用|分隔。這個(gè)正則很容易寫(xiě)錯(cuò)因?yàn)镴ava字符串本身也要轉(zhuǎn)義所以\\.寫(xiě)成\\.而不是\.。我見(jiàn)過(guò)有人照抄配置把轉(zhuǎn)義符弄丟了結(jié)果canal直接不輸出任何數(shù)據(jù)查了半天才發(fā)現(xiàn)是正則匹配問(wèn)題。4.2 修改全局配置再看canal.properties重點(diǎn)確認(rèn)以下幾項(xiàng)canal.port11111 canal.destinationsexample canal.instance.global.modetcpcanal.port是客戶端連接的TCP端口默認(rèn)11111不用改。canal.destinations指定啟動(dòng)哪些實(shí)例默認(rèn)example對(duì)應(yīng)conf目錄下的example文件夾。canal.instance.global.modetcp表示消費(fèi)端通過(guò)TCP直連方式拉取數(shù)據(jù)——這是最簡(jiǎn)單的方式后面聊MQ模式時(shí)再說(shuō)。4.3 啟動(dòng)Servercd /canal/bin ./startup.sh啟動(dòng)后看日志確認(rèn)沒(méi)有異常tail -f /canal/logs/canal/canal.log如果看到canal server is running now之類(lèi)的字樣并且沒(méi)有報(bào)錯(cuò)說(shuō)明Server起來(lái)了。再檢查實(shí)例狀態(tài)tail -f /canal/logs/example/example.log正常情況下能看到start successful。這里有個(gè)新手最容易遇到的情況日志里顯示canal is starting然后卡住不動(dòng)多半是MySQL連接沒(méi)通去查example.log里有沒(méi)有connect timeout或者Access denied字樣。4.4 寫(xiě)一個(gè)最小客戶端canal官方提供了一個(gè)客戶端庫(kù)canal.client用Maven引入dependency groupIdcom.alibaba.otter/groupId artifactIdcanal.client/artifactId version1.1.7/version /dependency然后寫(xiě)一個(gè)最簡(jiǎn)單的Java類(lèi)核心邏輯是建立連接 → 拉取消息 → 解析事件 → 打印變更內(nèi)容。import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.CanalEntry.EntryType; import java.net.InetSocketAddress; import java.util.List; public class CanalDemo { public static void main(String[] args) { // 連接canal server CanalConnector connector CanalConnectors.newSingleConnector( new InetSocketAddress(127.0.0.1, 11111), example, // instance名稱(chēng) , // 無(wú)認(rèn)證時(shí)填空 ); connector.connect(); // 訂閱所有表這里要和instance的filter.regex對(duì)應(yīng) connector.subscribe(test\\..*); while (true) { // 一次拉取100條不設(shè)置超時(shí) Message message connector.getWithoutAck(100); long batchId message.getId(); if (batchId -1 || message.getEntries().isEmpty()) { // 沒(méi)有數(shù)據(jù)睡1秒再試 try { Thread.sleep(1000); } catch (InterruptedException e) {} continue; } for (CanalEntry.Entry entry : message.getEntries()) { if (entry.getEntryType() ! EntryType.ROWDATA) { continue; } CanalEntry.RowChange rowChange CanalEntry.RowChange.parseFrom(entry.getStoreValue()); CanalEntry.EventType eventType rowChange.getEventType(); for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { if (eventType CanalEntry.EventType.INSERT) { System.out.println(INSERT: rowData.getAfterColumnsList()); } else if (eventType CanalEntry.EventType.UPDATE) { System.out.println(UPDATE before: rowData.getBeforeColumnsList()); System.out.println(UPDATE after: rowData.getAfterColumnsList()); } else if (eventType CanalEntry.EventType.DELETE) { System.out.println(DELETE: rowData.getBeforeColumnsList()); } } } // 確認(rèn)消費(fèi)成功 connector.ack(batchId); } } }這段代碼看著有點(diǎn)長(zhǎng)但核心就三件事連接、拉取、確認(rèn)。所有canal客戶端的骨架都是這個(gè)模式你以后無(wú)論用Spring Boot封裝還是寫(xiě)其他語(yǔ)言的客戶端都逃不開(kāi)這個(gè)套路。4.5 實(shí)測(cè)驗(yàn)證運(yùn)行上面的Java類(lèi)然后在MySQL里執(zhí)行INSERT INTO test.user (name, age) VALUES (張三, 25);立刻就能在控制臺(tái)看到類(lèi)似下面的輸出INSERT: [name: 張三, age: 25]看到這一行你的canal入門(mén)第一個(gè)Demo就算真正跑通了。從這一刻起任何數(shù)據(jù)庫(kù)變更都會(huì)實(shí)時(shí)出現(xiàn)在你的消費(fèi)端面前剩下的就是你怎么利用這些數(shù)據(jù)了。5. 小白最容易踩的五個(gè)坑含完整排查思路說(shuō)實(shí)話canal的部署和Demo真的很簡(jiǎn)單但越是這種看起來(lái)簡(jiǎn)單的組件越容易在細(xì)節(jié)上翻車(chē)。這一章我把自己和周?chē)瞬冗^(guò)的坑整理出來(lái)每一項(xiàng)都附上排查鏈路而不是直接給結(jié)論——因?yàn)橄麓文阌龅降目赡苤皇莻€(gè)相似但不完全一樣的問(wèn)題思路比答案更重要。5.1 坑一binlog格式不是ROW解析結(jié)果全是SQL現(xiàn)象客戶端能連上能收到消息但解析出來(lái)的RowData是空的只有SQL語(yǔ)句。排查鏈路第一步回到MySQL執(zhí)行SHOW VARIABLES LIKE binlog_format;如果結(jié)果是STATEMENT或MIXED基本就是問(wèn)題所在。第二步修改配置為ROW重啟MySQL。第三步重新執(zhí)行一條INSERT看canal是否輸出結(jié)構(gòu)化的列值。根因只有ROW格式才會(huì)記錄每一行變更的具體列值STATEMENT格式記錄的是原始SQL文本canal無(wú)法從中還原出行級(jí)別數(shù)據(jù)。5.2 坑二server-id沖突導(dǎo)致連接被拒現(xiàn)象canal日志里出現(xiàn)ERROR: Got error 1236或者直接連接不上MySQL。排查鏈路第一步檢查MySQL的server-idSELECT server_id;第二步檢查canal的canal.instance.properties里有沒(méi)有canal.instance.mysql.slaveId配置項(xiàng)這個(gè)值默認(rèn)是123456。如果和MySQL的server-id一樣就改成一個(gè)不同的值比如654321。第三步重啟canal實(shí)例觀察日志。根因MySQL主從復(fù)制機(jī)制要求每個(gè)從節(jié)點(diǎn)的server-id必須全局唯一。canal偽裝成一個(gè)從節(jié)點(diǎn)如果它的server-id和MySQL主庫(kù)相同主庫(kù)會(huì)認(rèn)為這是我自己直接拒絕復(fù)制請(qǐng)求。這里插一句生產(chǎn)中很多人會(huì)忽略這個(gè)參數(shù)因?yàn)樗哪J(rèn)值看起來(lái)沒(méi)存在感。建議部署清單里專(zhuān)門(mén)加一行確認(rèn)slaveId與所有MySQL節(jié)點(diǎn)均不重復(fù)。5.3 坑三位點(diǎn)丟失導(dǎo)致重復(fù)消費(fèi)或漏消費(fèi)現(xiàn)象canal重啟后客戶端收到了之前已經(jīng)處理過(guò)的舊數(shù)據(jù)或者直接跳過(guò)了部分?jǐn)?shù)據(jù)。排查鏈路第一步確認(rèn)canal的meta信息存儲(chǔ)在哪里。默認(rèn)情況下位點(diǎn)存在內(nèi)存中Server重啟就會(huì)丟失從當(dāng)前最新位置開(kāi)始消費(fèi)。如果你需要持久化要在canal.properties里配置canal.instance.global.spring.xmlclasspath:spring/memory-instance.xml改為file-instance.xml可以把meta寫(xiě)到本地文件。第二步檢查客戶端代碼是否正確ack。只get不ack重啟后會(huì)從上次ack的位置重新拉取產(chǎn)生重復(fù)消費(fèi)ack太快也可能出問(wèn)題如果先ack再處理業(yè)務(wù)邏輯處理失敗時(shí)數(shù)據(jù)就再也拉不回來(lái)了這是至少一次語(yǔ)義下最常見(jiàn)的取舍。第三步設(shè)計(jì)消費(fèi)冪等。無(wú)論位點(diǎn)多可靠分布式環(huán)境下重復(fù)消費(fèi)幾乎不可避免所以消費(fèi)端的冪等才是最終兜底。根因canal的位點(diǎn)機(jī)制是經(jīng)典的有且至少一次消費(fèi)語(yǔ)義重復(fù)消費(fèi)是常態(tài)漏消費(fèi)是因?yàn)槲稽c(diǎn)推進(jìn)到了業(yè)務(wù)成功之前。5.4 坑四多個(gè)客戶端實(shí)例同時(shí)消費(fèi)同一個(gè)instance現(xiàn)象多個(gè)服務(wù)實(shí)例都連了同一個(gè)canal destination結(jié)果每條數(shù)據(jù)只被其中一個(gè)實(shí)例收到而不是每個(gè)實(shí)例都收到。排查鏈路第一步確認(rèn)你的業(yè)務(wù)訴求是廣播還是負(fù)載均衡。canal不保證多個(gè)client同時(shí)消費(fèi)時(shí)每個(gè)client都能拿到全量數(shù)據(jù)默認(rèn)是競(jìng)爭(zhēng)消費(fèi)——一條數(shù)據(jù)只會(huì)給一個(gè)client。第二步如果確實(shí)需要多個(gè)下游分別消費(fèi)同一份數(shù)據(jù)可以有三種方案一是每個(gè)下游對(duì)應(yīng)一個(gè)獨(dú)立的instance二是通過(guò)MQ模式發(fā)布訂閱三是使用canal的CanalMQStarter配多個(gè)不同的group。根因canal的tcp模式的定位是一個(gè)destination對(duì)應(yīng)一個(gè)消費(fèi)組多個(gè)client是在消費(fèi)組內(nèi)競(jìng)爭(zhēng)拉取不是復(fù)制分發(fā)。5.5 坑五DDL事件和事務(wù)邊界導(dǎo)致的數(shù)據(jù)異常現(xiàn)象執(zhí)行ALTER TABLE之后消費(fèi)端收到的數(shù)據(jù)突然解析失敗或者一個(gè)事務(wù)內(nèi)的多變更變得不完整。排查鏈路第一步檢查消費(fèi)端是否處理了EntryType.ROWDATA之外的事件。DDL事件屬于ENTRY_TYPE的其他類(lèi)型如果你在消費(fèi)端直接跳過(guò)所有非ROWDATA事件DDL對(duì)下游可能造成結(jié)構(gòu)不一致。第二步開(kāi)啟canal的并行消費(fèi)時(shí)要注意事務(wù)一致性。如果下游是數(shù)據(jù)庫(kù)或緩存建議按事務(wù)邊界整體提交避免同一個(gè)事務(wù)的變更被拆到不同地方執(zhí)行。根因canal按事務(wù)為單位讀取binlog但消費(fèi)端如果并行處理事務(wù)邊界會(huì)被人為拆散。這五個(gè)坑基本覆蓋了我見(jiàn)過(guò)的90%小白問(wèn)題。說(shuō)實(shí)話每個(gè)坑背后都對(duì)應(yīng)一個(gè)理解盲區(qū)不是性能問(wèn)題不是bug而是對(duì)canal是怎么工作的理解不到位。所以每次出問(wèn)題先回到第3章的原理圖里定位是在哪個(gè)環(huán)節(jié)排查速度會(huì)快很多。6. 進(jìn)階思路從單機(jī)Demo走向生產(chǎn)環(huán)境Demo跑通、坑也避開(kāi)之后你可能會(huì)問(wèn)那真實(shí)項(xiàng)目里一般怎么用這一章我講三個(gè)最常用的落地方向以及小白從單機(jī)走向生產(chǎn)時(shí)需要補(bǔ)的功課。6.1 實(shí)戰(zhàn)方向一數(shù)據(jù)庫(kù)變更同步到Redis這是canal最常見(jiàn)的用法。假設(shè)你的商品表product里有價(jià)格字段價(jià)格變了緩存也得變。傳統(tǒng)做法是改業(yè)務(wù)代碼在更新數(shù)據(jù)庫(kù)后刪除緩存用canal之后消費(fèi)端收到UPDATE事件直接從JSON里提取最新值執(zhí)行SET key value或者刪除緩存。業(yè)務(wù)代碼零侵入。要注意兩點(diǎn)第一canal發(fā)來(lái)的是列名值的結(jié)構(gòu)不是SQL所以你需要自己把變更映射成業(yè)務(wù)對(duì)象這個(gè)映射邏輯一般寫(xiě)在消費(fèi)端一個(gè)獨(dú)立模塊里第二如果變更頻繁要注意給緩存加防抖比如1秒內(nèi)同一行的多次變更合并成一次刷緩存避免抖動(dòng)。6.2 實(shí)戰(zhàn)方向二異構(gòu)數(shù)據(jù)同步到Elasticsearch把MySQL的數(shù)據(jù)同步到ES做全文檢索也是高頻場(chǎng)景。流程大致是canal監(jiān)聽(tīng)MySQL變更 → 消費(fèi)端把變更拼成ES文檔 → 調(diào)用ES的index/update接口。這里有個(gè)常見(jiàn)的設(shè)計(jì)問(wèn)題ES里可能需要寬表比如訂單文檔里要包含用戶名、商品名而這些字段來(lái)自其他表。如果只訂閱訂單表拿不到關(guān)聯(lián)信息。解決思路有兩個(gè)一是訂閱多張表在消費(fèi)端做內(nèi)存關(guān)聯(lián)二是讓業(yè)務(wù)方寫(xiě)出包含所有所需字段的預(yù)聚合表canal只訂閱這一張。第二個(gè)方案更簡(jiǎn)單可靠推薦小白先這么做。6.3 從單機(jī)到集群三個(gè)生產(chǎn)級(jí)配置生產(chǎn)環(huán)境不可能只跑一個(gè)單機(jī)canal但我也要提醒你一開(kāi)始別急著上集群先把下面三件事做好比盲目加機(jī)器更有價(jià)值。第一個(gè)是位點(diǎn)持久化。前面說(shuō)過(guò)默認(rèn)位點(diǎn)在內(nèi)存在Server重啟就丟。生產(chǎn)環(huán)境務(wù)必配置成file模式或zookeeper模式。我用的是file模式最簡(jiǎn)單但要注意磁盤(pán)空間和文件權(quán)限。第二個(gè)是訂閱粒度收斂。開(kāi)發(fā)環(huán)境訂閱所有庫(kù)所有表沒(méi)問(wèn)題生產(chǎn)環(huán)境一定要用filter.regex收窄到具體業(yè)務(wù)庫(kù)和表不然會(huì)白白消耗大量性能還可能把其他團(tuán)隊(duì)的數(shù)據(jù)變更也帶進(jìn)你的消費(fèi)邏輯。第三個(gè)是監(jiān)控告警。canal是實(shí)時(shí)鏈路一旦它停了下游全部數(shù)據(jù)滯后。至少要做到進(jìn)程存活監(jiān)控、消費(fèi)延遲監(jiān)控canal有delayed指標(biāo)、客戶端積壓消息數(shù)監(jiān)控。這些指標(biāo)都可以通過(guò)JMX或者集成到現(xiàn)有監(jiān)控系統(tǒng)里別等業(yè)務(wù)方來(lái)問(wèn)為什么數(shù)據(jù)還沒(méi)同步過(guò)來(lái)你才發(fā)現(xiàn)問(wèn)題。至于集群模式canal支持多Server ZooKeeper的HA方案多個(gè)instance由ZK協(xié)調(diào)一臺(tái)掛了自動(dòng)切換到另一臺(tái)。這個(gè)配置本身不復(fù)雜但我建議在你的業(yè)務(wù)真正需要高可用之前先把單機(jī)的監(jiān)控和運(yùn)維做扎實(shí)因?yàn)榧阂步鉀Q不了消費(fèi)端邏輯的bug。7. 最后的實(shí)用建議這一章不聊原理、不聊配置就說(shuō)一些零散但真實(shí)的經(jīng)驗(yàn)。第一開(kāi)發(fā)環(huán)境一定要把binlog打開(kāi)。我見(jiàn)過(guò)不少團(tuán)隊(duì)本地MySQL默認(rèn)沒(méi)開(kāi)binlog結(jié)果canal跑不起來(lái)排查半天發(fā)現(xiàn)是本地環(huán)境的鍋。你可以在MySQL配置里直接默認(rèn)開(kāi)啟對(duì)開(kāi)發(fā)環(huán)境性能影響微乎其微。第二消費(fèi)端一定要做冪等。不管你用哪種位點(diǎn)配置重復(fù)消費(fèi)都是可能發(fā)生的所以消費(fèi)邏輯必須能反復(fù)執(zhí)行而結(jié)果一致。比如同步到ES用文檔ID做upsert天然冪等同步到Redis覆蓋寫(xiě)本來(lái)就冪等同步到另一個(gè)MySQL用主鍵做INSERT ... ON DUPLICATE KEY UPDATE就能保證不炸。第三binlog里的時(shí)間字段要小心。canal解析出的時(shí)間字段默認(rèn)是字符串格式時(shí)區(qū)可能有偏移。如果你發(fā)現(xiàn)同步的數(shù)據(jù)在時(shí)間上差了8個(gè)小時(shí)去檢查canal的時(shí)區(qū)配置。第四測(cè)試的時(shí)候別只測(cè)INSERT。我見(jiàn)過(guò)有人Demo只驗(yàn)證了insert就以為一切正常結(jié)果上線后update和delete全沒(méi)處理數(shù)據(jù)越差越多。測(cè)試至少覆蓋增刪改三種事件最好再測(cè)一個(gè)事務(wù)里多條變更的順序性。最后說(shuō)一句掏心窩的話canal入門(mén)真的不難難的是你要理解它和業(yè)務(wù)系統(tǒng)的邊界在哪里。它能幫你解決數(shù)據(jù)同步但它不會(huì)幫你把下游的消費(fèi)邏輯寫(xiě)好也不會(huì)幫你兜住消費(fèi)端拋出的異常。把它當(dāng)成一個(gè)可靠的數(shù)據(jù)搬運(yùn)工真正干活的內(nèi)容還是要靠你精心設(shè)計(jì)消費(fèi)端來(lái)處理。我用canal做了好幾個(gè)項(xiàng)目的增量同步從最開(kāi)始照貓畫(huà)虎跑通Demo到現(xiàn)在能快速定位各種奇怪問(wèn)題最大的感受就是工具本身只是管道關(guān)鍵還是對(duì)數(shù)據(jù)變化鏈路有一顆敬畏之心。希望你也能順著這篇文章跑通自己的第一個(gè)Demo然后在這個(gè)基礎(chǔ)上一步步把數(shù)據(jù)流動(dòng)做得更穩(wěn)、更優(yōu)雅。