Springboot2.3.x整合Canal的示例代碼
一、故事背景
前言…
最近工作中遇到瞭一個數據同步的問題
我們這邊系統的一個子業務需要依賴另一個系統的數據,當另一個系統數據變更時,我們這邊的數據庫要對數據進行同步…
那麼我自己想到的同步方式呢就兩種:
1、MQ訂閱,另一個系統數據變更後將變更數據方式到MQ 我們這邊訂閱接受
2、數據庫的觸發器
但是呢,兩者都被組長paas瞭!
1、MQ呢,會造成代碼侵入,但是另一個系統暫時不會做任何代碼更改…
2、數據庫的觸發器會直接跟生產數據庫強關聯,會搶占資源,甚至有可能造成生產數據庫的不穩定…
對此很是苦惱…
於是啊,隻能借由強大的google、百度,看看能不能解決我這個問題!一番搜索,有學習瞭一個很有趣的東西…
Canal
二、什麼是Canal
canal:阿裡開源mysql binlog 數據組件
官網解釋的相當詳細瞭(國產牛逼)…下邊我也是照搬過來的…
官網地址如下:https://github.com/alibaba/canal/wiki
早期,阿裡巴巴B2B公司因為存在杭州和美國雙機房部署,存在跨機房同步的業務需求。不過早期的數據庫同步業務,主要是基於trigger的方式獲取增量變更,不過從2010年開始,阿裡系公司開始逐步的嘗試基於數據庫的日志解析,獲取增量變更進行同步,由此衍生出瞭增量訂閱&消費的業務,從此開啟瞭一段新紀元。ps. 目前內部使用的同步,已經支持mysql5.x和oracle部分版本的日志解析
canal [kə’næl],譯意為水道/管道/溝渠,主要用途是基於 MySQL 數據庫增量日志解析,提供增量數據訂閱和消費
工作原理
- canal 模擬 MySQL slave 的交互協議,偽裝自己為 MySQL slave ,向 MySQL master 發送 dump 協議
- MySQL master 收到 dump 請求,開始推送 binary log 給 slave (即 canal )
- canal 解析 binary log 對象(原始為 byte 流)
canal呢,實際是就是運用瞭Mysql的主從復制原理…
MySQL主從復制實現
復制遵循三步過程:
- 主服務器將更改記錄到binlog中(這些記錄稱為binlog事件,可以通過來查看
show binary events
) - 從服務器將主服務器的二進制日志事件復制到其中繼日志。
- 中繼日志中的從服務器重做事件隨後將更新其舊數據。
如何運作
原理很簡單:
- Canal模擬MySQL從站的交互協議,偽裝成MySQL從站,然後將轉儲協議發送到MySQL主服務器。
- MySQL Master接收到轉儲請求,並開始將二進制日志推送到slave(即運河)。
- 運河將二進制日志對象解析為其自己的數據類型(最初為字節流)
通過官網的介紹,讓我們瞭解到,canal實際上就是偽裝為瞭一個從庫,我們隻需要訂閱到數據變更的主庫,那麼canal就會以從庫的身份讀取到其主庫的binlog日志!我們拿到canal解析好的binlog日志信息,就等於拿到瞭變更的數據啦!…
這樣的話呢,我們即保證瞭不影響其系統數據庫正常使用,又不會侵入他的項目代碼,一舉兩得
ok,接下來開始實戰篇…
三、Canal安裝
(1)事前準備
(1)數據庫開啟binlog
使用canal呢,有一個前提條件,即被訂閱的數據庫需要開啟binlog
如何查看是否開啟binlog呢?
登錄服務器上數據庫或在可視化工具中 執行查詢語句: 如果出現 log_bin ON 表示已開啟Binlog
show variables like 'log_bin';
如果服務器上的數據庫為自己安裝的,則找到配置文件my.conf 添加以下內容,如果買的雲實例,則詢問廠商開啟即可
在my.conf文件中的 [mysqld] 下添加以下三行內容
log-bin=mysql-bin # 開啟 binlog binlog-format=ROW # 選擇 ROW 模式 讀行 server_id=1 # 配置 MySQL replaction 需要定義,不要和 canal 的 slaveId 重復
(2)數據庫新建賬號,開啟MySQL slav權限
canaltest:作為slave 角色的賬戶 Canal123…:為密碼
CREATE USER canaltest IDENTIFIED BY 'Canal123..'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canaltest'@'%'; GRANT ALL PRIVILEGES ON *.* TO 'canaltest'@'%' ; FLUSH PRIVILEGES;
連接測試
那麼到這裡,準備工作就好瞭!
可能呢,有的小夥伴有點懵,你這是在幹啥?那麼咱們就來理那麼一理! 敲黑板瞭哈!
1、事前準備,是針對於訂閱數據庫的(即主庫)
2、實際步驟也就兩步 1:更改配置,開啟binlog 2:設置新賬號,賦予slave權限,供canal讀取Binlog橋梁使用
3、以上操作與canal本身沒啥關系,僅僅是使用canal的前提條件罷遼…
(2)Canal Admin 安裝
canal admin 是 一個可視化的 canal web管理運維工程,脫離以往服務器運維,面向web…
canal-admin設計上是為canal提供整體配置管理、節點運維等面向運維的功能,提供相對友好的WebUI操作界面,方便更多用戶快速和安全的操作
canal-admin的限定依賴:
- MySQL,用於存儲配置和節點等相關數據
- canal版本,要求>=1.1.4 (需要依賴canal-server提供面向admin的動態運維管理接口)
- 需要JRE 環境 (安裝JDK)
下載
wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.admin-1.1.4.tar.gz
解壓
mkdir /usr/local/canal-admin tar zxvf canal.admin-1.1.4.tar.gz -C /usr/local/canal-admin
進入canal-admin目錄下查看
cd /usr/local/canal-admin
修改配置
vim conf/application.yml
裡邊的配置 按照自己的實際情況更改…
server: port: 8089 spring: jackson: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 #這裡是配置canal-admin 所依賴的數據庫,,,存放web管理中設置的配置等,,, spring.datasource: address: 127.0.0.1:3306 database: canal_manager username: root password: 123456 driver-class-name: com.mysql.jdbc.Driver url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false hikari: maximum-pool-size: 30 minimum-idle: 1 # 連接所用的賬戶密碼 canal: adminUser: admin adminPasswd: leitest
導入canaladmin 所需要的數據庫文件
這裡需要註意瞭,要和 application.yml中的數據庫名對應,你可以選擇命令導入,也可以Navicat 可視化拖sql文件導入…一切…看你喜歡.
我這個玩canal的服務器呢,是新安裝的,mysql直接用docker安裝即可,具體可查看我的博客:
Docker在CentOS7下不能下載鏡像timeout的解決辦法(圖解)
CentOS 7安裝Docker
需要註意的是,使用docker 安裝的mysql 是無法直接使用 mysql -uroot -p
命令的哦,需要先將腳本復制到容器中,docker不熟練或覺得麻煩的同鞋,請直接使用Navicat可視化工具…
導入canal-admin服務所必需的sql文件
如果是服務器軟件軟件安裝的mysql 則直接執行以下命令即可
mysql -uroot -p #......... # 導入初始化SQL > source conf/canal_manager.sql
啟動
直接執行啟動腳本即可
cd bin ./startup.sh
默認賬戶密碼:
admin:123456
(3)Canal Server 安裝
canal-server 才是canal的核心我們前邊所講的canal的功能,實際上講述的就是canal-server的功能…admin 僅僅隻是一個web管理而已,不要搞混主次關系…
下載
wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.deployer-1.1.4.tar.gz
解壓
mkdir /usr/local/canal-server tar zxvf canal.deployer-1.1.4.tar.gz -C /usr/local/canal-server
啟動,並連接到canal-admin web端
首先,我們需要修改配置文件
cd /usr/local/canal-server vim /conf/canal_local.properties
註意瞭,密碼如何加密!!!
要記得,前邊 canal-admin 的 aplication.yml 中設置瞭賬戶密碼為 admin:leitest
# 連接所用的賬戶密碼 canal: adminUser: admin adminPasswd: leitest
所以,我們這裡需要對明文 leitest 加密並替換即可
使用數據庫函數 PASSWORD 加密即可
SELECT PASSWORD(‘要加密的明文’),然後去掉前邊的* 號就行
啟動並連接到admin
sh bin/startup.sh local
查看端口看是否有 11110 、11111、11112
netstat -untlp
看瞭一下,發現沒有,說明server 沒有啟動成功
看下日志
vim logs/canal/canal.log
解決辦法:
1、canal-admin 先停止後從起
2、canal server 先以之前的形式運行,不輸入後邊 local 命令
3、關閉canal server
4、再以canal server 連接 admin 形式啟動
admin頁面上新建server
修改配置,註釋 (instance連接信息,我們還是以前邊設置的 admin:leitest 為準,所有這裡需要註釋掉,如果不註釋,那麼我們代碼中連接則需要使用此賬號以及密碼)
接下來咱們創建instance
如何理解server 和instance 呢,我認為,可以把它當做 java 中的 class 和 bean 即 類和對象
server 為類 instance 為其具體的實例對象 ,可創建多個不同的實例…
而我們這邊監聽到主庫變化的呢,則是根據業務,對不同的實例即(instance )做不同配置即可…
根據自己情況進行過濾數據
canal.instance.filter.regex | mysql 數據解析關註的表,Perl正則表達式.多個正則之間以逗號(,)分隔,轉義符需要雙斜杠(\) 常見例子:1. 所有表:.* or .\… 2. canal schema下所有表: canal\…* 3. canal下的以canal打頭的表:canal\.canal.* 4. canal schema下的一張表:canal\.test15. 多個規則組合使用:canal\…*,mysql.test1,mysql.test2 (逗號分隔) | ||
---|---|---|---|
canal.instance.filter.druid.ddl | 是否使用druid處理所有的ddl解析來獲取庫和表名 | true | |
canal.instance.filter.query.dcl | 是否忽略dcl語句 | false | |
canal.instance.filter.query.dml | 是否忽略dml語句 (mysql5.6之後,在row模式下每條DML語句也會記錄SQL到binlog中,可參考MySQL文檔) | false | |
canal.instance.filter.query.ddl | 是否忽略ddl語句 | false |
更多設置請見官網:https://github.com/alibaba/canal/wiki/AdminGuide
如此一來,一個簡單的canal環境就搭建好瞭,接下來,咱們開始測試吧!
(4)springboot demo示例
引入canal所需依賴
<dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.1.4</version> </dependency>
配置
canal: # instance 實例所在ip host: 192.168.96.129 # tcp通信端口 port: 11111 # 賬號 canal-admin application.yml 設置的 username: admin # 密碼 password: leitest #實例名稱 instance: test
代碼
package com.leilei; import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.Message; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; import org.springframework.stereotype.Component; import java.net.InetSocketAddress; import java.util.List; /** * @author lei * @version 1.0 * @date 2020/9/27 22:23 * @desc 讀取binlog日志 */ @Component public class ReadBinLogService implements ApplicationRunner { @Value("${canal.host}") private String host; @Value("${canal.port}") private int port; @Value("${canal.username}") private String username; @Value("${canal.password}") private String password; @Value("${canal.instance}") private String instance; @Override public void run(ApplicationArguments args) throws Exception { CanalConnector conn = getConn(); while (true) { conn.connect(); //訂閱實例中所有的數據庫和表 conn.subscribe(".*\\..*"); // 回滾到未進行ack的地方 conn.rollback(); // 獲取數據 每次獲取一百條改變數據 Message message = conn.getWithoutAck(100); long id = message.getId(); int size = message.getEntries().size(); if (id != -1 && size > 0) { // 數據解析 analysis(message.getEntries()); }else { Thread.sleep(1000); } // 確認消息 conn.ack(message.getId()); // 關閉連接 conn.disconnect(); } } /** * 數據解析 */ private void analysis(List<CanalEntry.Entry> entries) { for (CanalEntry.Entry entry : entries) { // 隻解析mysql事務的操作,其他的不解析 if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN) { continue; if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) { // 解析binlog CanalEntry.RowChange rowChange = null; try { rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); } catch (Exception e) { throw new RuntimeException("解析出現異常 data:" + entry.toString(), e); if (rowChange != null) { // 獲取操作類型 CanalEntry.EventType eventType = rowChange.getEventType(); // 獲取當前操作所屬的數據庫 String dbName = entry.getHeader().getSchemaName(); // 獲取當前操作所屬的表 String tableName = entry.getHeader().getTableName(); // 事務提交時間 long timestamp = entry.getHeader().getExecuteTime(); for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { dataDetails(rowData.getBeforeColumnsList(), rowData.getAfterColumnsList(), dbName, tableName, eventType, timestamp); System.out.println("-------------------------------------------------------------"); } * 解析具體一條Binlog消息的數據 * * @param dbName 當前操作所屬數據庫名稱 * @param tableName 當前操作所屬表名稱 * @param eventType 當前操作類型(新增、修改、刪除) private static void dataDetails(List<CanalEntry.Column> beforeColumns, List<CanalEntry.Column> afterColumns, String dbName, String tableName, CanalEntry.EventType eventType, long timestamp) { System.out.println("數據庫:" + dbName); System.out.println("表名:" + tableName); System.out.println("操作類型:" + eventType); if (CanalEntry.EventType.INSERT.equals(eventType)) { System.out.println("新增數據:"); printColumn(afterColumns); } else if (CanalEntry.EventType.DELETE.equals(eventType)) { System.out.println("刪除數據:"); printColumn(beforeColumns); } else { System.out.println("更新數據:更新前數據--"); System.out.println("更新數據:更新後數據--"); System.out.println("操作時間:" + timestamp); private static void printColumn(List<CanalEntry.Column> columns) { for (CanalEntry.Column column : columns) { System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated()); * 獲取連接 public CanalConnector getConn() { return CanalConnectors.newSingleConnector(new InetSocketAddress(host, port), instance, username, password); }
測試查看
數據庫修改數據庫時
數據新增數據時
刪除數據(把我們才添加的小明刪掉)
當我們操作監控的數據庫DM L操作的時候呢,會被canal監聽到…我們呢,通過canal監聽,拿到修改的庫,修改的表,修改的字段,便可以根據自己業務進行數據處理瞭!
哎,這個時候啊,可能有小夥伴就要問瞭,那麼,我能不能直接獲取其操作的sql語句呢?
目前,我是自己解析其列來手動拼接的sql語句實現瞭
話不多說,先上效果:
canal 監聽到主庫sql變化—-> update students set id = '2', age = '999', name = '小三', city = '11', date = '2020-09-27 17:41:44', birth = '2020-09-27 18:00:48' where id=2
canal 監聽到主庫sql變化—-> delete from students where id=6
canal 監聽到主庫sql變化—-> insert into students (id,age,name,city,date,birth) VALUES ('89','98','測試新增','深圳','2020-09-27 22:46:53','')
canal 監聽到主庫sql變化—-> update students set id = '89', age = '98', name = '測試新增', city = '深圳', date = '2020-09-27 22:46:53', birth = '2020-09-27 22:46:56' where id=89
實際上呢,我們也就是拿到其執行前列數據變化 執行後列數據變化,自己拼接瞭一個sql罷瞭…附上代碼
package com.leilei; import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.protocol.CanalEntry.*; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.exception.CanalClientException; import com.google.protobuf.InvalidProtocolBufferException; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; import org.springframework.stereotype.Component; import java.net.InetSocketAddress; import java.util.List; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; /** * @author lei * @version 1.0 * @date 2020/9/27 22:33 * @desc 讀取binlog日志 */ @Component public class ReadBinLogToSql implements ApplicationRunner { //讀取的binlog sql 隊列緩存 一邊Push 一邊poll private Queue<String> canalQueue = new ConcurrentLinkedQueue<>(); @Value("${canal.host}") private String host; @Value("${canal.port}") private int port; @Value("${canal.username}") private String username; @Value("${canal.password}") private String password; @Value("${canal.instance}") private String instance; @Override public void run(ApplicationArguments args) throws Exception { CanalConnector conn = getConn(); while (true) { try { conn.connect(); //訂閱實例中所有的數據庫和表 conn.subscribe(".*\\..*"); // 回滾到未進行ack的地方 conn.rollback(); // 獲取數據 每次獲取一百條改變數據 Message message = conn.getWithoutAck(100); long id = message.getId(); int size = message.getEntries().size(); if (id != -1 && size > 0) { // 數據解析 analysis(message.getEntries()); } else { Thread.sleep(1000); } // 確認消息 conn.ack(message.getId()); } catch (CanalClientException | InvalidProtocolBufferException | InterruptedException e) { e.printStackTrace(); } finally { // 關閉連接 conn.disconnect(); } } } private void analysis(List<Entry> entries) throws InvalidProtocolBufferException { for (Entry entry : entries) { if (EntryType.ROWDATA == entry.getEntryType()) { RowChange rowChange = RowChange.parseFrom(entry.getStoreValue()); EventType eventType = rowChange.getEventType(); if (eventType == EventType.DELETE) { saveDeleteSql(entry); } else if (eventType == EventType.UPDATE) { saveUpdateSql(entry); } else if (eventType == EventType.INSERT) { saveInsertSql(entry); } } } } /** * 保存更新語句 * * @param entry */ private void saveUpdateSql(Entry entry) { try { RowChange rowChange = RowChange.parseFrom(entry.getStoreValue()); List<RowData> dataList = rowChange.getRowDatasList(); for (RowData rowData : dataList) { List<Column> afterColumnsList = rowData.getAfterColumnsList(); StringBuffer sql = new StringBuffer("update " + entry.getHeader().getTableName() + " set "); for (int i = 0; i < afterColumnsList.size(); i++) { sql.append(" ") .append(afterColumnsList.get(i).getName()) .append(" = '").append(afterColumnsList.get(i).getValue()) .append("'"); if (i != afterColumnsList.size() - 1) { sql.append(","); } } sql.append(" where "); List<Column> oldColumnList = rowData.getBeforeColumnsList(); for (Column column : oldColumnList) { if (column.getIsKey()) { sql.append(column.getName()).append("=").append(column.getValue()); break; } } canalQueue.add(sql.toString()); } } catch (InvalidProtocolBufferException e) { e.printStackTrace(); } } /** * 保存刪除語句 * * @param entry */ private void saveDeleteSql(Entry entry) { try { RowChange rowChange = RowChange.parseFrom(entry.getStoreValue()); List<RowData> rowDatasList = rowChange.getRowDatasList(); for (RowData rowData : rowDatasList) { List<Column> columnList = rowData.getBeforeColumnsList(); StringBuffer sql = new StringBuffer("delete from " + entry.getHeader().getTableName() + " where "); for (Column column : columnList) { if (column.getIsKey()) { sql.append(column.getName()).append("=").append(column.getValue()); break; } } canalQueue.add(sql.toString()); } } catch (InvalidProtocolBufferException e) { e.printStackTrace(); } } /** * 保存插入語句 * * @param entry */ private void saveInsertSql(Entry entry) { try { RowChange rowChange = RowChange.parseFrom(entry.getStoreValue()); List<RowData> datasList = rowChange.getRowDatasList(); for (RowData rowData : datasList) { List<Column> columnList = rowData.getAfterColumnsList(); StringBuffer sql = new StringBuffer("insert into " + entry.getHeader().getTableName() + " ("); for (int i = 0; i < columnList.size(); i++) { sql.append(columnList.get(i).getName()); if (i != columnList.size() - 1) { sql.append(","); } } sql.append(") VALUES ("); for (int i = 0; i < columnList.size(); i++) { sql.append("'" + columnList.get(i).getValue() + "'"); if (i != columnList.size() - 1) { sql.append(","); } } sql.append(")"); canalQueue.add(sql.toString()); } } catch (InvalidProtocolBufferException e) { e.printStackTrace(); } } /** * 獲取連接 */ public CanalConnector getConn() { return CanalConnectors.newSingleConnector(new InetSocketAddress(host, port), instance, username, password); } /** * 模擬消費canal轉換的sql語句 */ public void executeQueueSql() { int size = canalQueue.size(); for (int i = 0; i < size; i++) { String sql = canalQueue.poll(); System.out.println("canal 監聽到主庫sql變化----> " + sql); } } }
當然瞭,這隻是簡單的demo 演示,您可根據自己的業務進行修改完善即可…
上邊的安裝步驟呢,我也是不斷的測試過,沒有問題,當然可能或多或少有些坑沒有踩到,但是如果您按照我的步驟來,大概率是一馬平川的…
附上項目源碼:springboot-canal
到此這篇關於Springboot2.3.x整合Canal的文章就介紹到這瞭,更多相關Springboot2.3.x整合Canal內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- MySQL高性能實現Canal數據同步神器
- 關於SpringBoot整合Canal數據同步的問題
- Canal監聽MySQL的實現步驟
- Spring Boot + Canal 實現數據庫實時監控
- 淺談MySQL與redis緩存的同步方案