跳到主要内容

MySQL + Canal + Kafka 数据同步方案

概述​

本方案详细介绍如何使用 Canal 将 MySQL 的 binlog 数据实时同步到 Kafka，实现 CDC（Change Data Capture，变更数据捕获）功能。这种架构广泛应用于数据同步、缓存更新、搜索索引构建、数据仓库等场景。

方案优势​

  1. 实时性：基于 MySQL binlog，实现毫秒级数据同步
  2. 低侵入：无需修改业务代码，通过 binlog 采集数据
  3. 高可靠：支持断点续传，保证数据不丢失
  4. 高性能：异步处理，不影响 MySQL 主业务
  5. 易扩展：Kafka 天然支持多消费者，便于下游扩展
  6. 解耦合：生产者和消费者完全解耦，提升系统灵活性

应用场景​

  • 缓存同步：MySQL 数据变更实时同步到 Redis
  • 搜索索引：MySQL 数据变更实时更新到 Elasticsearch
  • 数据仓库：实时将业务数据同步到数据仓库
  • 消息通知：数据变更触发业务通知
  • 数据分发：将数据分发给多个下游系统
  • 数据分析：实时数据分析与监控

目录​

  1. 架构设计
  2. 前置条件
  3. MySQL 配置
  4. Canal 部署
  5. Kafka 配置
  6. 数据同步验证
  7. 高可用方案
  8. 监控与告警
  9. 故障排除
  10. 最佳实践
  11. 附录

1. 架构设计​

1.0 部署架构说明​

本文档采用分布式部署架构，各组件独立部署在不同服务器上：

组件服务器 IP端口说明
MySQL192.168.1.1003306数据库服务器
Canal Server192.168.1.10011111Canal 部署在 MySQL 同服务器
Canal Adapter192.168.1.1008081Canal Adapter（可选）
Kafka 节点 1192.168.1.1019092Kafka Broker 1
Kafka 节点 2192.168.1.1029092Kafka Broker 2
Kafka 节点 3192.168.1.1039092Kafka Broker 3

注意：以上 IP 地址仅为示例，请根据实际环境修改。

1.1 整体架构​

1.2 Canal 架构​

1.3 数据流向​

MySQL → binlog → Canal Server → Canal Adapter → Kafka → 消费者

详细流程：

  1. MySQL 产生数据变更（INSERT/UPDATE/DELETE）
  2. MySQL 将变更记录到 binlog
  3. Canal Server 模拟 MySQL Slave，拉取 binlog
  4. Canal Server 解析 binlog，转换为结构化事件
  5. Canal Adapter 将事件转换为 Kafka 消息格式
  6. Kafka 接收并存储消息
  7. 消费者从 Kafka 拉取消息并处理

2. 前置条件​

2.1 环境要求​

组件版本要求说明
MySQL5.6.x / 5.7.x / 8.0.x需要开启 binlog
Canal1.1.6+推荐使用最新稳定版
Kafka2.8+ / 3.0+推荐 KRaft 模式
JDK1.8+Canal 和 Kafka 运行环境
Zookeeper3.5+Kafka 传统模式需要，KRaft 模式不需要

2.2 硬件配置​

组件最小配置推荐配置
MySQL2核 4GB4核 8GB
Canal2核 2GB4核 4GB
Kafka2核 4GB4核 8GB

2.3 网络要求​

  • MySQL、Canal、Kafka 之间网络互通
  • 防火墙开放必要端口：
    • MySQL: 3306
    • Canal: 11111
    • Kafka: 9092, 9093

3. MySQL 配置​

3.1 开启 binlog​

编辑 MySQL 配置文件 /etc/my.cnf：

[mysqld]
# 开启 binlog
log-bin = /usr/local/mysql/binlog/mysql-bin.log
binlog_format = ROW
binlog_row_image = FULL
expire_logs_days = 7
max-binlog-size = 100M

# 设置 server-id（必须唯一）
server-id = 1

# GTID 模式（可选，推荐）
gtid-mode = ON
enforce-gtid-consistency = ON

# binlog 缓存
binlog_cache_size = 2M
sync_binlog = 1

# 忽略的数据库（可选）
binlog-ignore-db = mysql
binlog-ignore-db = information_schema
binlog-ignore-db = performance_schema

重要配置说明：

  • binlog_format = ROW：必须使用 ROW 格式，Canal 才能解析
  • binlog_row_image = FULL：记录完整的行数据，推荐配置
  • server-id：每个 MySQL 实例的 server-id 必须唯一
  • gtid-mode：开启 GTID 可以简化主从切换

3.2 重启 MySQL​

# 重启 MySQL 服务
systemctl restart mysql

# 验证 binlog 是否开启
mysql -uroot -p -e "SHOW VARIABLES LIKE 'log_bin';"
mysql -uroot -p -e "SHOW VARIABLES LIKE 'binlog_format';"

输出示例：

+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| log_bin       | ON    |
+---------------+-------+

+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| binlog_format | ROW   |
+---------------+-------+

3.3 创建 Canal 用户​

-- 登录 MySQL
mysql -uroot -p

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

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

-- 刷新权限
FLUSH PRIVILEGES;

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

-- 退出
QUIT;

权限说明：

  • SELECT：查询表结构和数据
  • REPLICATION SLAVE：读取 binlog
  • REPLICATION CLIENT：查看 binlog 位置

3.4 创建测试数据库和表​

-- 创建测试数据库
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;

4. Canal 部署​

4.1 下载 Canal​

# 下载 Canal
cd /opt
wget https://github.com/alibaba/canal/releases/download/canal-1.1.7/canal.deployer-1.1.7.tar.gz

# 解压
mkdir -p /opt/canal
tar -zxvf canal.deployer-1.1.7.tar.gz -C /opt/canal

# 创建用户
useradd -m -s /bin/bash canal
chown -R canal:canal /opt/canal

4.2 配置 Canal Server​

4.2.1 配置 canal.properties​

编辑 /opt/canal/conf/canal.properties：

# Canal Server 配置

# 注册中心配置（可选）
canal.register.ip = 192.168.1.100
canal.port = 11111
canal.metrics.pull.port = 11112

# Canal Admin 配置（可选）
canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110

# 实例配置
canal.destinations = example
canal.conf.dir = ../conf
canal.auto.scan = true
canal.auto.scan.interval = 5

# 存储配置
canal.instance.global.spring.xml = classpath:spring/file-instance.xml

# 数据库配置
canal.instance.master.address = 192.168.1.100:3306
canal.instance.master.journal.name =
canal.instance.master.position =
canal.instance.master.timestamp =
canal.instance.master.gtid =

# 数据库用户名密码
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal_password

# 数据库连接参数
canal.instance.connectionCharset = UTF-8
canal.instance.defaultDatabaseName = test_db
canal.instance.connectionProperties = socketTimeout=180000

# binlog 过滤
canal.instance.filter.regex = test_db\\..*
canal.instance.filter.black.regex = mysql\\.mysql_.*|information_schema\\..*

# binlog 解析配置
canal.instance.binlog.format = ROW,STATEMENT,MIXED
canal.instance.binlog.image = FULL,MINIMAL,NOBLOB

# 消息队列配置
canal.mq.topic = example
canal.mq.partition = 0
canal.mq.partitionsNum = 1
canal.mq.partitionHash = test_db.user:id,.*\\..*

配置说明：

  • canal.destinations：实例名称，对应配置目录
  • canal.instance.master.address：MySQL 地址
  • canal.instance.filter.regex：binlog 过滤规则（正则表达式）
  • canal.mq.topic：Kafka Topic 名称

4.2.2 配置实例​

编辑 /opt/canal/conf/example/instance.properties：

# Canal Instance 配置

# MySQL 配置
canal.instance.master.address=192.168.1.100:3306
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=

# 数据库认证
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal_password

# 连接参数
canal.instance.connectionCharset=UTF-8
canal.instance.defaultDatabaseName=test_db
canal.instance.connectionProperties=socketTimeout=180000

# binlog 过滤
canal.instance.filter.regex=test_db\\..*
canal.instance.filter.black.regex=mysql\\.mysql_.*|information_schema\\..*

# binlog 解析
canal.instance.binlog.format=ROW,STATEMENT,MIXED
canal.instance.binlog.image=FULL,MINIMAL,NOBLOB

# 消息队列配置
canal.mq.topic=example
canal.mq.partition=0
canal.mq.partitionsNum=1
canal.mq.partitionHash=test_db.user:id,.*\\..*

4.3 启动 Canal Server​

# 切换到 canal 用户
su - canal

# 启动 Canal
cd /opt/canal/bin
./startup.sh

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

4.4 配置 Systemd 服务​

创建 /etc/systemd/system/canal.service：

[Unit]
Description=Canal Server
After=network.target mysql.service

[Service]
Type=forking
User=canal
Group=canal
Environment="JAVA_HOME=/usr/lib/jvm/java-17-openjdk"
Environment="CANAL_HEAP_OPTS=-Xms2g -Xmx2g"
ExecStart=/opt/canal/bin/startup.sh
ExecStop=/opt/canal/bin/stop.sh
Restart=on-failure
RestartSec=10

[Install]
WantedBy=multi-user.target

启用并启动服务：

# 重新加载 systemd
systemctl daemon-reload

# 启用开机自启
systemctl enable canal

# 启动服务
systemctl start canal

# 查看服务状态
systemctl status canal

5. Kafka 配置​

5.1 创建 Topic​

假设 Kafka 集群已独立部署在以下服务器：

  • Kafka 节点 1: 192.168.1.101:9092
  • Kafka 节点 2: 192.168.1.102:9092
  • Kafka 节点 3: 192.168.1.103:9092

在任意一台 Kafka 服务器上创建 Topic：

# 切换到 kafka 用户（在 Kafka 服务器上执行）
su - kafka

# 创建 Topic（3 分区，3 副本）
cd /opt/kafka/kafka

./bin/kafka-topics.sh --create \
  --bootstrap-server 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092 \
  --topic example \
  --partitions 3 \
  --replication-factor 3

# 查看 Topic
./bin/kafka-topics.sh --list --bootstrap-server 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092

# 查看 Topic 详情
./bin/kafka-topics.sh --describe --bootstrap-server 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092 --topic example

注意：如果 Kafka 集群使用 KRaft 模式，可以只指定一个 broker 地址即可。

5.2 配置 Canal Adapter（可选）​

Canal Adapter 可以将数据转换为 JSON 格式并发送到 Kafka。

5.2.1 下载 Canal Adapter​

# 下载 Canal Adapter
cd /opt
wget https://github.com/alibaba/canal/releases/download/canal-1.1.7/canal.adapter-1.1.7.tar.gz

# 解压
mkdir -p /opt/canal-adapter
tar -zxvf canal.adapter-1.1.7.tar.gz -C /opt/canal-adapter

# 创建用户
useradd -m -s /bin/bash canal-adapter
chown -R canal-adapter:canal-adapter /opt/canal-adapter

5.2.2 配置 application.yml​

编辑 /opt/canal-adapter/conf/application.yml：

server:
  port: 8081

spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8

canal.conf:
  mode: tcp
  flatMessage: true
  zookeeperHosts:
  syncBatchSize: 1000
  retries: 0
  timeout:
  accessKey:
  secretKey:
  consumerProperties:
    canal.tcp.server.host: 127.0.0.1
    canal.tcp.server.port: 11111
    canal.tcp.batch.size: 500
    canal.tcp.username:
    canal.tcp.password:
  canalAdapters:
    - instance: example
      groups:
        - groupId: default
          outerAdapters:
            - name: kafka
              key: kafka1
              properties:
                # Kafka 集群地址（根据实际环境修改）
                kafka.servers: 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092
                kafka.topic: example
                partition: 0
                partitionCount: 3
                format: json

5.2.3 配置映射规则​

编辑 /opt/canal-adapter/conf/rdb/mytest_user.yml：

dataSourceKey: defaultDS
destination: example
groupId: g1
outerAdapterKey: kafka1
concurrent: true
dbMapping:
  database: test_db
  table: user
  targetTable: user
  targetPk:
    id: id
  mapAll: true
  targetColumns:
    id:
    username:
    email:
    age:
    created_at:
    updated_at:
  etlCondition: " where id>{}"
  commitBatch: 3000

5.2.4 启动 Canal Adapter​

# 切换到 canal-adapter 用户
su - canal-adapter

# 启动 Canal Adapter
cd /opt/canal-adapter/bin
./startup.sh

# 查看日志
tail -f /opt/canal-adapter/logs/adapter/adapter.log

5.3 消费 Kafka 消息​

5.3.1 使用命令行消费​

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

# 消费消息
./bin/kafka-console-consumer.sh \
  --bootstrap-server 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092 \
  --topic example \
  --from-beginning

5.3.2 使用 Java 消费​

创建 Maven 项目，添加依赖：

<dependencies>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>3.7.0</version>
    </dependency>
</dependencies>

创建消费者代码：

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class CanalKafkaConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        // Kafka 集群地址（根据实际环境修改）
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "canal-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("example"));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("offset = %d, key = %s, value = %s%n",
                            record.offset(), record.key(), record.value());
                }
            }
        } finally {
            consumer.close();
        }
    }
}

6. 数据同步验证​

6.1 测试数据变更​

-- 登录 MySQL
mysql -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;

6.2 查看 Canal 日志​

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

日志示例：

2024-01-12 10:00:00.123 [main] INFO  c.a.o.c.i.core.CanalInstanceWithManager - [0] init successful, detail configs CanalInstance {
    "instanceName" : "example",
    "database" : "test_db",
    "tableRegex" : "test_db\\..*"
}
2024-01-12 10:00:01.456 [main] INFO  c.a.o.c.i.core.CanalInstanceWithManager - [0] successful start canal instance : example

6.3 查看 Kafka 消息​

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

# 消费 Kafka 消息
./bin/kafka-console-consumer.sh \
  --bootstrap-server 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092 \
  --topic example \
  --from-beginning

消息示例（JSON 格式）：

{
  "data": [
    {
      "id": "4",
      "username": "user4",
      "email": "user4@example.com",
      "age": "35",
      "created_at": "2024-01-12 10:00:00",
      "updated_at": "2024-01-12 10:00:00"
    }
  ],
  "database": "test_db",
  "es": 1705036800000,
  "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": 1705036800123,
  "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. 高可用方案​

7.1 Canal HA 方案​

7.1.1 Canal Server HA​

使用 Zookeeper 实现 Canal Server HA：

# canal.properties
canal.zkServers = 192.168.1.100:2181,192.168.1.101:2181,192.168.1.102:2181
canal.zk.flush.timeout = 3000
canal.withoutNetty = false

# instance.properties
canal.instance.master.address = 192.168.1.100:3306,192.168.1.101:3306,192.168.1.102:3306
canal.instance.standby.address = 192.168.1.101:3306,192.168.1.102:3306

7.1.2 Canal Adapter HA​

配置多个 Canal Adapter 实例：

canal.conf:
  canalAdapters:
    - instance: example
      groups:
        - groupId: default
          outerAdapters:
            - name: kafka
              key: kafka1
              properties:
                kafka.servers: 192.168.1.101:9092,192.168.1.102:9092,192.168.1.103:9092
                kafka.topic: example
                partition: 0
                partitionCount: 3

7.2 Kafka HA 方案​

参考 Kafka 虚拟机部署指南 中的生产环境部署章节，部署 3 节点高可用 Kafka 集群。

7.3 MySQL HA 方案​

参考 MySQL 高可用故障转移（双主热备） 文档，部署 MySQL 主从复制或双主热备。

8. 监控与告警​

8.1 Canal 监控​

8.1.1 Canal Admin​

部署 Canal Admin 管理平台：

# 下载 Canal Admin
cd /opt
wget https://github.com/alibaba/canal/releases/download/canal-1.1.7/canal.admin-1.1.7.tar.gz

# 解压
mkdir -p /opt/canal-admin
tar -zxvf canal.admin-1.1.7.tar.gz -C /opt/canal-admin

# 配置 application.yml
vi /opt/canal-admin/conf/application.yml

配置内容：

server:
  port: 8089

spring:
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8

spring.datasource:
  address: 127.0.0.1:3306
  database: canal_manager
  username: root
  password: root
  driver-class-name: com.mysql.jdbc.Driver
  url: jdbc:mysql://127.0.0.1:3306/canal_manager?useUnicode=true&characterEncoding=UTF-8&useSSL=false
  hikari:
    maximum-pool-size: 30
    minimum-idle: 1

canal:
  adminUser: admin
  adminPasswd: admin

启动 Canal Admin：

cd /opt/canal-admin/bin
./startup.sh

访问 Canal Admin：http://<canal-admin-ip>:8089

8.1.2 Canal 指标​

Canal 提供以下监控指标：

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

8.2 Kafka 监控​

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

8.3 告警规则​

创建 Prometheus 告警规则 canal-alerts.yml：

groups:
  - name: canal
    interval: 30s
    rules:
      - alert: CanalDown
        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. 故障排除​

9.1 常见问题​

9.1.1 Canal 无法连接 MySQL​

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

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

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

# 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. 网络不通
telnet <mysql-host> 3306

9.1.2 Canal 解析 binlog 失败​

问题：Canal 无法解析 binlog

# 查看日志
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

9.1.3 Kafka 消息丢失​

问题：Kafka 消息丢失

# 检查 Canal 日志
tail -f /opt/canal/logs/example/example.log

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

# 常见原因：
# 1. Kafka 副本数不足
# 解决：增加副本数

# 2. Kafka 磁盘满
# 解决：清理 Kafka 日志

# 3. Canal 发送失败
# 解决：检查 Canal 配置

9.2 诊断工具​

9.2.1 Canal 诊断​

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

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

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

9.2.2 MySQL 诊断​

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

# 查看 binlog 内容
mysqlbinlog --start-position=4 /usr/local/mysql/binlog/mysql-bin.000001

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

9.2.3 Kafka 诊断​

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

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

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

10. 最佳实践​

10.1 性能优化​

10.1.1 Canal 性能优化​

# canal.properties
# 增加 Canal 处理线程数
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

10.1.2 Kafka 性能优化​

# server.properties
# 增加网络线程数
num.network.threads = 8
num.io.threads = 16

# 增加批处理大小
batch.size = 16384
linger.ms = 5

10.2 安全配置​

10.2.1 MySQL 安全​

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

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

10.2.2 Kafka 安全​

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

10.3 数据一致性​

10.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);
    }
}

10.3.2 事务处理​

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

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

    // 处理数据
    updateDatabase(json);

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

10.4 监控指标​

10.4.1 关键指标​

  • Canal 解析延迟：< 1 秒
  • Canal 投递延迟：< 1 秒
  • Kafka 消息延迟：< 5 秒
  • Kafka 消费延迟：< 10 秒
  • MySQL binlog 延迟：< 1 秒

10.4.2 告警阈值​

  • Canal 解析延迟 > 10 秒：Warning
  • Canal 投递延迟 > 10 秒：Warning
  • Kafka 消息延迟 > 30 秒：Warning
  • Kafka 消费延迟 > 60 秒：Critical
  • Canal 连接失败：Critical

11. 附录​

11.1 配置参数说明​

11.1.1 Canal 参数​

参数默认值说明
canal.instance.master.address-MySQL 地址
canal.instance.dbUsername-MySQL 用户名
canal.instance.dbPassword-MySQL 密码
canal.instance.filter.regex.*binlog 过滤规则
canal.mq.topic-Kafka Topic 名称
canal.mq.partitionsNum1Kafka 分区数

11.1.2 MySQL 参数​

参数默认值说明
log-bin-binlog 文件路径
binlog_formatROWbinlog 格式
binlog_row_imageFULLbinlog 行镜像
server-id-服务器 ID
gtid-modeOFFGTID 模式

11.2 常用命令参考​

11.2.1 Canal 命令​

# 启动 Canal
/opt/canal/bin/startup.sh

# 停止 Canal
/opt/canal/bin/stop.sh

# 重启 Canal
/opt/canal/bin/restart.sh

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

11.2.2 MySQL 命令​

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

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

# 查看 binlog 格式
mysql -uroot -p -e "SHOW VARIABLES LIKE 'binlog_format';"

11.2.3 Kafka 命令​

# 创建 Topic
./bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --topic example --partitions 3 --replication-factor 3

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

# 消费消息
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic example --from-beginning

11.3 官方资源链接​

11.4 参考资料​

总结​

通过本方案，你已经成功搭建了 MySQL + Canal + Kafka 的数据同步系统。该方案具有以下特点：

  1. 实时性：基于 binlog，实现毫秒级数据同步
  2. 可靠性：支持断点续传，保证数据不丢失
  3. 扩展性：Kafka 天然支持多消费者，便于扩展
  4. 解耦合：生产者和消费者完全解耦

下一步建议​

  1. 根据实际业务需求调整 Canal 和 Kafka 配置
  2. 配置监控和告警，及时发现和处理问题
  3. 实现消费者幂等性处理，避免重复消费
  4. 根据数据量选择合适的分区数和副本数
  5. 定期备份 MySQL 和 Kafka 数据

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