MySQL 双主热备 + Canal Admin + Kafka 集成方案

概述

本方案介绍如何使用 Canal Admin 管理 Canal Server 集群，将 MySQL 双主热备集群的数据变更实时同步到 Kafka 集群，实现高可用的 CDC（Change Data Capture，变更数据捕获）功能。

前置条件

在开始本方案之前，请确保已经完成以下部署：

MySQL 双主热备集群
- 参考：MySQL 高可用故障转移（双主热备）
- 已开启 binlog，格式为 ROW
- 已创建 Canal 用户并授权
Kafka 集群
- 参考：Kafka 虚拟机部署指南
- 已创建用于同步数据的 Topic
Zookeeper 集群（如果使用传统模式）
- 参考：Zookeeper 集群部署指南

方案优势

高可用性：MySQL 双主 + Canal 双实例，任一节点故障不影响数据同步
实时性：基于 MySQL binlog，实现毫秒级数据同步
低侵入：无需修改业务代码，通过 binlog 采集数据
易扩展：Kafka 天然支持多消费者，便于下游扩展
解耦合：生产者和消费者完全解耦，提升系统灵活性
可视化管理：通过 Canal Admin Web 界面管理 Canal Server 集群
动态配置：无需重启 Canal Server，即可添加、修改、删除 Instance
集中管理：统一管理多个 Canal Server 节点

1. 架构设计

1.1 整体架构

1.2 数据流向

MySQL 双主集群 → binlog → Canal Server 集群 → Canal Admin → Kafka 集群 → 消费者

详细流程：

应用通过 VIP (192.168.192.170) 访问 MySQL 主库
MySQL 产生数据变更（INSERT/UPDATE/DELETE）
MySQL 将变更记录到 binlog
双主之间自动同步 binlog
Canal Server 模拟 MySQL Slave，拉取本地 MySQL 的 binlog
Canal Server 解析 binlog，转换为结构化事件
Canal Admin 通过 TCP 协议管理 Canal Server 集群
Canal Admin 配置 Instance 的同步规则和目标
Canal Server 将事件发送到 Kafka 集群
Kafka 集群存储消息并多副本复制
消费者从 Kafka 拉取消息并处理

1.3 高可用机制

组件	高可用方案	故障切换时间
MySQL	双主复制 + Keepalived VIP	< 3 秒
Canal Server	双实例部署，各自连接本地 MySQL	< 5 秒
Canal Admin	单实例部署，故障时可通过 Web 界面快速切换	< 10 秒
Kafka	3 节点集群，KRaft 模式	< 10 秒

2. 部署规划

2.1 服务器规划

| 组件 | 主机名 | IP 地址 | 角色 | 说明 | |------|--------|---------|------| | MySQL 主库 1 | ubuntu-mysql-node1 | 192.168.192.171 | MySQL + Keepalived 主 | MySQL 主库 1 | | MySQL 主库 2 | ubuntu-mysql-node2 | 192.168.192.172 | MySQL + Keepalived 备 | MySQL 主库 2 | | MySQL VIP | - | 192.168.192.170 | 虚拟IP | 应用访问入口 | | Canal Admin | ubuntu-canal-admin | 192.168.192.180 | Canal Admin Web 管理界面 | 管理 Canal Server 集群 | | Canal Server 1 | ubuntu-canal-node1 | 192.168.192.181 | Canal Server | 连接 MySQL 主库 1 | | Canal Server 2 | ubuntu-canal-node2 | 192.168.192.182 | Canal Server | 连接 MySQL 主库 2 | | Kafka 节点 | ubuntu-kafka-1 | 192.168.192.211 | Kafka Broker | Ubuntu 单节点部署 |

2.2 端口规划

组件	端口	协议	说明
MySQL	3306	TCP	MySQL 服务端口
Canal Server	11111	TCP	Canal Server 端口
Canal Admin	8089	TCP	Canal Admin 端口
Canal Admin Manager	11110	TCP	Canal Admin 管理端口
Canal Metrics	11112	TCP	Canal 指标端口
Kafka	9092	TCP	Kafka PLAINTEXT 端口

2.3 MySQL 前置配置

确保 MySQL 双主集群已满足以下配置：

# binlog 配置（两台 MySQL 都需要）
log-bin = /var/lib/mysql/mysql-bin.log
binlog_format = ROW
binlog_row_image = FULL
gtid-mode=on
log-slave-updates=true
enforce-gtid-consistency=true

2.4 创建 Canal 用户

在两台 MySQL 服务器上执行：

-- 登录 MySQL
mysql -hlocalhost -uroot -p

-- 创建 Canal 用户
CREATE USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY 'canal@123456';

-- 授予权限
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';

-- 刷新权限
FLUSH PRIVILEGES;

-- 验证权限
SHOW GRANTS FOR 'canal'@'%';

quit;

2.5 创建测试数据库和表

在主库 1 上执行：

-- 登录 MySQL（通过 VIP）
mysql -h192.168.192.170 -uroot -p

-- 创建测试数据库
CREATE DATABASE IF NOT EXISTS test_db DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;

USE test_db;

-- 创建测试表
CREATE TABLE IF NOT EXISTS user (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    username VARCHAR(50) NOT NULL,
    email VARCHAR(100),
    age INT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    INDEX idx_username (username)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 插入测试数据
INSERT INTO user (username, email, age) VALUES
('user1', 'user1@example.com', 25),
('user2', 'user2@example.com', 30),
('user3', 'user3@example.com', 28);

-- 查看数据
SELECT * FROM user;

quit;

3. Canal Admin 部署

3.1 Canal Admin 介绍

Canal Admin 是 Canal 的 Web 管理界面，提供了可视化的 Canal Server 集群管理功能。使用 Canal Admin 可以：

可视化管理：通过 Web 界面管理 Canal Server 集群
动态配置：无需重启 Canal Server，即可添加、修改、删除 Instance
监控指标：实时查看 Canal Server 的运行状态 and 性能指标
配置版本管理：支持配置版本回滚和对比
集群管理：统一管理多个 Canal Server 节点

3.2 环境准备

在 Canal Admin 服务器（192.168.192.180）上执行：

3.2.1 安装 Java

# 更新软件包列表
sudo apt update

# 安装 OpenJDK 11
sudo apt install -y openjdk-11-jdk

# 验证安装
java -version

# 设置 JAVA_HOME 环境变量
echo 'export JAVA_HOME=/usr/lib/jvm/java-11-openjdk-amd64' | sudo tee -a /etc/environment
source /etc/environment

3.2.2 创建 Canal Admin 用户

# 创建 canal 用户
sudo useradd -m -s /bin/bash canal

# 创建 Canal Admin 安装目录
sudo mkdir -p /opt/canal-admin
sudo chown -R canal:canal /opt/canal-admin

3.3 Canal Admin 安装

在 Canal Admin 服务器（192.168.192.180）上执行：

# 进入安装目录
cd /opt

# 下载 Canal Admin (使用 sudo 确保有权限)
sudo wget https://github.com/alibaba/canal/releases/download/canal-1.1.8/canal.admin-1.1.8.tar.gz

# 解压到指定目录
sudo tar -zxvf canal.admin-1.1.8.tar.gz -C /opt/canal-admin

# 统一赋值权限
sudo chown -R canal:canal /opt/canal-admin

3.4 创建 Canal Admin 数据库

在任意一台 MySQL 服务器上执行：

-- 登录 MySQL
mysql -hlocalhost -uroot -p

-- 创建 Canal Admin 数据库
CREATE DATABASE IF NOT EXISTS canal_manager DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;

-- 创建 Canal 用户（注意：这里应该使用与后续配置一致的密码）
CREATE USER IF NOT EXISTS 'canal'@'192.168.192.%' IDENTIFIED BY 'canal';
GRANT ALL PRIVILEGES ON canal_manager.* TO 'canal'@'192.168.192.%';

-- 刷新权限
FLUSH PRIVILEGES;

-- 验证用户创建
SELECT User, Host FROM mysql.user WHERE User = 'canal';

USE canal_manager;

-- 创建 Canal Admin 会自动创建的表结构
quit;

重要提示：确保 Canal Admin 服务器能够访问 MySQL，测试连接：

# 在 Canal Admin 服务器上测试数据库连接
mysql -h192.168.192.170 -ucanal -p -e "SELECT 1;"

3.5 配置 Canal Admin

在 Canal Admin 服务器（192.168.192.180）上执行：

# 切换到 canal 用户
sudo -u canal bash

# 编辑配置文件
cat > /opt/canal-admin/conf/application.yml << 'EOF'
server:
  port: 8089
spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8

spring.datasource:
  # 注意：这里应该是你的 MySQL 主库 IP 地址
  address: 192.168.192.170:3306
  database: canal_manager
  username: canal
  password: canal
  driver-class-name: com.mysql.cj.jdbc.Driver
  url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false&allowPublicKeyRetrieval=true
  hikari:
    maximum-pool-size: 30
    minimum-idle: 1

canal:
  adminUser: admin

management:
  endpoints:
    web:
      exposure:
        include: health,info,metrics
EOF

# 验证配置文件语法
cat /opt/canal-admin/conf/application.yml

# 退出 canal 用户
exit

重要检查点：

确保 address 配置的是正确的 MySQL 服务器 IP
确保用户名密码与前面创建的数据库用户一致
使用 com.mysql.cj.jdbc.Driver 新版驱动类名

3.6 启动 Canal Admin

在 Canal Admin 服务器（192.168.192.180）上执行：

# 切换到 canal 用户
sudo -u canal bash

# 进入 Canal Admin 目录
cd /opt/canal-admin

# 启动 Canal Admin
sh bin/startup.sh

# 退出 canal 用户
exit

3.7 配置 Systemd 服务

在 Canal Admin 服务器（192.168.192.180）上执行：

# 创建 Systemd 服务文件
cat <<EOF | sudo tee /etc/systemd/system/canal-admin.service
[Unit]
Description=Canal Admin
After=network.target mysql.service

[Service]
Type=forking
User=canal
Group=canal
Environment="JAVA_HOME=/usr/lib/jvm/java-11-openjdk-amd64"
Environment="CANAL_ADMIN_HEAP_OPTS=-Xms1g -Xmx1g"
ExecStart=/opt/canal-admin/bin/startup.sh
ExecStop=/opt/canal-admin/bin/stop.sh
Restart=on-failure
RestartSec=10

[Install]
WantedBy=multi-user.target
EOF

# 重载 Systemd
sudo systemctl daemon-reload

# 启用开机自启
sudo systemctl enable canal-admin

# 启动服务
sudo systemctl start canal-admin

# 查看服务状态
sudo systemctl status canal-admin

# 查看详细日志
sudo journalctl -u canal-admin -f

3.8 访问 Canal Admin

在浏览器中访问 Canal Admin：

http://192.168.192.180:8089

默认用户名密码：

用户名：admin
密码：123456

首次登录后请修改密码。

集群配置=》集群名称：canal-cluster =》集群配置：192.168.192.191:2181,192.168.192.192:2181,192.168.192.193:2181 集群模版：

#################################################
#########   common argument  #############
#################################################
# tcp bind ip
canal.ip =
# register ip to zookeeper
canal.register.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
# canal instance user/passwd
# canal.user = canal
# canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458

# canal admin config
canal.admin.manager = 192.168.192.180:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 14c33c244faa5ad403e38be03037060e97bc7aea
# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster = canal-cluster
#canal.admin.register.name =

# 注意：Canal Admin集群模式下必须配置Zookeeper地址
# 即使Kafka使用KRaft模式，Canal Admin仍需要Zookeeper进行集群协调
canal.zkServers = 192.168.192.191:2181,192.168.192.192:2181,192.168.192.193:2181
# flush data to zk
# canal持久化数据到zookeeper上的更新频率,单位毫秒
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, rocketMQ, rabbitMQ
canal.serverMode = kafka
# flush meta cursor/parse position to file
# canal持久化数据到file上的目录
canal.file.data.dir = ${canal.conf.dir}
canal.file.flush.period = 1000
## memory store RingBuffer size, should be Math.pow(2,n)
# canal内存store中可缓存buffer记录数,需要为2的指数
canal.instance.memory.buffer.size = 16384
## memory store RingBuffer used memory unit size , default 1kb
canal.instance.memory.buffer.memunit = 1024 
## meory store gets mode used MEMSIZE or ITEMSIZE
# canal内存store中数据缓存模式
# 1. ITEMSIZE : 根据buffer.size进行限制,只限制记录的数量
# 2. MEMSIZE : 根据buffer.size * buffer.memunit的大小,限制缓存记录的大小    
canal.instance.memory.batch.mode = MEMSIZE
canal.instance.memory.rawEntry = true

## detecing config
# 是否开启心跳检查
canal.instance.detecting.enable = false
#canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()
# 心跳检查sql
canal.instance.detecting.sql = select 1
# 心跳检查频率,单位秒
canal.instance.detecting.interval.time = 3
# 心跳检查失败重试次数
canal.instance.detecting.retry.threshold = 3
# 非常注意：interval.time * retry.threshold值，应该参考既往DBA同学对数据库的故障恢复时间，  
# “太短”会导致集群运行态角色“多跳”；“太长”失去了活性检测的意义，导致集群的敏感度降低，Consumer断路可能性增加。
# 心跳检查失败后,是否开启自动mysql自动切换
canal.instance.detecting.heartbeatHaEnable = false
# 说明:比如心跳检查失败超过阀值后,如果该配置为true,canal就会自动链到mysql备库获取binlog数据


# support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
# 支持最大事务大小，将超过事务大小的事务切成多个事务交付
canal.instance.transaction.size = 1024
# mysql fallback connected to new master should fallback times
# canal发生mysql切换时,在新的mysql库上查找 binlog时需要往前查找的时间,单位秒
canal.instance.fallbackIntervalInSeconds = 60
# 说明:mysql主备库可能存在解析延迟或者时钟不统一,需要回退一段时间,保证数据不丢

# network config
# 网络链接参数,SocketOptions.SO_RCVBUF
canal.instance.network.receiveBufferSize = 16384
# 网络链接参数,SocketOptions.SO_SNDBUF
canal.instance.network.sendBufferSize = 16384
# 网络链接参数,SocketOptions.SO_TIMEOUT
canal.instance.network.soTimeout = 30

# binlog filter config
canal.instance.filter.druid.ddl = true
# ddl语句是否隔离发送,开启隔离可保证每次只返回发送一条ddl数据,不和其他dml语句混合返回.(otter ddl同步使用)
canal.instance.filter.query.dcl = true
# 是否忽略DML的query语句,比如insert/update/delete table.(mysql5.6的ROW模式可以包含statement模式的query记录)
canal.instance.filter.query.dml = true
# 是否忽略DDL的query语句,比如create table/alater table/drop table/rename table/create index/drop index. (目前支持的ddl类型主要为table级别的操作,create databases/trigger/procedure暂时划分为dcl类型)
canal.instance.filter.query.ddl = true
# 注意:上面三个参数默认都是false 如无必要建议设置为true，否则后面你过滤表的时候可能会出现不生效的情况
anal.instance.filter.table.error = true
canal.instance.filter.rows = false
# 把事务头尾过滤
canal.instance.filter.transaction.entry = true
canal.instance.filter.dml.insert = false
canal.instance.filter.dml.update = false
canal.instance.filter.dml.delete = false

# binlog format/image check
#canal.instance.binlog.format = ROW,STATEMENT,MIXED 
canal.instance.binlog.format = ROW,STATEMENT,MIXED
canal.instance.binlog.image = FULL,MINIMAL,NOBLOB

# binlog ddl isolation
canal.instance.get.ddl.isolation = false

# parallel parser config
canal.instance.parser.parallel = true
## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()
#canal.instance.parser.parallelThreadSize = 16
## disruptor ringbuffer size, must be power of 2
canal.instance.parser.parallelBufferSize = 256

# table meta tsdb info #关于时间序列版本
canal.instance.tsdb.enable = true
canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}
canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
canal.instance.tsdb.dbUsername = canal
canal.instance.tsdb.dbPassword = canal
# dump snapshot interval, default 24 hour
canal.instance.tsdb.snapshot.interval = 24
# purge snapshot expire , default 360 hour(15 days)
canal.instance.tsdb.snapshot.expire = 360

#################################################
#########   destinations  #############
#################################################
canal.destinations = 
# conf root dir
canal.conf.dir = ../conf
# auto scan instance dir add/remove and start/stop instance
# 如果配置为true,canal.conf.dir目录下的instance配置变化会自动触发
# a. instance目录新增: 触发instance配置载入,lazy为true时则自动启动
# b. instance目录删除:卸载对应instance配置,如已启动则进行关闭
# c. instance.properties文件变化:reload instance配置,如已启动自动进行重启操作
# 开启instance自动扫描
canal.auto.scan = true

# instance自动扫描的间隔时间,单位秒
canal.auto.scan.interval = 5
# set this value to 'true' means that when binlog pos not found, skip to latest.
# WARN: pls keep 'false' in production env, or if you know what you want.
canal.auto.reset.latest.pos.mode = false

canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml

canal.instance.global.mode = manager
# 全局lazy模式
canal.instance.global.lazy = false
canal.instance.global.manager.address = ${canal.admin.manager}
#canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
#canal.instance.global.spring.xml = classpath:spring/file-instance.xml
# 此处建议设置为classpath:spring/default-instance.xml
canal.instance.global.spring.xml = classpath:spring/default-instance.xml

##################################################
#########        MQ Properties      #############
##################################################
# aliyun ak/sk , support rds/mq
canal.aliyun.accessKey =
canal.aliyun.secretKey =
canal.aliyun.uid=

canal.mq.flatMessage = true
canal.mq.canalBatchSize = 50
canal.mq.canalGetTimeout = 100
# Set this value to "cloud", if you want open message trace feature in aliyun.
canal.mq.accessChannel = local

canal.mq.database.hash = true
canal.mq.send.thread.size = 30
canal.mq.build.thread.size = 8

##################################################
#########        Kafka        #############
##################################################
# kafka bootstrap 信息，可以不填所有的
kafka.bootstrap.servers = 192.168.192.211:9092,192.168.192.212:9092,192.168.192.213:9092
kafka.acks = all
kafka.compression.type = none
kafka.batch.size = 16384
# 建议设为50
kafka.linger.ms = 50
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"

##################################################
#########       RocketMQ      #############
##################################################
rocketmq.producer.group = test
rocketmq.enable.message.trace = false
rocketmq.customized.trace.topic =
rocketmq.namespace =
rocketmq.namesrv.addr = 127.0.0.1:9876
rocketmq.retry.times.when.send.failed = 0
rocketmq.vip.channel.enabled = false
rocketmq.tag = 

##################################################
#########       RabbitMQ      #############
##################################################
rabbitmq.host =
rabbitmq.virtual.host =
rabbitmq.exchange =
rabbitmq.username =
rabbitmq.password =
rabbitmq.deliveryMode =

4. Canal Server 部署

4.1 环境准备

在两台 Canal Server 服务器（192.168.192.181、192.168.192.182）上执行：

4.1.1 安装 Java

# 更新软件包列表
sudo apt update

# 安装 OpenJDK 11
sudo apt install -y openjdk-11-jdk

# 验证安装
java -version

# 设置 JAVA_HOME 环境变量
echo 'export JAVA_HOME=/usr/lib/jvm/java-11-openjdk-amd64' | sudo tee -a /etc/environment
source /etc/environment

4.1.2 创建 Canal 用户

# 创建 canal 用户
sudo useradd -m -s /bin/bash canal

4.2 Canal Server 安装

在两台 Canal Server 服务器（192.168.192.181、192.168.192.182）上执行：

# 进入安装目录
cd /opt

# 下载 Canal Deployer (使用 sudo 确保有权限)
sudo wget https://github.com/alibaba/canal/releases/download/canal-1.1.8/canal.deployer-1.1.8.tar.gz

# 创建安装目录
sudo mkdir -p /opt/canal

# 解压到指定目录
sudo tar -zxvf canal.deployer-1.1.8.tar.gz -C /opt/canal

# 统一赋值权限
sudo chown -R canal:canal /opt/canal

4.3 配置 Canal Server

在 Canal Server 1（192.168.192.181）上执行：

# 切换到 canal 用户
sudo -u canal bash

# 编辑 canal_local.properties，启用 Canal Admin
cat > /opt/canal/conf/canal_local.properties << 'EOF'
# Canal Server 连接 Canal Admin 配置
canal.admin.manager = 192.168.192.180:8089
canal.admin.port = 11110
canal.admin.user = admin
# 注意：这里的密码必须与 Canal Admin 数据库（canal_manager.user 表）中存储的密码加密后的内容一致
# 如果您已在 Canal Admin 界面更改了密码，请从数据库查询新的加密密码值
canal.admin.passwd = 14c33c244faa5ad403e38be03037060e97bc7aea

# Canal Server 注册信息
canal.register.ip = 192.168.192.181

# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster = canal-cluster
canal.admin.register.name = canal-server-1
EOF

# 退出 canal 用户
exit

4.3.2 通过 Canal Admin 管理 Instance

在 Admin 管理模式下，Instance 配置通过 Canal Admin Web 界面统一管理，无需在本地创建 instance.properties 文件。Canal Server 启动后会从 Admin 服务器动态获取实例配置。

在 Canal Server 2（192.168.192.182）上执行：

重要说明： 在 Canal Admin 管理模式下：

Canal Server 通过 canal_local.properties 连接到 Canal Admin 服务器
所有 Instance 配置（包括 MySQL 连接信息、过滤规则、Kafka 配置等）都通过 Admin 界面统一管理
Canal Server 启动后会自动从 Admin 服务器拉取相应的实例配置
无需在本地创建 instance.properties 文件，所有配置都由 Admin 动态推送

安全注意事项：

示例中使用的密码已进行 SHA1 加密处理，实际部署时请使用强密码
密码格式为十六进制字符串，长度为40位
重要：Canal Server 中配置的 canal.admin.passwd 必须与 Canal Admin 数据库（canal_manager.user 表）中存储的密码加密后的内容完全一致，如果在 Canal Admin 界面修改了密码，需要查询数据库获取新的加密密码并更新 Canal Server 配置

可通过以下命令生成加密密码：

echo -n "your_password" | sha1sum | cut -d ' ' -f 1

如果在数据库中配置密码，也可以使用数据库内置函数加密，例如 MySQL 中的 PASSWORD() 函数：
```
-- 在 MySQL 中加密密码
SELECT PASSWORD('your_password');
-- 或者使用 SHA1
SELECT SHA1('your_password');
```

从数据库查询当前密码：如果需要获取当前 Canal Admin 中设置的密码，可以查询数据库：

-- 连接到 canal_manager 数据库
mysql -h<mysql_host> -u<mysql_user> -p canal_manager
-- 查询用户表中的密码
SELECT id, username, password, gmt_create, gmt_modified FROM user;

# 切换到 canal 用户
sudo -u canal bash

# 编辑 canal_local.properties，启用 Canal Admin
cat > /opt/canal/conf/canal_local.properties << 'EOF'
# Canal Server 连接 Canal Admin 配置
canal.admin.manager = 192.168.192.180:8089
canal.admin.port = 11110
canal.admin.user = admin
# 注意：这里的密码必须与 Canal Admin 数据库（canal_manager.user 表）中存储的密码加密后的内容一致
# 如果您已在 Canal Admin 界面更改了密码，请从数据库查询新的加密密码值
canal.admin.passwd = 14c33c244faa5ad403e38be03037060e97bc7aea

# Canal Server 注册信息
canal.register.ip = 192.168.192.182

# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster = canal-cluster
canal.admin.register.name = canal-server-2
EOF

# 退出 canal 用户
exit

4.4 启动 Canal Server

在两台 Canal Server 服务器上执行：

# 递归设置canal安装目录的所属用户/组为canal
sudo chown -R canal:canal /opt/canal

# 如果使用canal家目录方案，确保目录存在且权限正确
sudo mkdir -p /home/canal
sudo chown -R canal:canal /home/canal
sudo chmod 700 /home/canal

# 启动 Canal
sudo -u canal /opt/canal/bin/startup.sh local

# 查看日志
sudo -u canal tail -f /opt/canal/logs/canal/canal.log
sudo -u canal tail -f /opt/canal/logs/example/example.log

4.5 配置 Systemd 服务

在两台 Canal Server 服务器上执行：

# 创建 Systemd 服务文件
sudo cat > /etc/systemd/system/canal.service << 'EOF'
[Unit]
Description=Canal Server
After=network.target

[Service]
Type=forking
User=canal
Group=canal
# 确认JAVA_HOME路径正确（可通过 `update-alternatives --config java` 验证）
Environment="JAVA_HOME=/usr/lib/jvm/java-11-openjdk-amd64"
# JVM内存配置（根据服务器内存调整，2g如果内存不足可改为1g）
Environment="CANAL_HEAP_OPTS=-Xms2g -Xmx2g"
# 核心新增：启动前先执行stop.sh，清理残留PID/进程
ExecStartPre=/opt/canal/bin/stop.sh
# 启动指令
ExecStart=/opt/canal/bin/startup.sh local
# 停止指令
ExecStop=/opt/canal/bin/stop.sh
# 仅当进程异常退出（非0码）时重启，避免无限重启
Restart=on-failure
RestartSec=10
# 限制重启次数（避免死循环）
StartLimitBurst=3
StartLimitInterval=60s

[Install]
WantedBy=multi-user.target
EOF

# 重载 Systemd
sudo systemctl daemon-reload

# 启用开机自启
sudo systemctl enable canal

# 启动服务
sudo systemctl start canal

# 查看服务状态
sudo systemctl status canal

# 重启1服务
sudo systemctl restart canal

5. Canal Admin 配置

5.1 通过 Canal Admin 添加 Canal Server

访问 Canal Admin：http://192.168.192.180:8089
登录（用户名：admin，密码：123456）注意：此处为示例密码，实际部署时请使用强密码并在 Canal Admin 的 application.yml 中配置加密后的密码
点击"Server 管理"
点击"添加 Server"
填写 Canal Server 1 信息：
- Server 名称：Canal Server 1
- Server IP：192.168.192.181
- Server 端口：11110
- Admin 端口：11110
点击"保存"
填写 Canal Server 2 信息：
- Server 名称：Canal Server 2
- Server IP：192.168.192.182
- Server 端口：11110
- Admin 端口：11110
点击"保存"

5.2 添加 Instance

点击"Instance 管理"
点击"添加 Instance"

#################################################
## mysql serverId , v1.0.26+ will autoGen
# slaveId 注意不要与数据源库的id 一样 
# canal.instance.mysql.slaveId=0

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

# position info
# 数据源mysql的ip:port
canal.instance.master.address=192.167.192.171: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

#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#canal.instance.standby.gtid=

# username/password
# 数据源mysql的用户，根据实际设置
# 注意：需要在源数据库上手动创建该用户并授予权限
# CREATE USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY 'canal';
# GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
canal.instance.dbUsername=canal
# 数据源mysql的密码，根据实际设置
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
# enable druid Decrypt database password
canal.instance.enableDruid=false
#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==

# table regex
# 需要同步的表，这里配置的意思是所有数据库下的所有表。
# 但是如果之前配置canal.instance.filter.query.dcl、canal.instance.filter.query.dml、canal.instance.filter.query.ddl 三个参数不设置为true 这个过滤可能失效
canal.instance.filter.regex=big_data\\..*
# table black regex
# 同上，只不过这里是黑名单，不同步的表
canal.instance.filter.black.regex=
# table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
# 需要同步的字段
#canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch
# table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
# 不需要同步的字段
#canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch

# mq config
# 默认数据同步的主题,所有未被canal.instance.filter.regex匹配的表都会同步到主题all_topic中
canal.mq.topic=all_topic
# dynamic topic route by schema or table regex
# 自己指定主题，big_data库中的所有表都会同步到big_data_topic中
canal.mq.dynamicTopic=big_data_topic:big_data\\..*
canal.mq.partition=0
# hash partition config
canal.mq.partitionsNum=3
#canal.mq.partitionHash=test.table:id^name,.*\\..*
#################################################

假设需要新增 big_data 数据库的同步：

步骤 1：创建 big_data Topic

# 在任意一台 Kafka 服务器上执行
cd /opt/kafka/kafka

# 创建 all_topic Topic（3 分区，3 副本）
./bin/kafka-topics.sh --create \
  --bootstrap-server 192.168.192.211:9092 \
  --topic all_topic \
  --partitions 3 \
  --replication-factor 3

# 创建 big_data Topic（3 分区，3 副本）
./bin/kafka-topics.sh --create \
  --bootstrap-server 192.168.192.211:9092 \
  --topic big_data_topic \
  --partitions 3 \
  --replication-factor 3

# 查看 Topic
./bin/kafka-topics.sh --list --bootstrap-server 192.168.192.211:9092

# 查看 Topic 详情
./bin/kafka-topics.sh --describe --bootstrap-server 192.168.192.211:9092 --topic big_data_topic

# 4. 验证消息生产和消费
./bin/kafka-console-producer.sh \
  --bootstrap-server 192.168.192.211:9092 \
  --topic big_data_topic

# 在 Kafka 上消费消息
cd /opt/kafka
./bin/kafka-console-consumer.sh \
  --bootstrap-server 192.168.192.211:9092 \
  --topic big_data_topic \
  --from-beginning

exit

6.3 动态扩展的优势

对比项	修改 Canal Server 配置	Canal Admin 动态扩展
需要重启 Canal Server	✅ 是	❌ 否
影响现有数据库同步	⚠️ 是（重启期间）	❌ 否
配置复杂度	简单	中等
灵活性	低	高
可视化管理	❌ 无	✅ 有
推荐场景	数据库数量固定	数据库频繁变更

7. 数据同步验证

7.1 测试数据变更

在主库 1 上执行：

-- 登录 MySQL（通过 VIP）
mysql -h192.168.192.170 -uroot -p

USE test_db;

-- 插入数据
INSERT INTO user (username, email, age) VALUES ('user4', 'user4@example.com', 35);

-- 更新数据
UPDATE user SET age = 36 WHERE username = 'user4';

-- 删除数据
DELETE FROM user WHERE username = 'user4';

-- 查看数据
SELECT * FROM user;

quit;

7.2 查看 Canal 日志

在两台 Canal Server 服务器上执行：

# 查看 Canal Server 日志
sudo -u canal tail -f /opt/canal/logs/example/example.log

# 查看 Canal Admin 日志
sudo -u canal tail -f /opt/canal-admin/logs/canal-admin/manager.log

7.3 查看 Kafka 消息

在任意一台 Kafka 服务器上执行：

# 消费 Kafka 消息
cd /opt/kafka/kafka
./bin/kafka-console-consumer.sh \
  --bootstrap-server 192.168.192.211:9092 \
  --topic test_db \
  --from-beginning

# 退出 kafka 用户
exit

消息示例（JSON 格式）：

{
  "data": [
    {
      "id": "4",
      "username": "user4",
      "email": "user4@example.com",
      "age": "35",
      "created_at": "2026-01-23 10:00:00",
      "updated_at": "2026-01-23 10:00:00"
    }
  ],
  "database": "test_db",
  "es": 1737264000000,
  "id": 4,
  "isDdl": false,
  "mysqlType": {
    "id": "BIGINT",
    "username": "VARCHAR(50)",
    "email": "VARCHAR(100)",
    "age": "INT",
    "created_at": "TIMESTAMP",
    "updated_at": "TIMESTAMP"
  },
  "old": [
    {
      "age": "35"
    }
  ],
  "pkNames": [
    "id"
  ],
  "sql": "",
  "sqlType": {
    "id": -5,
    "username": 12,
    "email": 12,
    "age": 4,
    "created_at": 93,
    "updated_at": 93
  },
  "table": "user",
  "ts": 1737264000123,
  "type": "INSERT"
}

字段说明：

data：变更后的数据
database：数据库名称
es：执行时间（毫秒）
id：binlog 位置
isDdl：是否为 DDL 语句
mysqlType：MySQL 字段类型
old：变更前的数据（UPDATE 时有值）
pkNames：主键字段
sql：执行的 SQL 语句
sqlType：SQL 类型
table：表名
ts：时间戳
type：操作类型（INSERT/UPDATE/DELETE）

7.4 验证 MySQL 双主同步

在主库 2 上执行：

-- 登录 MySQL
mysql -hlocalhost -uroot -p

USE test_db;

-- 查看数据
SELECT * FROM user;

quit;

7.5 验证 Keepalived 故障切换

# 在主库 1 上停止 MySQL
sudo systemctl stop mysql

# 查看 VIP 是否切换到主库 2
ip a

# 在主库 2 上查看 VIP
ssh 192.168.192.172 "ip a"

# 恢复主库 1
sudo systemctl start mysql

# 查看 VIP 是否切换回主库 1
ip a

8. 高可用方案

8.1 Canal Server 高可用

8.1.1 双实例部署

Canal Server 采用双实例部署，每个实例连接本地 MySQL：

Canal Server 1：部署在 192.168.192.181，连接 MySQL 主库 1（192.168.192.171）
Canal Server 2：部署在 192.168.192.182，连接 MySQL 主库 2（192.168.192.172）

8.1.2 故障切换机制

当 MySQL 主库故障时：

Keepalived VIP 自动切换到备库
应用通过 VIP 访问新的主库
Canal Server 2 继续解析本地 MySQL 的 binlog
数据同步不中断

8.1.3 数据一致性

由于 MySQL 双主之间实时同步 binlog，两个 Canal Server 实例解析的数据是一致的，不会出现数据重复或丢失。

8.2 Canal Admin 高可用

8.2.1 单实例部署

Canal Admin 采用单实例部署，故障时可通过 Web 界面快速切换 Canal Server。

8.2.2 故障切换机制

当 Canal Admin 故障时：

通过 Canal Admin Web 界面快速切换 Canal Server
Canal Server 继续解析 binlog
数据同步不中断

8.2.3 数据一致性

Canal Admin 通过数据库管理配置，配置变更实时生效，保证数据一致性。

8.3 Kafka 高可用

Kafka 3 节点集群提供高可用：

多副本存储：每个 Topic 分区有 3 个副本
自动故障转移：Broker 故障时，Leader 自动切换
数据不丢失：min.insync.replicas=2 保证至少 2 个副本同步成功

9. 监控与告警

9.1 Canal Server 监控

9.1.1 关键指标

解析延迟：binlog 解析延迟时间
解析速度：每秒解析的 binlog 事件数
投递延迟：消息投递到 Kafka 的延迟
投递速度：每秒投递的消息数
连接状态：与 MySQL 的连接状态
binlog 位置：当前 binlog 位置

9.1.2 告警规则

groups:
  - name: canal-server
    interval: 30s
    rules:
      - alert: CanalServerDown
        expr: up{job="canal"} == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Canal Server down"
          description: "Canal Server {{ $labels.instance }} is down"

      - alert: CanalParseDelayHigh
        expr: canal_parse_delay_seconds > 60
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Canal parse delay high"
          description: "Canal parse delay is {{ $value }} seconds"

      - alert: CanalSendDelayHigh
        expr: canal_send_delay_seconds > 60
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Canal send delay high"
          description: "Canal send delay is {{ $value }} seconds"

      - alert: CanalMySQLConnectionLost
        expr: canal_mysql_connection_status == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Canal MySQL connection lost"
          description: "Canal lost connection to MySQL"

9.2 Canal Admin 监控

9.2.1 关键指标

Canal Admin 状态：Canal Admin 服务运行状态
Canal Server 连接状态：Canal Admin 与 Canal Server 的连接状态
Instance 数量：管理的 Canal Instance 数量
配置变更历史：配置变更记录

9.2.2 告警规则

groups:
  - name: canal-admin
    interval: 30s
    rules:
      - alert: CanalAdminDown
        expr: up{job="canal-admin"} == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Canal Admin down"
          description: "Canal Admin {{ $labels.instance }} is down"

      - alert: CanalServerConnectionLost
        expr: canal_admin_server_connection_status == 0
        for: 1m
        labels:
          severity: warning
        annotations:
          summary: "Canal Admin lost connection to Canal Server"
          description: "Canal Admin lost connection to Canal Server {{ $labels.server }}"

      - alert: InstanceCountHigh
        expr: canal_admin_instance_count > 100
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Canal Admin instance count high"
          description: "Canal Admin has {{ $value }} instances"

9.3 Kafka 监控

参考 Kafka 虚拟机部署指南中的监控与告警章节。

9.4 MySQL 监控

参考 MySQL 高可用故障转移（双主热备）中的监控与告警章节。

10. 故障排除

10.1 Canal Server 故障排除

10.1.1 Canal Server 无法连接 MySQL

问题：Canal Server 启动失败，无法连接 MySQL

# 查看 Canal Server 日志
sudo -u canal tail -f /opt/canal/logs/example/example.log

# 常见原因：
# 1. MySQL 用户权限不足
mysql -ucanal -p -h 192.168.192.171

# 2. MySQL 未开启 binlog
mysql -uroot -p -e "SHOW VARIABLES LIKE 'log_bin';"

# 3. MySQL binlog 格式不正确
mysql -uroot -p -e "SHOW VARIABLES LIKE 'binlog_format';"

# 4. 网络不通
ping 192.168.192.171
telnet 192.168.192.171 3306

10.1.2 Canal Server 解析 binlog 失败

问题：Canal Server 无法解析 binlog

# 查看 Canal Server 日志
sudo -u canal tail -f /opt/canal/logs/example/example.log

# 常见原因：
# 1. binlog 格式不是 ROW
# 解决：修改 MySQL 配置 binlog_format=ROW

# 2. binlog_row_image 不是 FULL
# 解决：修改 MySQL 配置 binlog_row_image=FULL

# 3. 表结构变更
# 解决：重启 Canal Server

10.1.3 Canal Server 发送消息到 Kafka 失败

问题：Canal Server 无法发送消息到 Kafka

# 查看 Canal Server 日志
sudo -u canal tail -f /opt/canal/logs/example/example.log

# 查看 Kafka 日志
tail -f /opt/kafka/kafka/logs/server.log

# 常见原因：
# 1. Kafka 集群不可达
telnet 192.168.1.101 9092

# 2. Kafka Topic 不存在
./bin/kafka-topics.sh --list --bootstrap-server 192.168.192.211:9092

# 3. Kafka 配置错误
# 解决：检查 Canal Admin 配置中的 Kafka 服务器地址

10.2 Canal Admin 故障排除

10.2.1 Canal Admin 无法启动

问题：Canal Admin 启动失败

# 查看 Canal Admin 日志
sudo -u canal tail -f /opt/canal-admin/logs/canal-admin/manager.log

# 常见原因：
# 1. Java 版本不兼容
java -version

# 2. 数据库连接失败
mysql -h192.168.192.171 -ucanal -p -D canal_manager

# 3. 端口被占用
sudo netstat -tlnp | grep 8089

# 4. 配置文件错误
cat /opt/canal-admin/conf/application.yml

10.2.2 Canal Admin 无法连接 Canal Server

问题：Canal Admin 无法连接 Canal Server

# 查看 Canal Admin 日志
sudo -u canal tail -f /opt/canal-admin/logs/canal-admin/manager.log

# 常见原因：
# 1. Canal Server 未启动
sudo systemctl status canal

# 2. Canal Server 端口不通
telnet 192.168.192.181 11110

# 3. Canal Server Admin 端口配置错误
# 解决：检查 canal.properties 中的 canal.admin.manager.port

# 4. 网络不通
ping 192.168.192.181

# 5. 防火墙阻止
sudo ufw status

10.2.3 Canal Admin 无法添加 Instance

问题：Canal Admin 无法添加 Instance

# 查看 Canal Admin 日志
sudo -u canal tail -f /opt/canal-admin/logs/canal-admin/manager.log

# 常见原因：
# 1. Canal Server 未连接
# 解决：检查 Canal Server 状态和连接

# 2. Instance 名称已存在
# 解决：使用不同的 Instance 名称

# 3. 数据库过滤规则错误
# 解决：检查数据库过滤规则格式

# 4. Kafka 配置错误
# 解决：检查 Kafka 服务器地址和端口

10.3 诊断工具

10.3.1 Canal Server 诊断

# 查看 Canal Server 状态
curl http://localhost:11111/status

# 查看 Canal Server 实例列表
curl http://localhost:11111/destinations

# 查看 Canal Server 实例详情
curl http://localhost:11111/destination/example

10.3.2 Canal Admin 诊断

# 查看 Canal Admin 状态
curl http://localhost:8089/actuator/health

# 查看 Canal Admin 指标
curl http://localhost:8089/actuator/metrics

10.3.3 MySQL 诊断

# 查看 binlog 位置
mysql -uroot -p -e "SHOW MASTER STATUS;"

# 查看 binlog 事件
mysql -uroot -p -e "SHOW BINLOG EVENTS IN 'mysql-bin.000001' LIMIT 10;"

10.3.4 Kafka 诊断

# 查看 Topic 信息
./bin/kafka-topics.sh --describe --bootstrap-server 192.168.192.211:9092 --topic test_db

# 查看消费者组
./bin/kafka-consumer-groups.sh --describe --bootstrap-server 192.168.192.211:9092 --group test_group

# 查看 Topic 消息
./bin/kafka-console-consumer.sh --bootstrap-server 192.168.192.211:9092 --topic test_db --from-beginning --max-messages 10

11. 最佳实践

11.1 性能优化

11.1.1 Canal Server 性能优化

# canal.properties 优化配置
canal.instance.network.receiveBufferSize = 16384
canal.instance.network.sendBufferSize = 16384
canal.instance.parser.parallel = true
canal.instance.parser.parallelThreadSize = 16

# 增加批处理大小
canal.mq.batchSize = 500
canal.mq.maxBatchTimeout = 1000

11.1.2 Canal Admin 性能优化

# application.yml 优化配置
canal:
  adminManager:
    port: 11110
  metrics:
    port: 11112
    pull:
      batch-size: 500
      timeout: 10000

11.1.3 Kafka 性能优化

参考 Kafka 虚拟机部署指南中的性能优化章节。

11.2 安全配置

11.2.1 MySQL 安全

-- 限制 Canal 用户只能从特定 IP 访问
CREATE USER 'canal'@'192.168.192.%' IDENTIFIED WITH mysql_native_password BY 'canal@123456';

-- 使用 SSL 连接
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%' REQUIRE SSL;

11.2.2 Canal Admin 安全

# application.yml 安全配置
server:
  port: 8089

spring:
  security:
    user:
      name: admin
      # 注意：实际密码存储在 canal_manager 数据库的 user 表中
      # 通过以下命令可以查询数据库中的密码：
      # mysql -h<mysql_host> -u<mysql_user> -p -e "SELECT * FROM canal_manager.user;"
      # 如果更改了数据库中的密码，记得同时更新 Canal Server 中的 canal.admin.passwd 配置
      password: '{cipher}encrypted_password_from_database'
      roles: ADMIN

重要说明： Canal Admin 的用户认证信息存储在 canal_manager 数据库中，具体在 user 表中。如果需要查询当前密码，可以执行以下 SQL 查询：

-- 查询用户表中的密码信息
SELECT id, username, password, gmt_create, gmt_modified FROM canal_manager.user;

配置说明： 在 Canal Admin 管理模式下，以下配置项需要注意：

连接配置：Canal Server 通过 canal_local.properties 连接到 Canal Admin
服务器模式：canal.serverMode 等服务器级别配置也在 Admin 界面中统一管理
Kafka 配置：Kafka 连接信息在 Admin 界面中配置，不需要在本地配置文件中指定
实例配置：具体的实例配置（如 MySQL 连接、过滤规则等）由 Admin 动态推送，不需要在本地配置

11.2.3 Kafka 安全

参考 Kafka 虚拟机部署指南中的安全配置章节。

11.3 数据一致性

11.3.1 幂等性处理

消费者需要实现幂等性处理，避免重复消费：

public class CanalKafkaConsumer {
    private RedisTemplate<String, String> redisTemplate;

    public void processMessage(String message) {
        // 解析消息
        JSONObject json = JSON.parseObject(message);
        String id = json.getString("id");
        String type = json.getString("type");

        // 生成唯一键
        String key = "canal:" + type + ":" + id;

        // 检查是否已处理
        if (redisTemplate.hasKey(key)) {
            return;
        }

        // 处理消息
        processData(json);

        // 标记已处理
        redisTemplate.opsForValue().set(key, "1", 24, TimeUnit.HOURS);
    }
}

11.3.2 事务处理

确保消费者处理消息时的事务一致性：

@Transactional
public void processMessage(String message) {
    // 解析消息
    JSONObject json = JSON.parseObject(message);

    // 处理数据
    updateDatabase(json);

    // 更新缓存
    updateCache(json);
}

11.4 监控指标

11.4.1 关键指标

Canal Server 解析延迟：< 1 秒
Canal Server 投递延迟：< 1 秒
Canal Admin 响应时间：< 1 秒
Kafka 消息延迟：< 5 秒
Kafka 消费延迟：< 10 秒
MySQL binlog 延迟：< 1 秒

11.4.2 告警阈值

Canal Server 解析延迟 > 10 秒：Warning
Canal Server 投递延迟 > 10 秒：Warning
Canal Admin 响应时间 > 5 秒：Warning
Kafka 消息延迟 > 30 秒：Warning
Kafka 消费延迟 > 60 秒：Critical
Canal Server 连接失败：Critical
Canal Admin 连接失败：Critical

12. 附录

12.1 配置参数说明

12.1.1 Canal Server 参数

参数	默认值	说明
`canal.instance.master.address`	-	MySQL 地址
`canal.instance.dbUsername`	-	MySQL 用户名
`canal.instance.dbPassword`	-	MySQL 密码
`canal.instance.filter.regex`	.*	binlog 过滤规则
`canal.instance.filter.black.regex`	-	binlog 黑名单
`canal.admin.manager`	-	Canal Admin 服务器地址
`canal.admin.port`	11110	Canal Admin 端口

12.1.2 Canal Admin 参数

参数	默认值	说明
`server.port`	8089	Canal Admin Web 界面端口
`spring.datasource.address`	127.0.0.1:3306	MySQL 服务器地址
`canal.adminManager.port`	11110	Canal Server Admin 端口
`canal.metrics.port`	11112	Canal Server 指标端口

12.1.3 MySQL 参数

参数	默认值	说明
`log-bin`	-	binlog 文件路径
`binlog_format`	ROW	binlog 格式
`binlog_row_image`	FULL	binlog 行镜像
`server-id`	-	服务器 ID
`gtid-mode`	OFF	GTID 模式

12.2 常用命令参考

12.2.1 Canal Server 命令

# 启动 Canal Server
sudo -u canal /opt/canal/bin/startup.sh

# 停止 Canal Server
sudo -u canal /opt/canal/bin/stop.sh

# 重启 Canal Server
sudo -u canal /opt/canal/bin/restart.sh

# 查看 Canal Server 状态
sudo systemctl status canal

# 查看 Canal Server 日志
sudo -u canal tail -f /opt/canal/logs/canal/canal.log

12.2.2 Canal Admin 命令

# 启动 Canal Admin
sudo -u canal /opt/canal-admin/bin/startup.sh

# 停止 Canal Admin
sudo -u canal /opt/canal-admin/bin/stop.sh

# 重启 Canal Admin
sudo -u canal /opt/canal-admin/bin/restart.sh

# 查看 Canal Admin 状态
sudo systemctl status canal-admin

# 查看 Canal Admin 日志
sudo -u canal tail -f /opt/canal-admin/logs/canal-admin/manager.log

12.2.3 MySQL 命令

# 查看 binlog 状态
mysql -uroot -p -e "SHOW MASTER STATUS;"

# 查看 binlog 事件
mysql -uroot -p -e "SHOW BINLOG EVENTS IN 'mysql-bin.000001' LIMIT 10;"

# 查看主从复制状态
mysql -uroot -p -e "SHOW SLAVE STATUS\G"

12.2.4 Kafka 命令

# 创建 Topic
./bin/kafka-topics.sh --create --bootstrap-server <host:port> --topic <name> --partitions <num> --replication-factor <num>

# 查看 Topic
./bin/kafka-topics.sh --list --bootstrap-server <host:port>

# 消费消息
./bin/kafka-console-consumer.sh --bootstrap-server <host:port> --topic <name> --from-beginning

12.3 官方资源链接

Canal 官方网站: https://github.com/alibaba/canal
Canal 文档: https://github.com/alibaba/canal/wiki
Canal Admin: https://github.com/alibaba/canal/wiki/Canal-Admin-Guide
Canal HA: https://github.com/alibaba/canal/wiki/Canal-HA
MySQL 官方文档: https://dev.mysql.com/doc/
Kafka 官方文档: https://kafka.apache.org/documentation/

12.4 参考资料

MySQL 高可用故障转移（双主热备）: ../MySQL/mysql-ha-failover-dual-master.md
Kafka 虚拟机部署指南: ./kafka-vm-deployment-guide.md
Zookeeper 集群部署指南: ./zookeeper-cluster-deployment-guide.md
MySQL binlog: https://dev.mysql.com/doc/refman/8.0/en/binary-log.html
Kafka 消费者: https://kafka.apache.org/documentation/#consumerapi

总结

通过本方案，你已经成功将 MySQL 双主热备集群与 Canal Server + Canal Admin + Kafka 集成，实现了高可用的数据同步系统。该方案具有以下特点：

高可用性：MySQL 双主 + Canal Server 双实例，任一节点故障不影响数据同步
实时性：基于 binlog，实现毫秒级数据同步
可靠性：Kafka 3 节点集群，数据多副本存储
扩展性：Kafka 天然支持多消费者，便于扩展
解耦合：生产者和消费者完全解耦
可视化管理：通过 Canal Admin Web 界面管理 Canal Server 集群
动态扩展：通过 Canal Admin 动态添加新数据库，无需重启 Canal Server

关键要点

Canal Server 双实例部署：每个 Canal Server 实例连接本地 MySQL，避免跨网络访问
独立运行：两个 Canal Server 实例独立运行，互不影响
数据一致性：MySQL 双主实时同步，保证 Canal Server 解析的数据一致
Canal Admin 可视化管理：通过 Web 界面管理 Canal Server 集群，动态配置 Instance
动态扩展：通过 Canal Admin 动态添加新数据库，无需重启 Canal Server

下一步建议

根据实际业务需求调整 Canal Server 和 Canal Admin 配置
配置监控和告警，及时发现和处理问题
实现消费者幂等性处理，避免重复消费
根据数据量选择合适的分区数和副本数
定期备份 MySQL 和 Kafka 数据
进行故障演练，验证高可用方案的有效性
学习 Canal Admin 的使用，充分利用可视化管理功能

如有任何问题或建议，欢迎反馈！

概述​

前置条件​

方案优势​

目录​

1. 架构设计​

1.1 整体架构​

1.2 数据流向​

1.3 高可用机制​

2. 部署规划​

2.1 服务器规划​

2.2 端口规划​

2.3 MySQL 前置配置​

2.4 创建 Canal 用户​

2.5 创建测试数据库和表​

3. Canal Admin 部署​

3.1 Canal Admin 介绍​

3.2 环境准备​

3.2.1 安装 Java​

3.2.2 创建 Canal Admin 用户​

3.3 Canal Admin 安装​

3.4 创建 Canal Admin 数据库​

3.5 配置 Canal Admin​

3.6 启动 Canal Admin​

3.7 配置 Systemd 服务​

3.8 访问 Canal Admin​

4. Canal Server 部署​

4.1 环境准备​

4.1.1 安装 Java​

4.1.2 创建 Canal 用户​

4.2 Canal Server 安装​

4.3 配置 Canal Server​

4.3.2 通过 Canal Admin 管理 Instance​

4.4 启动 Canal Server​

4.5 配置 Systemd 服务​

5. Canal Admin 配置​

5.1 通过 Canal Admin 添加 Canal Server​

5.2 添加 Instance​

步骤 1：创建 big_data Topic​

6.3 动态扩展的优势​

7. 数据同步验证​

7.1 测试数据变更​

7.2 查看 Canal 日志​

7.3 查看 Kafka 消息​

7.4 验证 MySQL 双主同步​

7.5 验证 Keepalived 故障切换​

8. 高可用方案​

8.1 Canal Server 高可用​

8.1.1 双实例部署​

8.1.2 故障切换机制​

8.1.3 数据一致性​

8.2 Canal Admin 高可用​

8.2.1 单实例部署​

8.2.2 故障切换机制​

8.2.3 数据一致性​

8.3 Kafka 高可用​

9. 监控与告警​

9.1 Canal Server 监控​

9.1.1 关键指标​

9.1.2 告警规则​

9.2 Canal Admin 监控​

9.2.1 关键指标​

9.2.2 告警规则​

9.3 Kafka 监控​

9.4 MySQL 监控​

10. 故障排除​

10.1 Canal Server 故障排除​

10.1.1 Canal Server 无法连接 MySQL​

10.1.2 Canal Server 解析 binlog 失败​

10.1.3 Canal Server 发送消息到 Kafka 失败​

10.2 Canal Admin 故障排除​

10.2.1 Canal Admin 无法启动​

10.2.2 Canal Admin 无法连接 Canal Server​

10.2.3 Canal Admin 无法添加 Instance​

10.3 诊断工具​

10.3.1 Canal Server 诊断​

10.3.2 Canal Admin 诊断​

10.3.3 MySQL 诊断​

10.3.4 Kafka 诊断​

11. 最佳实践​

11.1 性能优化​

概述

前置条件

方案优势

目录

1. 架构设计

1.1 整体架构

1.2 数据流向

1.3 高可用机制

2. 部署规划

2.1 服务器规划

2.2 端口规划

2.3 MySQL 前置配置

2.4 创建 Canal 用户

2.5 创建测试数据库和表

3. Canal Admin 部署

3.1 Canal Admin 介绍

3.2 环境准备

3.2.1 安装 Java

3.2.2 创建 Canal Admin 用户

3.3 Canal Admin 安装

3.4 创建 Canal Admin 数据库

3.5 配置 Canal Admin

3.6 启动 Canal Admin

3.7 配置 Systemd 服务

3.8 访问 Canal Admin

4. Canal Server 部署

4.1 环境准备

4.1.1 安装 Java

4.1.2 创建 Canal 用户

4.2 Canal Server 安装

4.3 配置 Canal Server

4.3.2 通过 Canal Admin 管理 Instance

4.4 启动 Canal Server

4.5 配置 Systemd 服务

5. Canal Admin 配置

5.1 通过 Canal Admin 添加 Canal Server

5.2 添加 Instance

步骤 1：创建 big_data Topic

6.3 动态扩展的优势

7. 数据同步验证

7.1 测试数据变更

7.2 查看 Canal 日志

7.3 查看 Kafka 消息

7.4 验证 MySQL 双主同步

7.5 验证 Keepalived 故障切换

8. 高可用方案

8.1 Canal Server 高可用

8.1.1 双实例部署

8.1.2 故障切换机制

8.1.3 数据一致性

8.2 Canal Admin 高可用

8.2.1 单实例部署

8.2.2 故障切换机制

8.2.3 数据一致性

8.3 Kafka 高可用

9. 监控与告警

9.1 Canal Server 监控

9.1.1 关键指标

9.1.2 告警规则

9.2 Canal Admin 监控

9.2.1 关键指标

9.2.2 告警规则

9.3 Kafka 监控

9.4 MySQL 监控

10. 故障排除

10.1 Canal Server 故障排除

10.1.1 Canal Server 无法连接 MySQL

10.1.2 Canal Server 解析 binlog 失败

10.1.3 Canal Server 发送消息到 Kafka 失败

10.2 Canal Admin 故障排除

10.2.1 Canal Admin 无法启动

10.2.2 Canal Admin 无法连接 Canal Server

10.2.3 Canal Admin 无法添加 Instance

10.3 诊断工具

10.3.1 Canal Server 诊断

10.3.2 Canal Admin 诊断

10.3.3 MySQL 诊断

10.3.4 Kafka 诊断

11. 最佳实践

11.1 性能优化