Canal 和 MySQL 配置指南

1. 环境依赖

在开始配置之前，请确保已安装并配置以下环境：

Docker：用于运行 MySQL 和 Canal 的容器化服务。
MySQL：确保安装的是支持 binlog 的版本。
Canal：阿里巴巴开源的数据库增量订阅和消费组件。

2. MySQL 配置

为了使 Canal 能够正确监听 MySQL 的 binlog 日志，必须进行如下配置：

2.1 开启 binlog

MySQL 使用 binlog 来记录所有的数据库更改。确保 MySQL 配置文件中开启了 binlog，并将日志格式设置为 ROW。

在 MySQL 的配置文件 my.cnf 或 my.ini 中添加或修改以下内容：

# 开启 binlog
log-bin=mysql-bin

# 设置 binlog 格式为 ROW 模式
binlog-format=ROW

# 为 MySQL 设置 server_id，确保唯一，避免与 Canal 的 slaveId 冲突
server_id=1

2.2 设置 MySQL 用户权限

为 Canal 分配足够的权限以读取 binlog。进入 MySQL 并执行以下命令，为 Canal 用户分配 REPLICATION SLAVE 和 REPLICATION CLIENT 权限：

CREATE USER 'test'@'%' IDENTIFIED BY '123456';

GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'test'@'%';

确保用户名、密码和主机名与 Canal 的配置一致。

3. Canal 配置

3.1 启动 Canal 容器

通过 Docker 启动 Canal 容器：

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

3.2 复制配置文件

为了自定义配置文件，可以将 Canal 容器内的 instance.properties 文件复制到宿主机，便于修改：

docker cp canal:/home/admin/canal-server/conf/example/instance.properties "D:/docker/canal/conf/"

其中"D:/docker/canal/conf/"是绝对路径，因为我发现如果输入 /d/docker/canal/conf/它其实寻找的是 "C:/d/docker/canal/conf/"，所以这里设置绝对路径。

3.3 修改 `instance.properties` 配置文件

打开复制到本地的 instance.properties 文件，进行如下配置修改：

#################################################
## MySQL 配置
canal.instance.mysql.slaveId=1

# MySQL 地址与端口
canal.instance.master.address=172.17.0.3:3306  # MySQL 容器的 IP 地址

# MySQL 用户名与密码
canal.instance.dbUsername=test
canal.instance.dbPassword=123456

# 数据库表过滤规则
canal.instance.filter.regex=.*\\..*  # 订阅所有表

canal.instance.master.address=172.17.0.3:3306启动容器的时候指定自定义网络那么IP就可以换成容器名，如果默认是bridge网络，那么需要指定IP。

3.4 重新启动 Canal 容器并挂载配置文件

修改完配置文件后，重新启动 Canal 容器，并将本地修改过的配置文件挂载到容器内：

docker run --name canal -p 11111:11111 \-v /d/docker/canal/conf/instance.properties:/home/admin/canal-server/conf/example/instance.properties \--network bridge -d canal/canal-server

在这里，bridge 是你设置的 Docker 自定义网络，用于确保 Canal 和 MySQL 容器能通过网络相互通信。

4. Canal 集成到项目

4.1 Maven 依赖

将以下依赖添加到项目的 pom.xml 文件中：

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

<dependency><groupId>com.alibaba.otter</groupId><artifactId>canal.protocol</artifactId><version>1.1.5</version>
</dependency>

4.2 编写 Canal 客户端代码

以下是 Canal 客户端代码的示例，用于连接 Canal 服务并监听 MySQL 数据库的变更：

package com.example.demo;

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 java.net.InetSocketAddress;
import java.util.List;

public class CanalClient {
// Canal 服务地址private static final String CANAL_SERVER = "127.0.0.1";private static final int CANAL_PORT = 11111;private static final String DESTINATION = "example";
public static void main(String[] args) {// 连接 Canal 服务CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(CANAL_SERVER, CANAL_PORT),DESTINATION, "", "");
connector.connect();connector.subscribe(".*\\..*");connector.rollback();
while (true) {// 获取数据Message message = connector.getWithoutAck(100);long batchId = message.getId();if (batchId != -1) {processEntries(message.getEntries());connector.ack(batchId);  // 确认处理成功}}}
private static void processEntries(List<CanalEntry.Entry> entries) {for (CanalEntry.Entry entry : entries) {if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {continue;}try {CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {switch (rowChange.getEventType()) {case INSERT:System.out.println("新增数据: " + rowData.getAfterColumnsList());break;case UPDATE:System.out.println("更新数据: " + rowData.getAfterColumnsList());break;case DELETE:System.out.println("删除数据: " + rowData.getBeforeColumnsList());break;default:break;}}} catch (Exception e) {e.printStackTrace();}}}
}

5. 常见问题及排查

5.1 Canal 无法连接 MySQL

问题：出现 Connection refused 错误。
原因：MySQL 容器的 IP 地址或端口未正确配置，或 MySQL 没有正确暴露端口。
解决方案：检查 MySQL 容器的 IP 地址并确认在 instance.properties 中配置正确。
- 进入到容器里面：docker exec -it canal /bin/bash
- 找到存放日志的文件夹：cd canal-server/logs/example/
- 查看最后一百行日志：tail -100 example.log