Canal 介绍和使用指南

Canal 是阿里开源的一款 MySQL 数据库增量日志解析工具，提供增量数据订阅和消费。使用 Canal 能够实现异步更新数据，配合 MQ 使用可在很多业务场景下发挥巨大作用。

Canal 简介

canal 是阿里开源的一款 MySQL 数据库增量日志解析工具，提供增量数据订阅和消费。

使用 Canal 能够实现异步更新数据，配合 MQ 使用可在很多业务场景下发挥巨大作用。

工作原理

MySQL 主备复制原理

MySQL master 将数据变更写入二进制日志(binary log)，日志中的记录叫做二进制日志事件(binary log events，可以通过 show binlog events 进行查看)
MySQL slave 将 master 的 binary log events 拷贝到它的中继日志(relay log)
MySQL slave 重放 relay log 中事件，将数据变更反映刀它自己的数据

图片来源：https://avisheksharma.wordpress.com/2015/01/07/step-wise-guide-to-setup-mysql-replication/

Canal 工作原理

Canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave，向 MySQL master 发送 dump 协议
MySQL master 收到 dump 请求，开始推送 binary log 给 slave(即 Canal)
Canal 解析 binary log 对象(原始为 byte 流)

环境准备

你应该事先准备好一个 MySQL 环境，并按以下步骤进行设置。

开启 binlog

需要先开启 MySQL 的 binlog 写入功能，配置binlog-format为ROW模式，具体my.conf中配置如下：

ini

[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

修改配置文件之后，重启 MySQL。

使用命令查看是否打开 binlog 模式，如输出以下内容则说明 binlog 已开启。

mysql> show variables like 'log_bin';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| log_bin       | ON    |
+---------------+-------+

查看binlog_format配置是否正确。

mysql> show variables like 'binlog_format';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| binlog_format | ROW   |
+---------------+-------+

添加授权

Canal 的原理是模拟自己为 MySQL slave，所以一定要为账号授予作为 MySQL slave 的相关权限。

下面的命令是先创建一个名为canal的账号，再对其进行授权，如果已有帐户可直接 grant。

sql

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

安装 Canal

打开官方 release 页面，根据需要选择对应的软件包下载即可。

将下载后的软件包解压，可看到以下目录。

log

bin
conf
lib
logs

修改配置文件：canal-server/conf/example/instance.properties。

将canal.instance.master.address修改为你的 MySQL 地址。

将canal.instance.tsdb.dbUsername修改为你上面授权的账号。

将canal.instance.tsdb.dbPassword修改为你上面授权账号的密码。

配置示例如下：

ini

#################################################
## mysql serverId , v1.0.26+ will autoGen
# canal.instance.mysql.slaveId=0

# enable gtid use true/false
canal.instance.gtidon=false

# position info
canal.instance.master.address=127.0.0.1:3306
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

当然我还是推荐开发和测试阶段使用 Docker 搭建环境。

执行以下命令，拉取canal-server最新镜像。

docker pull canal/canal-server:latest

如果因网络问题无法直接拉取 Docker 镜像，也可以选择 clone 代码到本地编译。

git clone git@github.com:alibaba/canal.git
cd canal/docker && sh build.sh

启动容器。

docker run -d --name canal-server -p 11111:11111 canal/canal-server

进入容器。

docker exec -it canal-server /bin/bash

修改配置。

vi canal-server/conf/example/instance.properties

注意：如果是 macOS 平台容器内使用host.docker.internal表示localhost。
即：canal.instance.master.address=host.docker.internal:3306

修改完配置后重启容器。

docker container restart canal-server

Canal Client

Canal 特别设计了 Client-Server 模式，交互协议使用 protobuf v3，Client 端可采用不同语言实现不同的消费逻辑。

启动 Canal Server 之后，我们可以使用 Canal 客户端连接 Canal 进行消费，本文以 Go 客户端canal-go为例，演示如何从 canal-server 消费数据。

package main

import (
	"fmt"
	"time"

	pbe "github.com/Q1mi/canal-go/protocol/entry"

	"github.com/Q1mi/canal-go/client"
	"google.golang.org/protobuf/proto"
)

// canal-go client demo

func main() {
	// 连接 canal-server
	// 请修改为你的 canal-server 配置
	connector := client.NewSimpleCanalConnector(
		"127.0.0.1", 11111, "", "", "example", 60000, 60*60*1000)
	err := connector.Connect()
	if err != nil {
		panic(err)
	}

	// mysql 数据解析关注的表，Perl 正则表达式.
	err = connector.Subscribe(".*\\..*")
	if err != nil {
		fmt.Printf("connector.Subscribe failed, err:%v\n", err)
		panic(err)
	}

	// 消费消息
	for {
		message, err := connector.Get(100, nil, nil)
		if err != nil {
			fmt.Printf("connector.Get failed, err:%v\n", err)
			continue
		}
		batchId := message.Id
		if batchId == -1 || len(message.Entries) <= 0 {
			time.Sleep(time.Second)
			fmt.Println("===暂无数据===")
			continue
		}
		printEntry(message.Entries)
	}
}

func printEntry(entries []*pbe.Entry) {
	for _, entry := range entries {
		// 忽略事务开启和事务关闭类型
		if entry.GetEntryType() == pbe.EntryType_TRANSACTIONBEGIN ||
			entry.GetEntryType() == pbe.EntryType_TRANSACTIONEND {
			continue
		}
		// RowChange 对象，包含了一行数据变化的所有特征
		rowChange := new(pbe.RowChange)
		// protobuf 解析
		err := proto.Unmarshal(entry.GetStoreValue(), rowChange)
		if err != nil {
			fmt.Printf("proto.Unmarshal failed, err:%v\n", err)
		}
		if rowChange == nil {
			continue
		}
		// 获取并打印 Header 信息
		header := entry.GetHeader()
		fmt.Printf("binlog[%s : %d], name[%s,%s], eventType: %s\n",
			header.GetLogfileName(),
			header.GetLogfileOffset(),
			header.GetSchemaName(),
			header.GetTableName(),
			header.GetEventType(),
		)
		//判断是否为 DDL 语句
		if rowChange.GetIsDdl() {
			fmt.Printf("isDdl:true, sql:%v\n", rowChange.GetSql())
		}

		// 获取操作类型：insert/update/delete 等
		eventType := rowChange.GetEventType()
		for _, rowData := range rowChange.GetRowDatas() {
			if eventType == pbe.EventType_DELETE {
				printColumn(rowData.GetBeforeColumns())
			} else if eventType == pbe.EventType_INSERT || eventType == pbe.EventType_UPDATE {
				printColumn(rowData.GetAfterColumns())
			} else {
				fmt.Println("---before---")
				printColumn(rowData.GetBeforeColumns())
				fmt.Println("---after---")
				printColumn(rowData.GetAfterColumns())
			}
		}
	}
}

func printColumn(columns []*pbe.Column) {
	for _, col := range columns {
		fmt.Printf("%s:%s  updated=%v\n", col.GetName(), col.GetValue(), col.GetUpdated())
	}
}

Canal Kafka/RocketMQ

Canal1.1.1版本之后，默认支持将 Canal Server 接收到的 binlog 数据投递到 MQ，目前默认支持的 MQ 系统有 Kafka、RocketMQ、RabbitMQ、PulsarMQ。

这里以介绍使用 Canal Server 将 binlog 数据投递到 Kafka 为例。

配置

请事先准备好 Kafka 环境，可参考 liwenzhou 之前写的Go 操作 Kafka 之 kafka-go使用 Docker 快速搭建本地 Kafka 环境。

修改 instance 配置

在instance.properties配置文件中设置 MQ 相关配置。

ini

#  按需修改成自己的数据库信息
#################################################
...
canal.instance.master.address=127.0.0.1:3306
# username/password,数据库的用户名和密码
...
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal
...
# mq config
# 设置默认的 topic
canal.mq.topic=example
# 针对库名或者表名发送动态 topic
#canal.mq.dynamicTopic=mytest,.*,mytest.user,mytest\\..*,.*\\..*
canal.mq.partition=0
# hash partition config
#canal.mq.partitionsNum=3
#库名.表名: 唯一主键，多个表之间用逗号分隔
#canal.mq.partitionHash=mytest.person:id,mytest.role:id
#################################################

其中，canal.mq.dynamicTopic配置说明。

Canal 1.1.3版本之后, 支持配置格式为：schema 或 schema.table，多个配置之间使用逗号或分号分隔。

例子1：test\.test 指定匹配的单表，发送到以 test_test 为名字的 topic 上
例子2：.\.. 匹配所有表，则每个表都会发送到各自表名的 topic 上
例子3：test 指定匹配对应的库，一个库的所有表都会发送到库名的 topic 上
例子4：test\..* 指定匹配的表达式，针对匹配的表会发送到各自表名的 topic 上
例子5：test,test1\.test1，指定多个表达式，会将 test 库的表都发送到 test 的 topic 上，test1\.test1的表发送到对应的 test1_test1 topic 上，其余的表发送到默认的 canal.mq.topic 值

为满足更大的灵活性，Canal 还允许对匹配条件的规则指定发送的 topic 名字，配置格式：topicName:schema或topicName:schema.table。

例子1: test:test\.test 指定匹配的单表，发送到以 test 为名字的 topic 上
例子2: test:.\.. 匹配所有表，因为有指定 topic，则每个表都会发送到 test 的 topic 下
例子3: test:test 指定匹配对应的库，一个库的所有表都会发送到 test 的 topic 下
例子4：testA:test\..* 指定匹配的表达式，针对匹配的表会发送到 testA 的 topic 下
例子5：test0:test,test1:test1\.test1，指定多个表达式，会将 test 库的表都发送到 test0的 topic 下，test1\.test1的表发送到对应的 test1的 topic 下，其余的表发送到默认的 canal.mq.topic 值

修改 canal 配置文件

默认配置文件路径为/usr/local/canal/conf/canal.properties

ini

# ...
# 可选项: tcp(默认), kafka,RocketMQ,rabbitmq,pulsarmq
canal.serverMode = kafka
# ...

# 是否为 flat json 格式对象
canal.mq.flatMessage = true
# Canal 的 batch size, 默认50K, 由于 kafka 最大消息体限制请勿超过1M(900K 以下)
canal.mq.canalBatchSize = 50
# Canal get 数据的超时时间, 单位: 毫秒, 空为不限超时
canal.mq.canalGetTimeout = 100
canal.mq.accessChannel = local

...

##################################################
#########                    Kafka                   #############
##################################################
# 此处配置修改为你的 Kafka 环境地址
kafka.bootstrap.servers = 127.0.0.1:9092
kafka.acks = all
kafka.compression.type = none
kafka.batch.size = 16384
kafka.linger.ms = 1
kafka.max.request.size = 1048576
kafka.buffer.memory = 33554432
kafka.max.in.flight.requests.per.connection = 1
kafka.retries = 0

kafka.kerberos.enable = false
kafka.kerberos.krb5.file = ../conf/kerberos/krb5.conf
kafka.kerberos.jaas.file = ../conf/kerberos/jaas.conf

# sasl demo
# kafka.sasl.jaas.config = org.apache.kafka.common.security.scram.ScramLoginModule required \\n username=\"alice\" \\npassword="alice-secret\";
# kafka.sasl.mechanism = SCRAM-SHA-512
# kafka.security.protocol = SASL_PLAINTEXT

更多详细内容请查看Canal-Kafka-RocketMQ-QuickStart。

按上述修改 Canal 配置后，重启 Canal 服务即可。

其他

此外，Canal 也支持HA 模式和多节点 MySQL，Canal1.14+版本还提供了 Admin 管理界面，读者可以根据Canal-Admin-QuickStart自行尝试。

参考资料

Canal QuickStart

Canal Docker-QuickStart

Canal AdminGuide

Canal 介绍和使用指南 ​

Canal 简介 ​

工作原理 ​

MySQL 主备复制原理 ​

Canal 工作原理 ​

环境准备 ​

开启 binlog ​

添加授权 ​

安装 Canal ​

Canal Client ​

Canal Kafka/RocketMQ ​

配置 ​

修改 instance 配置 ​

修改 canal 配置文件 ​

其他 ​

参考资料 ​