MySQL高性能實現Canal數據同步神器
簡介
Canal是阿裡巴巴基於Java開源的數據同步工具。平時業務場景使用比較多的如下:
同步數據到ES、Redis緩存中。
數據同步。
業務需要監聽數據。
圖片來源阿裡巴巴github倉庫:https://github.com/alibaba/canal
配置MySQL
跟配置主從復制類似。
數據執行如下命令:
-- 給canal單獨創建用戶 用戶名:canal 密碼:123456 create user 'canal'@'%' identified by '123456'; -- 授權隻能查詢 *.*表示所有庫 grant SELECT, REPLICATION SLAVE, REPLICATION CLIENT on *.* to 'canal'@'%' identified by '123456';
修改數據庫配置文件:
vim /etc/mysql/conf.d/my.cnf
my.cnf內容如下:
[mysqld] # 服務器id(一般都配置成服務器IP) server_id=156 # 開啟日志文件 log-bin=mysql-bin # 選擇ROW(行)模式 binlog-format=ROW
重啟MySQL
查詢日志文件名和偏移量
SHOW MASTER STATUS;
centos7 安裝 canal
下載地址:https://github.com/alibaba/canal/releases/tag/canal-1.1.5
命令行下載
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz
解壓
tar -xvf canal.deployer-1.1.5.tar.gz
編輯 conf/example/instance.properties文件
## mysql serverId , v1.0.26+ will autoGen ## v1.0.26版本後會自動生成slaveId,所以可以不用配置 # canal.instance.mysql.slaveId=0 # 數據庫地址 canal.instance.master.address=192.168.126.156:3306 # binlog日志名稱 canal.instance.master.journal.name=mysql-bin.000007 # mysql主庫鏈接時起始的binlog偏移量 canal.instance.master.position=154 # mysql主庫鏈接時起始的binlog的時間戳 canal.instance.master.timestamp= canal.instance.master.gtid= # username/password # 在MySQL服務器授權的賬號密碼 canal.instance.dbUsername=canal canal.instance.dbPassword=123456 # 字符集 canal.instance.connectionCharset = UTF-8 # enable druid Decrypt database password canal.instance.enableDruid=false # table regex .*\\..*表示監聽所有表 也可以寫具體的表名,用,隔開 canal.instance.filter.regex=.*\\..* # mysql 數據解析表的黑名單,多個表用,隔開 canal.instance.filter.black.regex=
啟動
./bin/startup.sh
jps 查看進程
[root@localhost canal]# jps 7472 Jps 5371 CanalLauncher
java客戶端
pom.xml 依賴
<dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.1.4</version> </dependency>
客戶端代碼
package com.terry.test; 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 static com.alibaba.otter.canal.protocol.CanalEntry.*; import static com.alibaba.otter.canal.protocol.CanalEntry.EntryType.TRANSACTIONBEGIN; import java.net.InetSocketAddress; import java.util.List; /** * canal 測試 * @author terry * @version 1.0 * @date 2022/5/14 22:19 */ public class CanalTest { public static void main(String[] args) { // 創建鏈接 CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.126.156", 11111), "example", "", ""); try { //打開連接 connector.connect(); //訂閱數據庫表,全部表 connector.subscribe(".*\\..*"); //回滾到未進行ack的地方,下次fetch的時候,可以從最後一個沒有ack的地方開始拿 connector.rollback(); while (true) { // 獲取指定數量的數據 Message message = connector.getWithoutAck(100); //獲取批量ID long batchId = message.getId(); //獲取批量的數量 int size = message.getEntries().size(); //如果沒有數據 if (batchId == -1 || size == 0) { System.out.println("等待數據中..."); try { //線程休眠2秒 Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } } else { //如果有數據,處理數據 printEntry(message.getEntries()); } //進行 batch id 的確認。確認之後,小於等於此 batchId 的 Message 都會被確認。 connector.ack(batchId); } } catch (Exception e) { e.printStackTrace(); } finally { connector.disconnect(); } } /** * 打印canal server解析binlog獲得的實體類信息 */ private static void printEntry(List<Entry> entrys) { for (Entry entry : entrys) { if (entry.getEntryType() == TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) { //開啟/關閉事務的實體類型,跳過 continue; } //RowChange對象,包含瞭一行數據變化的所有特征 //比如isDdl 是否是ddl變更操作 sql 具體的ddl sql beforeColumns afterColumns 變更前後的數據字段等等 RowChange rowChage; try { rowChage = RowChange.parseFrom(entry.getStoreValue()); } catch (Exception e) { throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e); } //獲取操作類型:insert/update/delete類型 EventType eventType = rowChage.getEventType(); //打印Header信息 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)); //判斷是否是DDL語句 if (rowChage.getIsDdl()) { System.out.println("================》;isDdl: true,sql:" + rowChage.getSql()); } //獲取RowChange對象裡的每一行數據,打印出來 for (RowData rowData : rowChage.getRowDatasList()) { //如果是刪除語句 if (eventType == EventType.DELETE) { printColumn(rowData.getBeforeColumnsList()); //如果是新增語句 } else if (eventType == EventType.INSERT) { printColumn(rowData.getAfterColumnsList()); //如果是更新的語句 } else { //變更前的數據 System.out.println("------->; before"); printColumn(rowData.getBeforeColumnsList()); //變更後的數據 System.out.println("------->; after"); printColumn(rowData.getAfterColumnsList()); } } } } private static void printColumn(List<CanalEntry.Column> columns) { for (CanalEntry.Column column : columns) { System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated()); } } }
日志打印輸出,當數據庫更改後就會接收到消息。
等待數據中…
等待數據中…
================》; binlog[mysql-bin.000007:345] , name[test,t_user_0] , eventType : INSERT
id : 5 update=true
name : 王五 update=true
等待數據中…
等待數據中…
參考文獻:
https://blog.csdn.net/yehongzhi1994/article/details/107880162
https://github.com/alibaba/canal
到此這篇關於MySQL高性能實現Canal數據同步神器的文章就介紹到這瞭,更多相關MySQL Canal數據同步內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!
推薦閱讀:
- Canal監聽MySQL的實現步驟
- 關於SpringBoot整合Canal數據同步的問題
- Springboot2.3.x整合Canal的示例代碼
- Spring Boot + Canal 實現數據庫實時監控
- 淺談MySQL與redis緩存的同步方案