Canal監聽MySQL的實現步驟
1、Mysql數據庫開啟binlog模式
註意:Mysql容器,此處Mysql版本為5.7
#進入容器 docker exec -it mysql /bin/bash #進入配置目錄 cd /etc/mysql/mysql.conf.d #修改配置文件 vi mysqld.cnf
(1) 修改mysqld.cnf配置文件,添加如下配置:
log-bin=mysql-bin server-id=12345
(2) 創建賬號 用於測試使用,使用root賬號創建用戶並授予權限
create user canal@'%' IDENTIFIED by 'canal'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT,SUPER ON *.* TO 'canal'@'%'; FLUSH PRIVILEGES;
(3) 重啟mysql容器
docker restart mysql
2、Docker下Canal容器安裝
(1)拉取canal鏡像
docker pull docker.io/canal/canal-server
(2)創建Canal容器
docker run -p 11111:11111 --name canal -d docker.io/canal/canal-server
(3)進入容器,修改核心配置canal.properties 和instance.properties,canal.properties 是canal自身的配置,instance.properties是需要同步數據的數據庫連接配置。
#進入容器 docker exec -it canal /bin/bash cd canal-server/conf/ #修改 canal.properties vi canal.properties cd example/ #修改 instance.properties vi instance.properties
修改canal.properties的id,不能和mysql的server-id重復,如下圖:
修改instance.properties,配置數據庫連接地址:
這裡的canal.instance.filter.regex
有多種配置,如下:
可以參考地址如下: https://github.com/alibaba/canal/wiki/AdminGuide
mysql 數據解析關註的表,Perl正則表達式.
多個正則之間以逗號(,)分隔,轉義符需要雙斜杠(\\)
常見例子:
1. 所有表:.* or .*\\..*
2. canal schema下所有表: canal\\..*
3. canal下的以canal打頭的表:canal\\.canal.*
4. canal schema下的一張表:canal.test1
5. 多個規則組合使用:canal\\..*,mysql.test1,mysql.test2 (逗號分隔)
註意:此過濾條件隻針對row模式的數據有效(ps. mixed/statement因為不解析sql,所以無法準確提取tableName進行過濾)
配置完成後,設置開機啟動,並記得重啟canal。
docker update --restart=always canal docker restart canal
3、Canal Client項目搭建
(1)引入依賴
<parent> <artifactId>spring-boot-parent</artifactId> <groupId>org.springframework.boot</groupId> <version>2.1.4.RELEASE</version> </parent> <dependencies> <!--canal依賴--> <dependency> <groupId>com.xpand</groupId> <artifactId>starter-canal</artifactId> <version>0.0.1-SNAPSHOT</version> </dependency> </dependencies>
註意:canal依賴stater在中央倉庫是不存在的,需要手動放進本地倉庫或者你公司裡面的nexus
這裡以放進本地倉庫為例:
- 首先解壓spring-boot-starter-canal-master.zip
- 在spring-boot-starter-canal-master\starter-canal文件夾下執行mvn clean install
- 此時在本地倉庫就會存在jar包
- 引入依賴
<!--canal依賴--> <dependency> <groupId>com.xpand</groupId> <artifactId>starter-canal</artifactId> <version>0.0.1-SNAPSHOT</version> </dependency>
(2)啟動類編寫
@SpringBootApplication @EnableCanalClient public class CanalApplication { public static void main(String[] args) { SpringApplication.run(CanalApplication.class,args); } }
(3)監聽器編寫
@CanalEventListener public class CanalDataEventListener { /*** * 增加數據監聽 * @param eventType * @param rowData */ @InsertListenPoint public void onEventInsert(CanalEntry.EventType eventType, CanalEntry.RowData rowData) { rowData.getAfterColumnsList().forEach((c) -> System.out.println("By--Annotation: " + c.getName() + " :: " + c.getValue())); } /*** * 修改數據監聽 * @param rowData */ @UpdateListenPoint public void onEventUpdate(CanalEntry.RowData rowData) { System.out.println("UpdateListenPoint"); rowData.getAfterColumnsList().forEach((c) -> System.out.println("By--Annotation: " + c.getName() + " :: " + c.getValue())); } /*** * 刪除數據監聽 * @param eventType */ @DeleteListenPoint public void onEventDelete(CanalEntry.EventType eventType) { System.out.println("DeleteListenPoint"); } /*** * 自定義數據修改監聽 * @param eventType * @param rowData */ @ListenPoint(destination = "example", schema = "torlesse_test", table = {"tb_user", "tb_order"}, eventType = CanalEntry.EventType.UPDATE) public void onEventCustomUpdate(CanalEntry.EventType eventType, CanalEntry.RowData rowData) { System.err.println("DeleteListenPoint"); rowData.getAfterColumnsList().forEach((c) -> System.out.println("By--Annotation: " + c.getName() + " :: " + c.getValue())); } @ListenPoint(destination = "example", schema = "test_canal", //所要監聽的數據庫名 table = {"tb_user"}, //所要監聽的數據庫表名 eventType = {CanalEntry.EventType.UPDATE, CanalEntry.EventType.INSERT, CanalEntry.EventType.DELETE}) public void onEventCustomUpdateForTbUser(CanalEntry.EventType eventType, CanalEntry.RowData rowData){ getChangeValue(eventType,rowData); } public static void getChangeValue(CanalEntry.EventType eventType, CanalEntry.RowData rowData){ if(eventType == CanalEntry.EventType.DELETE){ rowData.getBeforeColumnsList().forEach(column -> { //獲取刪除前的數據 System.out.println(column.getName() + " == " + column.getValue()); }); }else { rowData.getBeforeColumnsList().forEach(column -> { //打印改變前的字段名和值 System.out.println(column.getName() + " == " + column.getValue()); }); rowData.getAfterColumnsList().forEach(column -> { //打印改變後的字段名和值 System.out.println(column.getName() + " == " + column.getValue()); }); } } }
到此就可以實現Canal監聽Mysql
項目gitee地址:test-canal
詳情可以查看:
https://github.com/alibaba/canal (阿裡官方)
https://github.com/alibaba/canal/wiki/AdminGuide (阿裡官方)
https://github.com/chenqian56131/spring-boot-starter-canal (自制starter)
到此這篇關於Canal監聽MySQL的實現步驟的文章就介紹到這瞭,更多相關Canal監聽MySQL內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- MySQL高性能實現Canal數據同步神器
- 關於SpringBoot整合Canal數據同步的問題
- Springboot2.3.x整合Canal的示例代碼
- Spring Boot + Canal 實現數據庫實時監控
- 詳解Java SpringAOP切面類