思維導圖
前言
我們都知道一個系統最重要的是數據，數據是保存在數據庫里。但是很多時候不單止要保存在數據庫中，還要同步保存到Elastic Search、HBase、Redis等等。
這時我注意到阿里開源的框架Canal，他可以很方便地同步數據庫的增量數據到其他的存儲應用。所以在這里總結一下，分享給各位讀者參考~
一、什么是canal
我們先看官網的介紹
canal，譯意為水道/管道/溝渠，主要用途是基於 MySQL 數據庫增量日志解析，提供增量數據訂閱和消費。
這句介紹有幾個關鍵字：增量日志，增量數據訂閱和消費。
這里我們可以簡單地把canal理解為一個用來同步增量數據的一個工具。
接下來我們看一張官網提供的示意圖：
canal的工作原理就是把自己偽裝成MySQL slave，模擬MySQL slave的交互協議向MySQL Mater發送 dump協議，MySQL mater收到canal發送過來的dump請求，開始推送binary log給canal，然后canal解析binary log，再發送到存儲目的地，比如MySQL，Kafka，Elastic Search等等。
二、canal能做什么
以下參考canal官網。
與其問canal能做什么，不如說數據同步有什么作用。
但是canal的數據同步不是全量的，而是增量。基於binary log增量訂閱和消費，canal可以做：
數據庫鏡像
數據庫實時備份
索引構建和實時維護
業務cache(緩存)刷新
帶業務邏輯的增量數據處理
三、如何搭建canal
3.1 首先有一個MySQL服務器
當前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x
我的Linux服務器安裝的MySQL服務器是5.7版本。
MySQL的安裝這里就不演示了，比較簡單，網上也有很多教程。
然后在MySQL中需要創建一個用戶，並授權：
-- 使用命令登錄：mysql -u root -p
-- 創建用戶 用戶名：canal 密碼：Canal@123456
create user 'canal'@'%' identified by 'Canal@123456';
-- 授權 *.*表示所有庫
grant SELECT, REPLICATION SLAVE, REPLICATION CLIENT on *.* to 'canal'@'%' identified by 'Canal@123456';
下一步在MySQL配置文件my.cnf設置如下信息：（詳見上一篇binlog配置）
[mysqld]
# 打開binlog
log-bin=mysql-bin
# 選擇ROW(行)模式
binlog-format=ROW
# 配置MySQL replaction需要定義，不要和canal的slaveId重復
server_id=1
改了配置文件之后，重啟MySQL，使用命令查看是否打開binlog模式：
查看binlog日志文件列表：
查看當前正在寫入的binlog文件：
MySQL服務器這邊就搞定了，很簡單。
3.2 安裝canal
去官網下載頁面進行下載：https://github.com/alibaba/canal/releases
我這里下載的是1.1.4的版本：
解壓canal.deployer-1.1.4.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=127.0.0.1:3306
# binlog日志名稱
canal.instance.master.journal.name=mysql-bin.000001
# 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=Canal@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=
我這里用的是win10系統，所以在bin目錄下找到startup.bat啟動：
啟動就報錯，坑呀：
要修改一下啟動的腳本startup.bat：
然后再啟動腳本：
這就啟動成功了。
Java客戶端操作
首先引入maven依賴：
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.1.4</version>
</dependency>
然后創建一個canal項目，使用SpringBoot構建，如圖所示：
在CannalClient類使用Spring Bean的生命周期函數afterPropertiesSet()：
@Component
public class CannalClient implements InitializingBean {
private final static int BATCH_SIZE = 1000;
@Override
public void afterPropertiesSet() throws Exception {
// 創建鏈接
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1", 11111), "example", "", "");
try {
//打開連接
connector.connect();
//訂閱數據庫表,全部表
connector.subscribe(".*\\..*");
//回滾到未進行ack的地方，下次fetch的時候，可以從最后一個沒有ack的地方開始拿
connector.rollback();
while (true) {
// 獲取指定數量的數據
Message message = connector.getWithoutAck(BATCH_SIZE);
//獲取批量ID
long batchId = message.getId();
//獲取批量的數量
int size = message.getEntries().size();
//如果沒有數據
if (batchId == -1 || size == 0) {
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() == EntryType.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<Column> columns) {
for (Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
}
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
以上就完成了Java客戶端的代碼。這里不做具體的處理，僅僅是打印，先有個直觀的感受。
最后我們開始測試，首先啟動MySQL、Canal Server，還有剛剛寫的Spring Boot項目。然后創建表：
CREATE TABLE `tb_commodity_info` (
`id` varchar(32) NOT NULL,
`commodity_name` varchar(512) DEFAULT NULL COMMENT '商品名稱',
`commodity_price` varchar(36) DEFAULT '0' COMMENT '商品價格',
`number` int(10) DEFAULT '0' COMMENT '商品數量',
`description` varchar(2048) DEFAULT '' COMMENT '商品描述',
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='商品信息表';
1
2
3
4
5
6
7
8
然后我們在控制台就可以看到如下信息：
如果新增一條數據到表中：
INSERT INTO tb_commodity_info VALUES('3e71a81fd80711eaaed600163e046cc3','叉燒包','3.99',3,'又大又香的叉燒包，老人小孩都喜歡');
1
控制台可以看到如下信息：
總結
canal的好處在於對業務代碼沒有侵入，因為是基於監聽binlog日志去進行同步數據的。實時性也能做到准實時，其實是很多企業一種比較常見的數據同步的方案。
通過上面的學習之后，我們應該都明白canal是什么，它的原理，還有用法。實際上這僅僅只是入門，因為實際項目中我們不是這樣玩的…
實際項目我們是配置MQ模式，配合RocketMQ或者Kafka，canal會把數據發送到MQ的topic中，然后通過消息隊列的消費者進行處理。
Canal的部署也是支持集群的，需要配合ZooKeeper進行集群管理。
Canal還有一個簡單的Web管理界面。
下一篇就講一下集群部署Canal，配合使用Kafka，同步數據到Redis。
參考資料：Canal官網
絮叨
上面所有例子的代碼都上傳Github了：
https://github.com/yehongzhi/mall
如果你覺得這篇文章對你有用，就點贊，關注，評論吧~
你的三連是我創作的最大動力~
拒絕做一條咸魚，我是一個努力讓大家記住的程序員。我們下期再見！！！
能力有限，如果有什么錯誤或者不當之處，請大家批評指正，一起學習交流！
————————————————
版權聲明：本文為CSDN博主「java技術愛好者_R」的原創文章，遵循CC 4.0 BY-SA版權協議，轉載請附上原文出處鏈接及本聲明。
原文鏈接：https://blog.csdn.net/yehongzhi1994/article/details/107880162