SpringBoot系列之集成阿里canal監聽MySQL Binlog

1、什么是阿里canal？
canal是阿里開源的，對數據庫增量日志解析，提供增量數據訂閱和消費的組件。引用官網的圖片，canal的工作原理主要是模擬 MySQL slave 的交互協議，偽裝自己為 MySQL slave，向master發送dump 協議，獲取到數據后，解析 binary log 對象數據。

 

 

2、canal環境搭建
本文基於Window系統。

使用canal需要確保數據庫開啟了binlog：

show variables like'log_%';

如果沒開啟，在mysql my.ini配置文件添加配置，注意文件內存為的時候，注意編碼格式必須為ANSI，不然會編譯報錯

[mysqld] # 開啟 binlog log-bin=mysql-bin # 選擇 ROW 模式 binlog-format=ROW # 配置 MySQL replaction 需要定義，不要和 canal 的 slaveId 重復 server_id=1

 

配置文件修改是否正確，使用命令，查看日志

mysqld --console

 

重啟MySQL實例

net stop mysql
net start mysql

 

binlog開啟后，創建一個canal用戶並授權，官網配置是@%，表示所有服務器，因為本地測試的，所以改為localhost就可以

CREATE USER canal IDENTIFIED BY 'canal'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%' identified by 'canal'; FLUSH PRIVILEGES;

 

下載canal服務端，到官網releases下載對應資料，canal.deployer-1.1.5.tar.gz是服務端，解壓后在conf文件夾里找到\example\instance.properties，修改數據庫配置信息，dbUsername，dbPassword數據庫賬號密碼

# position info(master數據庫配置) canal.instance.master.address=gz-cdb-l5ixwzm1.sql.tencentcdb.com:59039 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(用戶名密碼) canal.instance.dbUsername=canal canal.instance.dbPassword=canal1uo#A#9R # 加上默認數據庫 canal.instance.defaultDatabaseName=tajiax_canal canal.instance.connectionCharset = UTF-8 # enable druid Decrypt database password canal.instance.enableDruid=false #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==

 

查看canal實例名稱，文件在D:\dev\canal.deployer-1.1.5\conf\canal.properties：

canal.destination=example

到canal服務器安裝目錄D:\dev\canal.deployer-1.1.5\bin，找到startup.bat執行。

在碼雲有示例->\ly\canal-application

3、Canal客戶端測試
JDK 1.8
SpringBoot2.2.1
Maven 3.2+
開發工具
IntelliJ IDEA
smartGit

3.1、引入依賴

<dependency>
    <groupId>top.javatool</groupId>
    <artifactId>canal-spring-boot-starter</artifactId>
    <version>1.2.1-RELEASE</version>
</dependency>

 

3.2、配置canal
# canal實例名稱，要跟canal-server運行時設置的destination一致

# D:\dev\canal.deployer-1.1.5\conf\canal.properties canal.destination=example # canal默認監聽端口 canal.server=127.0.0.1:11111 logging.level.top.javatool.canal.client=warn

 

3.3、編寫監聽器，監聽Canal消息

package com.lynch.entity; import org.springframework.stereotype.Component; import top.javatool.canal.client.annotation.CanalTable; import top.javatool.canal.client.handler.EntryHandler; @CanalTable("tb_user") @Component public class UserHandler implements EntryHandler<UserEntity> { // @Autowired // private RedisHandler redisHandler; // @Autowired // private Cache<Long, Item> itemCache;
 @Override public void insert(UserEntity item) { System.out.println("insert," + item); // 寫數據到JVM進程緩存 //itemCache.put(item.getId(), item); // 寫數據到redis //redisHandler.saveItem(item);
 } @Override public void update(UserEntity before, UserEntity after) { System.out.println("update before," + before); System.out.println("update after," + after); // 寫數據到JVM進程緩存 //itemCache.put(after.getId(), after); // 寫數據到redis //redisHandler.saveItem(after);
 } @Override public void delete(UserEntity item) { System.out.println("delete," + item); // 刪除數據到JVM進程緩存 //itemCache.invalidate(item.getId()); // 刪除數據到redis //redisHandler.deleteItemById(item.getId());
 } }

當表tb_user進行insert、update、delete操作時，會發現監控已生效。

注意：更改canal實例名稱為canaltest
1、在D:\dev\canal.deployer-1.1.5\conf目錄下，把example文件夾改成canaltest
2、在D:\dev\canal.deployer-1.1.5\conf\canal.properties文件中，找到canal.destination配置項改為canaltest
3、修改canal配置
canal.destination=canaltest