1. 工作原理

请参考：https://github.com/alibaba/canal。

下面的内容整理自该文档。

1.1. MySQL 主备工作原理

• MySQL Master 将数据变更写入二进制日志（binary log，其中的记录叫二进制日志事件 binary log events，可以通过 show binlog events 进行查看）
• MySQL Slave 将 Master 的 binary log events 拷贝到它的中继日志（Relay Log）
• MySQL Slave 重放 Relay Log 中的事件，将数据变更反应到它自己的数据副本

1.2. canal 工作原理

• Canal 模拟 MySQL Slave 的交互协议，将自己伪装为 MySQL Slave ，向 MySQL Master 发送 dump 协议
• MySQL Master 收到 dump 请求，开始推送 binary log 给 Slave（即 Canal）
• Canal 解析 binary log 对象（原始数据为 byte 流）

2. 集成 Kafka

Canal 作为 MySQL binlog 增量获取和解析工具，可将变更记录投递到 MQ 系统中，比如 Kafka/RocketMQ，可以借助 MQ 的多语言能力。

Read Also：

1. https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart

2.1. 环境说明

• 操作系统：Ubuntu 18.04.6 LTS (Bionic Beaver)
• JDK：openjdk version "17.0.5" 2022-10-18（canal、zookeeper 都要用）
• Zookeeper：3.5.10（单机版、安装请参考 https://github.com/alibaba/canal/wiki/Zookeeper-QuickStart）
• MySQL：5.7.40-0ubuntu0.18.04.1
• Kafka：3.3.1

2.2. 修改 MySQL 配置

添加如下配置：

server-id               = 1
log_bin                 = /var/log/mysql/mysql-bin.log
binlog_format=ROW

然后重启 MySQL。

添加用户及授权：

mysql> CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';
Query OK, 0 rows affected (0.01 sec)

mysql> GRANT SHOW VIEW, SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
Query OK, 0 rows affected (0.00 sec)

mysql> FLUSH PRIVILEGES;
Query OK, 0 rows affected (0.00 sec)

2.3. 下载

去官网 releases 页面下载 Canal Deployer，本文选择的版本是 canal-1.1.6。将下载后的压缩包解压，然后复制到如下两个目录：

• /usr/local/canal-1
• /usr/local/canal-2

（用来模拟两个 Canal Server Instance）

2.4. 修改配置文件

在以上两个目录中分别执行下面的操作。

1. 修改 conf/example/instance.properties：

canal.instance.master.address=127.0.0.1:3306

canal.instance.dbUsername=canal
canal.instance.dbPassword=canal

# mq config
# 按需更改
canal.mq.topic=test
# dynamic topic route by schema or table regex
#canal.mq.dynamicTopic=mytest1.user,topic2:mytest2\\..*,.*\\..*
canal.mq.partition=0
# hash partition config
#canal.mq.enableDynamicQueuePartition=false
#canal.mq.partitionsNum=3
#canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6
#canal.mq.partitionHash=test.table:id^name,.*\\..*

2. 修改 conf/canal.properties：

xxxxxxxxxx
# 将第二个目录的该配置修改为 21111
canal.port = 11111
# 将第二个目录的该配置修改为 21112
canal.metrics.pull.port = 11112

# 将第二个目录的该配置修改为 21110
canal.admin.port = 11110

# zookeeper 地址
canal.zkServers = 127.0.0.1:2181

canal.serverMode = kafka

canal.destinations = example

# 配置 HA 必须修改这两个配置
#canal.instance.global.spring.xml = classpath:spring/file-instance.xml
canal.instance.global.spring.xml = classpath:spring/default-instance.xml

# 是否为 flat json 格式对象
canal.mq.flatMessage = true

# Kafka Bootstrap Servers 地址
kafka.bootstrap.servers = 10.2.14.27:9092

2.5. 启动

x
/usr/local/canal-1/bin/startup.sh

日志在：/usr/local/canal-1/logs/canal/{canal,canal_stdout}.log，在我的测试环境中启动发生错误：

xxxxxxxxxx
Unrecognized VM option 'AggressiveOpts'
Error: Could not create the Java Virtual Machine.
Error: A fatal exception has occurred. Program will exit.

解决方法是将 startup.sh 中的

xxxxxxxxxx
JAVA_OPTS="$JAVA_OPTS -Xss256k -XX:+AggressiveOpts -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"

改为：

xxxxxxxxxx
JAVA_OPTS="$JAVA_OPTS -Xss256k -XX:-UseBiasedLocking -XX:-OmitStackTraceInFastThrow -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=$base/logs"

然后启动：

xxxxxxxxxx
/usr/local/canal-1/bin/startup.sh
/usr/local/canal-2/bin/startup.sh

说明：

1. 使用 bin/stop.sh 脚本关闭 Canal

2.6. 测试

• 连接到 MySQL 创建 Database、Table
• INSERT、UPDATE、DELETE 若干条记录，观察 Kafka Topic 中的 Record
• 交替启动、关闭用于测试的两个 Canal Server Instance，验证集群的高可用性

JSON 格式的 binlog 示例如下：

x
{
  "data": [{
    "id": "2",
    "name": "Tim"
  }],
  "database": "test",
  "es": 1668579597000,
  "id": 8,
  "isDdl": false,
  "mysqlType": {
    "id": "int",
    "name": "varchar(100)"
  },
  "old": [{
    "id": "1"
  }],
  "pkNames": null,
  "sql": "",
  "sqlType": {
    "id": 4,
    "name": 12
  },
  "table": "test",
  "ts": 1668579597617,
  "type": "UPDATE"
}

2.7. Binlog Position 存储位置

在集群模式下，存储在 Zookeeper 中：

[zk: localhost:2181(CONNECTED) 1] get /otter/canal/destinations/example/1001/cursor
{"@type":"com.alibaba.otter.canal.protocol.position.LogPosition","identity":{"slaveId":-1,"sourceAddress":{"address":"localhost","port":3306}},"postion":{"gtid":"","included":false,"journalName":"mysql-bin.000001","position":11913,"serverId":1,"timestamp":1668579597000}}

在单点模式下，存储在 conf/example/meta.dat 文件中：

xxxxxxxxxx
{"clientDatas":[{"clientIdentity":{"clientId":1001,"destination":"example","filter":""},"cursor":{"identity":{"slaveId":-1,"sourceAddress":{"address":"localhost","port":3306}},"postion":{"gtid":"","included":false,"journalName":"mysql-bin.000001","position":15988,"serverId":1,"timestamp":1668580330000}}}],"destination":"example"}

3. MQ 顺序性问题

请参考：https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart#mq%E9%A1%BA%E5%BA%8F%E6%80%A7%E9%97%AE%E9%A2%98。

Appendix 1 - 创建虚拟机的 Vagrantfile

Vagrant.configure("2") do |config|
    config.vm.box = "generic/ubuntu1804"

    vms = Array(101..101)
    vms.each do |seq|
        config.vm.define :"canal-#{seq}" do |vagrant|
            vagrant.vm.hostname = "canal-#{seq}"
            vagrant.vm.network "private_network", ip: "192.168.56.#{seq}"
            vagrant.vm.provider "virtualbox" do |vb|
                vb.customize ["modifyvm", :id, "--name", "canal-#{seq}"]
                vb.gui = false
                vb.memory = "4096"
                vb.cpus = "8"
            end
        end
    end
end