教你如何用阿里canal

什么是canal？

阿里巴巴 B2B 公司，因为业务的特性，卖家主要集中在国内，买家主要集中在国外，所以衍生出了同步杭州和美国异地机房的需求，从 2010 年开始，阿里系公司开始逐步的尝试基于数据库的日志解析，获取增量变更进行同步，由此衍生出了增量订阅&消费的业务。

Canal 是用 Java 开发的基于数据库增量日志解析，提供增量数据订阅&消费的中间件。目前。 Canal 主要支持了 MySQL 的 Binlog 解析，解析完成后才利用 Canal Client 来处理获得的相关数据。（数据库同步需要阿里的 Otter 中间件，基于 Canal）。

什么是mysql的binlog？

MySQL 的二进制日志可以说 MySQL 最重要的日志了，它记录了所有的 DDL 和 DML(除了数据查询语句)语句，以事件形式记录，还包含语句所执行的消耗的时间， MySQL 的二进制日志是事务安全型的。

一般来说开启二进制日志大概会有 1%的性能损耗。二进制有两个最重要的使用场景:

其一： MySQL Replication 在 Master 端开启 Binlog， Master 把它的二进制日志传递给 Slaves来达到 Master-Slave 数据一致的目的。

其二：自然就是数据恢复了，通过使用 MySQL Binlog 工具来使恢复数据。二进制日志包括两类文件：二进制日志索引文件（文件名后缀为.index）用于记录所有的二进制文件，二进制日志文件（文件名后缀为.00000*）记录数据库所有的 DDL 和 DML(除了数据查询语句)语句事件。

mysql的主从复制原理

1） Master 主库将改变记录，写到二进制日志(Binary Log)中；

2） Slave 从库向 MySQL Master 发送 dump 协议，将 Master 主库的 binary log events 拷贝到它的中继日志(relay log)；

3） Slave 从库读取并重做中继日志中的事件，将改变的数据同步到自己的数据库。

canal的工作原理

就是把自己伪装成 Slave，假装从 Master 复制数据。

canal的使用场景

1）原始场景：阿里 Otter 中间件的一部分

Otter 是阿里用于进行异地数据库之间的同步框架， Canal 是其中一部分。

2）常见场景 1：更新缓存

3）常见场景 2：抓取业务表的新增变化数据，用于制作实时统计

Canal 的下载和安装

下载并解压 Jar 包

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

到/opt/module/canal 包下

注意： canal 解压后是分散的，我们在指定解压目录的时候需要将 canal 指定上

mkdir /opt/module/canal
tar -zxvf canal.deployer-1.1.2.tar.gz -C /opt/module/canal

修改 canal.properties 的配置

pwd
/opt/module/canal/conf
vim canal.properties
#################################################
######### common argument #############
#################################################
canal.id = 1
canal.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
canal.zkServers =
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, RocketMQ
canal.serverMode = tcp
# flush meta cursor/parse position to file

说明：这个文件是 canal 的基本通用配置， canal 端口号默认就是 11111，修改 canal 的输出 model，默认 tcp，改为输出到 kafka多实例配置如果创建多个实例，通过前面 canal 架构，我们可以知道，一个 canal 服务中可以有多个 instance， conf/下的每一个 example 即是一个实例，每个实例下面都有独立的配置文件。默认只有一个实例 example，如果需要多个实例处理不同的 MySQL 数据的话，直接拷贝出多个 example，并对其重新命名，命名和配置文件中指定的名称一致，然后修改canal.properties 中的 canal.destinations=实例 1，实例 2，实例 3。

#################################################
######### destinations #############
#################################################
canal.destinations = example

修改 instance.properties

我们这里只读取一个 MySQL 数据，所以只有一个实例，这个实例的配置文件在

conf/example 目录下

pwd
/opt/module/canal/conf/example
vim instance.properties

1）配置 MySQL 服务器地址

#################################################
## mysql serverId , v1.0.26+ will autoGen
canal.instance.mysql.slaveId=20
# enable gtid use true/false
canal.instance.gtidon=false
# position info
canal.instance.master.address=test:3306

2）配置连接 MySQL 的用户名和密码，默认就是我们前面授权的 canal

# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
canal.instance.defaultDatabaseName =test
# enable druid Decrypt database password
canal.instance.enableDruid=false

实时监控测试

TCP 模式测试

创建 canal 项目

在 canal 模块中配置 pom.xml

<dependencies>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.1.2</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.4.1</version>
</dependency>
</dependencies>

通用监视类 –CanalClient

1） Canal 封装的数据结构

2）在 canal 模块下创建 app 包，并在包下创建 CanalClient （java 代码）

代码如下：

import com.alibaba.fastjson.JSONObject;
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 com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.Random;、
public class Cana lClient {
public static main(String[] args) throws
InvalidProtocolBufferException {
//1.获取 canal 连接对象
CanalConnector canalConnector =
CanalConnectors.newSingleConnector(new
InetSocketAddress("test"), "example", "", "");
while (true) {
canalConnector.connect();
canalConnector.subscribe("TEST.*");
Message message = canalConnector.get(100);
List<CanalEntry.Entry> entries = message.getEntries();
if (entries.size() <= 0) {
System.out.println("没有数据，休息一会");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
} else {
for (CanalEntry.Entry entry : entries) {
String
entry.getHeader().getTableName();
tableName
=
CanalEntry.EntryType
entry.getEntryType();
entryType
=
if
(CanalEntry.EntryType.ROWDATA.equals(entryType)) {
ByteString storeValue = entry.getStoreValue();
CanalEntry.RowChange
CanalEntry.RowChange.parseFrom(storeValue);
rowChange
=
eventType
=
rowChange.getEventType();
List<CanalEntry.RowData> rowDatasList =
rowChange.getRowDatasList();
for (CanalEntry.RowData rowData : rowDatasList)
{
List<CanalEntry.Column> beforeColumnsList =
rowData.getBeforeColumnsList();
JSONObject beforeData = new JSONObject();
for (CanalEntry.Column column :
beforeColumnsList) {
beforeData.put(column.getName(),
column.getValue());
}
JSONObject afterData = new JSONObject();
List<CanalEntry.Column> afterColumnsList =
rowData.getAfterColumnsList();
for (CanalEntry.Column column :
afterColumnsList) {
afterData.put(column.getName(),
column.getValue());
}
System.out.println("TableName:" + tableName
+
",EventType:" + eventType +
",After:" + beforeData +
",After:" + afterData);
}
}
}
}
}
}
}

Kafka 模式测试

1）修改 canal.properties 中 canal 的输出 model，默认 tcp，改为输出到 kafka

#################################################
######### common argument #############
#################################################
canal.id = 1
canal.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
canal.zkServers =
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, RocketMQ
canal.serverMode = kafka
# flush meta cursor/parse position to file

2）修改 Kafka 集群的地址

##################################################
######### MQ #############
##################################################
canal.mq.servers = test1:9092,test2:9092,test3:9092

3）修改 instance.properties 输出到 Kafka 的主题以及分区数

# mq config
canal.mq.topic=canal_test
canal.mq.partitionsNum=1
# hash partition config
#canal.mq.partition=0
#canal.mq.partitionHash=mytest.person:id,mytest.role:id

注意：默认还是输出到指定 Kafka 主题的一个 kafka 分区，因为多个分区并行可能会打乱 binlog 的顺序，如果要提高并行度，首先设置 kafka 的分区数 >1, 然后设置canal.mq.partitionHash 属性

4）启动 Canal

 cd /opt/module/canal/
bin/startup.sh

5）看到 CanalLauncher 你表示启动成功，同时会创建 canal_test 主题

6）启动 Kafka 消费客户端测试，查看消费情况

bin/kafka-console-consumer.sh --bootstrap-server

7）向 MySQL 中插入数据后查看消费者控制台

插入数据

INSERT INTO user_info VALUES('1001','zhangsan','male'),('1002','lisi','female');

Kafka 消费者控制台

{"data":[{"id":"1001","name":"zhangsan","sex":"male"},{"id":"1002","name":"lisi","sex":"female"}],"database":"test-2021","es":1639360729000,"id":1,"isDdl":false,"mysqlType":{"id":"varchar(255)","name":"varchar(255)","sex":"varchar(255)"},"old":null,"sql":"","sqlType":{"id":12,"name":12,"sex":12},"table":"user_info","ts":1639361038454,"type":"INSERT"}