本文轉載自微信公眾號「微觀技術」，作者Tom哥  。轉載本文請聯系微觀技術公眾號。

 大家好，我是Tom哥~

MySQL 數據庫大家一定都不陌生，今天跟大家聊聊數據同步的事

關于數據同步，我們常見的策略就是 同步雙寫、異步消息

1、同步雙寫：字面意思，同步+雙寫。比如老庫模型重構，數據遷移到新庫，遷移過程中，如果有數據變更，既要寫到老庫，也要寫到新庫，兩邊同步更新。

• 優點：同步機制，保證了數據的實效性。
• 缺點：額外增加同步處理邏輯，會有性能損耗

2、異步消息：如果依賴方過多，我們通常是將變更數據異構發送到MQ消息系統，感興趣的業務可以訂閱消息Topic，拉取消息，然后按自己的業務邏輯處理。

• 優點：架構解耦，可以采用異步來做，降低主鏈路的性能損耗。如果是多個消費方，不會出現指數性能疊加
• 缺點：異步機制，無法滿足實時性，有一定延遲。只能達到最終一致性。

上面兩種方案，都是采用硬編碼，那么有沒有通用的技術方案。不關心你是什么業務，寫入什么數據，對平臺來講可以抽象成一張張 MySQL 表，直接同步表數據。只有使用方才真正去關心數據內容。

可以參考 MySQL 的主從同步原理，拉取 binlog，只要將里面的數據解析出來即可。

流行的中間件是阿里開源的 Canal，今天我們就來做個技術方案，大概內容如下：

一、Canal 介紹

Canal，譯意為水道/管道/溝渠，主要用途是基于 MySQL 數據庫增量日志解析，提供增量數據訂閱和消費。

Canal 誕生之初是為了解決多個備庫與主庫間數據同步，對主庫造成的壓力。

慢慢的，這個管道被發揚光大，應用場景也越來越多

工作原理很簡單，把自己偽裝成 MySQL 的 slave，模擬 MySQL slave 的交互協議向 MySQL master 發送 dump 請求。

MySQL master 收到canal發送過來的dump請求，開始推送binary log給canal，然后canal解析binlog 日志，再存儲到不同的存儲介質中，比如：MySQL、Kafka、Elastic Search、Pulsar 等

業務場景：

• 數據庫實時備份
• ES 數據索引的構建和維護
• 分布式緩存(如：Redis)的同步維護
• 數據異構，訂閱方可以按自己的業務需求訂閱消費，如：Kafka、Pulsar 等

二、安裝 MySQL

1、拉取 MySQL 鏡像

docker pull mysql:5.7 

2、查看鏡像

docker images 

3、啟動 MySQL 進程

docker run \ 
--name mysql \ 
-p 3306:3306 \ 
-e MYSQL_ROOT_PASSWORD=123456 \ 
-d mysql:5.7 

4、查看進程

[root@iZbp12gqydkgwid86ftoauZ mysql]# docker ps -a 
CONTAINER ID   IMAGE     COMMAND                  CREATED         STATUS         PORTS                               NAMES 
e92827897538   mysql     "docker-entrypoint.s…"   4 seconds ago   Up 2 seconds   0.0.0.0:3306->3306/tcp, 33060/tcp   mysql 

5、進入 MySQL 容器

docker exec -it 167bfa3785f1 /bin/bash 

注意：修改一些配置文件，可能會遇到一些問題，如：

docker容器中使用vi或vim提示bash: vi: command not found的處理方法

因為沒有安裝vi編輯器，可以執行下面命令

apt-get update 
apt-get install vim 

6、常用 MySQL 客戶端命令

# 登陸 mysql 
mysql -uroot -p111111 
 
# 顯示數據庫列表 
show databases; 
 
# 選擇數據庫 
use mysql; 
 
# 顯示所有表 
show tables; 
 
# 顯示表結構 
describe 表名; 
 
其他更多命令： 
https://www.cnblogs.com/bluecobra/archive/2012/01/11/2318922.html 

三、MySQL 相關配置

創建一個 MySQL 用戶，用戶名：tom ，密碼：123456

create user 'tom'@'%' identified by '123456'; 

為用戶：tom 授予所有庫的讀寫權限

grant SELECT, REPLICATION SLAVE, REPLICATION CLIENT on *.* to 'tom'@'%' identified by '123456'; 

修改 MySQL 配置文件 my.cnf，位置：/etc/my.cnf

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

注意：需要重啟MySQL容器實例，執行命令 docker restart mysql

查看binlog模式：

查看binlog日志文件列表：

查看當前正在寫入的binlog文件：

四、安裝 Canal

1、從官網下載安裝包

下載地址：

https://github.com/alibaba/canal/releases

本文實驗用的是最新版本 v1.1.5，主要是對不同的客戶端的個性化支持，屬于生態擴展。

其他更多特性，大家可以去官網查看

解壓 tar.gz 壓縮包

tar -zxvf canal.deployer-1.1.5.tar.gz  

打開配置文件 conf/example/instance.properties，修改配置如下：

## 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=156 
# mysql主庫鏈接時起始的binlog的時間戳 
canal.instance.master.timestamp= 
canal.instance.master.gtid= 
 
# username/password 
# 在MySQL服務器授權的賬號密碼 
canal.instance.dbUsername=root 
canal.instance.dbPassword=111111 
# 字符集 
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= 

啟動命令

./startup.sh 

由于采用的阿里云的 ECS 服務器，發現沒有安裝 JAVA 環境。

Oracle 官網下載 JDK 8 的安裝包

下載地址：

https://www.oracle.com/java/technologies/downloads/#java8

然后，通過下面的命令將安裝包上傳到 ECS 服務器

scp jdk-8u311-linux-x64.tar.gz root@118.31.168.234:/root/java      //上傳文件 

安裝 JDK 8 環境

文檔：https://developer.aliyun.com/article/701864

五、啟動 Canal

進入 canal.deployer-1.1.5/bin

執行啟動腳本：

./startup.sh  

進入 canal.deployer-1.1.5/logs/example

如果 example.log 日志文件中，出現下面的內容，表示啟動成功

2022-01-03 08:23:10.165 [canal-instance-scan-0] INFO  c.a.otter.canal.instance.core.AbstractCanalInstance - stop CannalInstance for null-example  
2022-01-03 08:23:10.177 [canal-instance-scan-0] INFO  c.a.otter.canal.instance.core.AbstractCanalInstance - stop successful.... 
2022-01-03 08:23:10.298 [canal-instance-scan-0] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example  
2022-01-03 08:23:10.298 [canal-instance-scan-0] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table filter : ^.*\..*$ 
2022-01-03 08:23:10.298 [canal-instance-scan-0] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table black filter : ^mysql\.slave_.*$ 
2022-01-03 08:23:10.299 [canal-instance-scan-0] INFO  c.a.otter.canal.instance.core.AbstractCanalInstance - start successful.... 

六、工程實驗

創建一個 SpringBoot 工程，spring-boot-bulking-canal

引入相關pom依賴

<dependency> 
    <groupId>com.alibaba.otter</groupId> 
    <artifactId>canal.client</artifactId> 
    <version>1.1.4</version> 
</dependency> 

 

編寫java類，與 canal 服務端 建立連接，拉取數據庫的變更數據

// 創建鏈接 
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); 
        long batchId = message.getId(); 
        printEntry(message.getEntries()); 
        // batch id 提交 
        connector.ack(batchId); 
    } 
} catch (Exception e) { 
    e.printStackTrace(); 
} finally { 
    connector.disconnect(); 
} 

在 ds1 數據庫下創建 MySQL 表

CREATE TABLE `person` ( 
  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '自增主鍵', 
  `income` bigint(20) NOT NULL COMMENT '收入', 
  `expend` bigint(20) NOT NULL COMMENT '支出', 
  PRIMARY KEY (`id`), 
  KEY `idx_income` (`income`) 
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='個人收支表'; 

插入一條記錄：

insert into person values(100,1000,1000); 

Java類解析binlog，在控制臺打印變更日志：

binlog[mysql-bin.000002:1946] , table[ds1,person] , eventType : INSERT 
id : 100    update=true 
income : 1000    update=true 
expend : 1000    update=true 

對 id=100 記錄做修改：

update person set income=2000, expend=2000 where id=100; 

控制臺打印變更日志：

binlog[mysql-bin.000002:2252] , table[ds1,person] , eventType : UPDATE 
------->; before 
id : 100    update=false 
income : 1000    update=false 
expend : 1000    update=false 
------->; after 
id : 100    update=false 
income : 2000    update=true 
expend : 2000    update=true