CDC 中间件 Canal 全解析
一、Canal 是什么
Canal（水道/管道）是阿里巴巴开源的一款 MySQL 数据库增量日志解析工具，提供增量数据订阅和消费的功能。它的核心能力是：实时感知 MySQL 数据变更，并将变更数据推送给下游系统。
一句话概括：Canal 就是一个伪装成 MySQL Slave 的程序，接收 Master 推送的 Binlog，解析后发给你。
在 CDC 链路中的位置：
MySQL → [Canal] → Kafka → 你的应用
Canal 就是这个链路中间的 "CDC中间件" 角色
注：
博客：
https://blog.csdn.net/badao_liumang_qizhi
二、Canal 解决什么问题
2.1 阿里的业务背景
早期阿里 B2B 公司有杭州和美国两个机房，需要做跨机房数据同步。最初采用的方案是业务层双写（写杭州的同时写美国），但这种方案问题很多：
- 性能差（跨洋网络延迟）
- 一致性难保证
- 侵入业务代码
后来改为基于 Binlog 的增量同步方案：只在一个机房写入，通过解析 Binlog 将变更同步到另一个机房。Canal 就是这个方案的核心组件。
2.2 典型应用场景
| 场景 | 说明 |
|---|---|
| 数据库镜像/备份 | 实时同步到备库 |
| 数据库实时同步到 ES | 数据变更后搜索引擎自动更新 |
| 数据库实时同步到 Redis | 缓存自动更新，不用手动失效 |
| 跨机房数据同步 | Canal 的诞生场景 |
| 业务事件驱动 | 数据变更触发下游业务逻辑（你的需求） |
| 数据分析/ETL | 实时数据入湖入仓 |
三、Canal 核心原理
3.1 MySQL 主从复制回顾
正常的主从复制：
┌────────────────┐ ┌────────────────┐
│ MySQL Master │ │ MySQL Slave │
│ │ │ │
│ 1.客户端写入 │ │ │
│ 2.生成Binlog │ 推送binlog │ 3.I/O Thread │
│ 3.Dump Thread │─────────────→│ 接收binlog │
│ 读取binlog │ │ 写入RelayLog │
│ │ │ 4.SQL Thread │
│ │ │ 回放SQL │
└────────────────┘ └────────────────┘
关键协议步骤：
- Slave 向 Master 发送
COM_REGISTER_SLAVE注册自己 - Slave 向 Master 发送
COM_BINLOG_DUMP请求 Binlog（携带位点信息） - Master 的 Dump Thread 持续推送 Binlog Event
3.2 Canal 的伪装原理
Canal 的做法：
┌────────────────┐ ┌────────────────┐
│ MySQL Master │ │ Canal │
│ │ │ (伪装成Slave) │
│ │ │ │
│ Master 认为 │ 推送binlog │ 1.注册为Slave │
│ 它只是又一个 │─────────────→│ 2.接收binlog │
│ 普通的Slave │ │ 3.解析二进制格式│
│ │ │ 4.转为结构化数据│
│ │ │ 5.推送给下游 │
└────────────────┘ └────────────────┘
│
▼
┌────────────────┐
│ Kafka / MQ / │
│ 直接回调应用 │
└────────────────┘
Canal 做了什么：
- 伪装：模拟 MySQL Slave 的通信协议，向 Master 注册
- 接收：Master 以为 Canal 是 Slave，持续推送 Binlog
- 解析：将二进制 Binlog Event 解析为可读的结构化数据
- 推送：将解析后的数据发送到 Kafka、RocketMQ 或让客户端主动拉取
Canal 不执行 SQL——真正的 Slave 会回放 SQL 来同步数据，但 Canal 不需要，它只需要读取和解析 Binlog 的内容。
四、Canal 架构设计
4.1 整体架构
┌──────────────────────────────────────────────────────────────────┐
│ Canal Server │
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ Canal Instance (一个实例对应一个MySQL实例的binlog订阅) │ │
│ │ │ │
│ │ ┌───────────┐ ┌───────────┐ ┌───────────┐ │ │
│ │ │ EventParser│ │EventSink │ │EventStore │ │ │
│ │ │ │ │ │ │ │ │ │
│ │ │ 连接MySQL │──→│ 过滤/路由 │──→│ 存储事件 │ │ │
│ │ │ 解析Binlog│ │ 归并处理 │ │ 环形缓冲区 │ │ │
│ │ └───────────┘ └───────────┘ └─────┬─────┘ │ │
│ │ │ │ │
│ └──────────────────────────────────────────┼───────────────────┘ │
│ │ │
│ ┌───────────────────────────────────────────┼──────────────────┐ │
│ │ ▼ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ Canal Client / MQ Producer │ │ │
│ │ │ │ │ │
│ │ │ 拉取EventStore中的数据 → 发送到 Kafka/RocketMQ/TCP │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └──────────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────────┘
4.2 核心组件详解
EventParser（事件解析器）
| 职责 | 说明 |
|---|---|
| 连接管理 | 与 MySQL Master 建立连接，发送 Dump 命令 |
| 心跳检测 | 定期发送心跳，检测连接是否存活 |
| Binlog 解析 | 将二进制 Binlog Event 解码为结构化对象 |
| 位点管理 | 记录消费到哪个位置（binlog文件+offset 或 GTID） |
| 断线重连 | 连接断开后自动重连，从上次位点继续 |
EventSink（事件沉淀器）
| 职责 | 说明 |
|---|---|
| 数据过滤 | 根据配置的正则表达式过滤库表（只关心指定表） |
| 数据归并 | 将同一事务内的多个变更归并为一组 |
| 数据路由 | 支持按规则路由到不同的下游 |
EventStore（事件存储）
| 职责 | 说明 |
|---|---|
| 环形缓冲区 | 基于内存的 RingBuffer，缓存解析后的事件 |
| 背压控制 | 缓冲区满时阻塞 Parser，防止 OOM |
| 位点确认 | Client 消费确认后释放空间 |
4.3 Canal 支持的下游投递模式
Canal Server
│
├──→ 模式1：TCP 直连（Canal Client 拉取）
│ 应用直接连接 Canal Server，主动 pull 数据
│ 适合：简单场景，消费者少
│
├──→ 模式2：Kafka 投递
│ Canal 解析后直接写入 Kafka Topic
│ 适合：多消费者、需要持久化、生产环境首选
│
├──→ 模式3：RocketMQ 投递
│ Canal 解析后写入 RocketMQ
│ 适合：阿里云环境
│
└──→ 模式4：RabbitMQ 投递
Canal 解析后写入 RabbitMQ
适合：中小规模场景
五、Canal 数据流转详细流程
5.1 从 MySQL 写入到 Canal 解析
时间轴：
t0 业务代码执行 INSERT INTO xxx (member_id, order_code, ...) VALUES (213681, 'CK001', ...)
│
t1 MySQL InnoDB 执行写入 → 写 redo log (prepare)
│
t2 MySQL 写入 Binlog → redo log (commit) → 事务完成
│
│ ← 从这里开始是Canal的工作 →
│
t3 Canal EventParser 的网络线程收到 MySQL 推送的 Binlog Event（二进制）
│
t4 Canal 解析二进制 Binlog Event：
│ ├── 解析 TableMapEvent → 确定是哪张表、列信息
│ ├── 解析 WriteRowsEvent → 提取出 INSERT 的每行数据
│ └── 组装为 Canal 内部的 Entry 对象
│
t5 EventSink 过滤：
│ ├── 配置了 filter = "aaa\\..*" （只关心 aaa 库的所有表）
│ ├── 当前事件表名 = aaa.xxx → 通过过滤
│ └── 写入 EventStore 环形缓冲区
│
t6 MQ Producer 从 EventStore 拉取 → 序列化为 JSON → 发送到 Kafka
│
t7 Kafka 接收消息 → 写入 topic: binlog-xxx-xxx-xxx 的某个 Partition
5.2 Canal 输出的 JSON 消息格式
Canal 发送到 Kafka 的消息长这样：
{
"database": "aaa",
"table": "xxx",
"type": "INSERT",
"ts": 1707552600000,
"isDdl": false,
"pkNames": ["id"],
"data": [
{
"id": "12345",
"member_id": "213681",
"warehouse_id": "100",
"order_code": "CK20250210001",
"reference_order_code": "SO20250210001",
"order_inout_type": "OUT",
"trans_qty": "5",
"create_time": "2025-02-10 14:30:00",
"update_time": "2025-02-10 14:30:00"
}
],
"old": null
}
注意：Canal 输出的所有字段值都是 String 类型（即使数据库中是 Integer）。这就是为什么在代码中需要 Integer.valueOf(memberIdObj.toString()) 做类型转换。
六、Canal 部署架构
6.1 单机部署
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ MySQL Master │────────→│ Canal Server │────────→│ Kafka │
│ │ binlog │ (单节点) │ 消息 │ │
└─────────────┘ └─────────────┘ └─────────────┘
问题：Canal Server 单点故障，挂了就不同步了。
6.2 高可用部署（HA）
┌─────────────────────────┐
│ ZooKeeper │
│ (选主 + 位点存储) │
└────────────┬────────────┘
│
┌──────────────────┼──────────────────┐
│ │ │
▼ ▼ ▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│Canal Server │ │Canal Server │ │Canal Server │
│ (Active) │ │ (Standby) │ │ (Standby) │
│ 正在工作 │ │ 待命 │ │ 待命 │
└──────┬──────┘ └─────────────┘ └─────────────┘
│
│ binlog
▼
┌─────────────┐
│ MySQL Master │
└─────────────┘
工作原理：
- 多个 Canal Server 同时运行，通过 ZooKeeper 选举一个 Active 节点
- 只有 Active 节点连接 MySQL 读取 Binlog
- Active 节点将消费位点（binlog position）存储到 ZooKeeper
- Active 宕机 → ZooKeeper 检测到 → Standby 被选为新 Active → 从 ZooKeeper 读取位点继续消费
6.3 多实例部署
┌─────────────┐ ┌──────────────────────────────────┐
│MySQL实例A │────────→│ Canal Instance A │
│(订单库) │ │ filter: aaa\\..* │──→ Kafka topic-A
└─────────────┘ └──────────────────────────────────┘
┌─────────────┐ ┌──────────────────────────────────┐
│MySQL实例B │────────→│ Canal Instance B │
│(库存库) │ │ filter: bbb\\..* │──→ Kafka topic-B
└─────────────┘ └──────────────────────────────────┘
一个 Canal Server 可以运行多个 Instance，每个 Instance 对应一个 MySQL 实例的 Binlog 订阅。
七、Canal 关键配置
7.1 canal.properties（Server 级配置）
# Canal Server 端口
canal.port = 11111
# 使用的数据存储方式：memory（内存）、file（文件）、default
canal.instance.store.mode = memory
canal.instance.store.buffer.size = 16384
# 消息投递模式：tcp / kafka / rocketmq / rabbitmq
canal.serverMode = kafka
# Kafka 配置
kafka.bootstrap.servers = 10.xxx.xxx.xxx: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.retries = 0
# ZooKeeper 地址（HA模式需要）
canal.zkServers = 127.0.0.1:2181
7.2 instance.properties（Instance 级配置）
# MySQL 连接信息
canal.instance.master.address = 127.0.0.1:3306
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal123
canal.instance.connectionCharset = UTF-8
# 起始消费位点（首次启动时）
canal.instance.master.journal.name = # binlog文件名，空表示从最新位置
canal.instance.master.position = # binlog偏移量
# 或者使用 GTID 模式
canal.instance.master.gtid =
# 表过滤：正则表达式
# 格式：schema.table，支持通配符
canal.instance.filter.regex = aaa\\..*
# 排除某些表
canal.instance.filter.black.regex =
# Kafka Topic 配置
canal.mq.topic = binlog-xx-xx-xxx
# 分区数
canal.mq.partitionsNum = 3
# 分区策略：按主键hash
canal.mq.partitionHash = aaa.xxx:id
7.3 MySQL 端需要的配置
-- 1. 开启 Binlog
[mysqld]
log-bin = mysql-bin
binlog-format = ROW -- 必须是 ROW 格式
binlog-row-image = FULL -- 记录完整行数据（不只是变更字段）
server-id = 1 -- 唯一的 server ID
-- 2. 创建 Canal 用户
CREATE USER 'canal'@'%' IDENTIFIED BY 'canal123';
-- 3. 授予 Binlog 读取权限
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;
八、Canal 位点管理（如何保证不丢数据）
8.1 什么是位点
位点就是 Canal 消费 Binlog 的进度标记，类似于书签，记录"我读到哪里了"。
Binlog 文件结构：
mysql-bin.000001: [Event1][Event2][Event3]...[EventN] ← 已满，切换到下一个文件
mysql-bin.000002: [Event1][Event2]...[Event M]
↑
Canal 位点：(mysql-bin.000002, offset=M)
"我读到第2个文件的第M个事件了"
8.2 位点存储
| 存储方式 | 说明 | 适用场景 |
|---|---|---|
| 内存 | 位点保存在内存中，重启丢失 | 测试环境 |
| 文件 | 位点写入本地文件 meta.dat | 单机部署 |
| ZooKeeper | 位点写入 ZK 节点 | HA 高可用部署 |
8.3 位点确认流程
1. Canal 从位点 P1 开始读取 Binlog
2. 读到一批 Event（位点到了 P2）
3. 解析 + 过滤 + 写入 EventStore
4. 下游客户端（或Kafka Producer）消费成功
5. 客户端发送 ACK（确认到 P2）
6. Canal 更新位点为 P2 并持久化
7. 下次从 P2 继续读取
如果步骤 4-5 之间 Canal 宕机：
→ 重启后从 P1 重新读取
→ 可能会有重复消息（At Least Once 语义）
→ 下游需要做幂等处理
九、Canal vs 其他 CDC 工具对比
| 维度 | Canal（阿里） | Debezium（Red Hat） | Maxwell（Zendesk） | Flink CDC |
|---|---|---|---|---|
| 语言 | Java | Java | Java | Java |
| 支持数据库 | MySQL/MariaDB | MySQL/PG/MongoDB/Oracle/SQLServer | 仅 MySQL | MySQL/PG/MongoDB |
| 输出目标 | Kafka/RocketMQ/RabbitMQ/TCP | 仅 Kafka | Kafka/Kinesis/Redis/stdout | Flink DataStream |
| 部署方式 | 独立进程 | Kafka Connect 插件 | 独立进程 | Flink Job |
| 高可用 | ZooKeeper 选主 | Kafka Connect 分布式 | 无内置HA | Flink 容错 |
| 社区 | 国内社区活跃 | 国际社区活跃 | 较小 | 国内外都活跃 |
| 数据格式 | 自定义 JSON | CloudEvents/Avro | JSON | Row 对象 |
| 学习成本 | 中等 | 较高（需了解Kafka Connect） | 低 | 高（需了解Flink） |
| 国内使用 | ⭐⭐⭐⭐⭐ 最广泛 | ⭐⭐⭐ 逐渐增多 | ⭐⭐ 较少 | ⭐⭐⭐⭐ 大数据场景多 |
选型建议：
- 国内 Java 项目、中等规模 → Canal（生态好、文档中文、坑少）
- 多数据源、国际化团队 → Debezium（功能全、标准化好）
- 快速验证 → Maxwell（零配置即可跑通）
- 实时计算场景 → Flink CDC（直接对接计算引擎）
十、通用代码示例：模拟 Canal 核心流程
以下示例模拟了 Canal 的核心工作原理（不依赖真实 MySQL），帮助理解其内部设计：
10.1 Binlog Event 模拟
package com.example.canal.core;
import java.util.*;
/**
* 模拟 MySQL Binlog Event（简化版）.
* 真实场景中这是二进制数据，Canal需要按MySQL协议解析.
*/
public class RawBinlogEvent {
private String binlogFileName;
private long offset;
private String database;
private String table;
private String eventType; // WRITE_ROWS / UPDATE_ROWS / DELETE_ROWS
private long timestamp;
private List<Map<String, String>> rows; // 变更后的行
private List<Map<String, String>> beforeRows; // 变更前的行（UPDATE时）
// ----- 构造工具方法 -----
public static RawBinlogEvent insertEvent(String db, String table,
Map<String, String> row) {
RawBinlogEvent event = new RawBinlogEvent();
event.database = db;
event.table = table;
event.eventType = "WRITE_ROWS";
event.timestamp = System.currentTimeMillis();
event.rows = Collections.singletonList(row);
return event;
}
public static RawBinlogEvent updateEvent(String db, String table,
Map<String, String> beforeRow,
Map<String, String> afterRow) {
RawBinlogEvent event = new RawBinlogEvent();
event.database = db;
event.table = table;
event.eventType = "UPDATE_ROWS";
event.timestamp = System.currentTimeMillis();
event.rows = Collections.singletonList(afterRow);
event.beforeRows = Collections.singletonList(beforeRow);
return event;
}
// getter/setter 省略
public String getDatabase() { return database; }
public String getTable() { return table; }
public String getEventType() { return eventType; }
public long getTimestamp() { return timestamp; }
public List<Map<String, String>> getRows() { return rows; }
public List<Map<String, String>> getBeforeRows() { return beforeRows; }
public String getBinlogFileName() { return binlogFileName; }
public long getOffset() { return offset; }
public void setBinlogFileName(String f) { this.binlogFileName = f; }
public void setOffset(long o) { this.offset = o; }
}
10.2 EventParser（模拟连接 MySQL 读取 Binlog）
package com.example.canal.parser;
import com.example.canal.core.RawBinlogEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* 模拟 Canal EventParser.
* 真实Canal中：通过MySQL协议接收binlog二进制流 → 解析为事件对象.
* 本示例：从模拟队列中获取事件（省略网络协议部分）.
*/
public class EventParser implements Runnable {
private static final Logger log = LoggerFactory.getLogger(EventParser.class);
private final String masterAddress;
private final String username;
private final String password;
private final AtomicBoolean running = new AtomicBoolean(false);
/**
* 模拟 MySQL 推送过来的 binlog 事件（真实场景是网络接收）.
*/
private final BlockingQueue<RawBinlogEvent> inputQueue = new LinkedBlockingQueue<>();
/**
* 解析后的事件输出到 Sink.
*/
private final BlockingQueue<RawBinlogEvent> outputQueue;
/**
* 当前消费位点.
*/
private volatile String currentBinlogFile = "mysql-bin.000001";
private volatile long currentOffset = 0;
public EventParser(String masterAddress, String username, String password,
BlockingQueue<RawBinlogEvent> outputQueue) {
this.masterAddress = masterAddress;
this.username = username;
this.password = password;
this.outputQueue = outputQueue;
}
/**
* 启动（模拟连接MySQL + Dump协议）.
*/
@Override
public void run() {
running.set(true);
log.info("EventParser启动, 连接MySQL: {}, 从位点 {}:{} 开始",
masterAddress, currentBinlogFile, currentOffset);
// 模拟：持续接收 binlog 事件
while (running.get()) {
try {
// 真实场景：从网络流中读取二进制binlog event
// 模拟：从输入队列中获取
RawBinlogEvent event = inputQueue.poll(1, java.util.concurrent.TimeUnit.SECONDS);
if (event == null) {
continue;
}
// 真实场景：二进制解析（Binlog Event Header + Body）
// 模拟：已经是结构化对象
// 更新位点
event.setBinlogFileName(currentBinlogFile);
event.setOffset(++currentOffset);
// 传递给 EventSink
outputQueue.put(event);
log.debug("Parser解析事件: {}.{} type={} offset={}",
event.getDatabase(), event.getTable(), event.getEventType(), currentOffset);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
log.info("EventParser已停止");
}
/**
* 模拟接收MySQL推送的binlog（实际由网络层完成）.
*/
public void simulateReceiveBinlog(RawBinlogEvent event) {
inputQueue.offer(event);
}
public void stop() {
running.set(false);
}
public String getCurrentPosition() {
return currentBinlogFile + ":" + currentOffset;
}
}
10.3 EventSink（过滤 + 路由）
package com.example.canal.sink;
import com.example.canal.core.RawBinlogEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Pattern;
/**
* 模拟 Canal EventSink.
* 职责：接收Parser解析后的事件 → 按正则过滤库表 → 传递给Store.
*/
public class EventSink implements Runnable {
private static final Logger log = LoggerFactory.getLogger(EventSink.class);
private final AtomicBoolean running = new AtomicBoolean(false);
/**
* 表过滤正则（例如 "ylh_stock\\..*" 表示只处理 ylh_stock 库的所有表）.
*/
private final Pattern filterPattern;
/**
* 从Parser接收事件.
*/
private final BlockingQueue<RawBinlogEvent> inputQueue;
/**
* 过滤后输出到Store.
*/
private final BlockingQueue<RawBinlogEvent> outputQueue;
private long filteredCount = 0;
private long passedCount = 0;
public EventSink(String filterRegex,
BlockingQueue<RawBinlogEvent> inputQueue,
BlockingQueue<RawBinlogEvent> outputQueue) {
this.filterPattern = Pattern.compile(filterRegex);
this.inputQueue = inputQueue;
this.outputQueue = outputQueue;
}
@Override
public void run() {
running.set(true);
log.info("EventSink启动, 过滤规则: {}", filterPattern.pattern());
while (running.get()) {
try {
RawBinlogEvent event = inputQueue.poll(1, TimeUnit.SECONDS);
if (event == null) {
continue;
}
// 构造匹配字符串：database.table
String fullTableName = event.getDatabase() + "." + event.getTable();
// 正则过滤
if (filterPattern.matcher(fullTableName).matches()) {
outputQueue.put(event);
passedCount++;
log.debug("Sink通过: {} type={}", fullTableName, event.getEventType());
} else {
filteredCount++;
log.debug("Sink过滤: {}", fullTableName);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
log.info("EventSink已停止, 通过: {}, 过滤: {}", passedCount, filteredCount);
}
public void stop() {
running.set(false);
}
}
10.4 EventStore（环形缓冲区）
package com.example.canal.store;
import com.example.canal.core.RawBinlogEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
/**
* 模拟 Canal EventStore.
* 真实Canal使用RingBuffer实现，这里用有界阻塞队列简化.
* 职责：缓存事件，供下游MQ Producer拉取.
*/
public class EventStore {
private static final Logger log = LoggerFactory.getLogger(EventStore.class);
/**
* 有界缓冲区（模拟RingBuffer）.
* 满了则阻塞Parser/Sink，形成背压.
*/
private final ArrayBlockingQueue<RawBinlogEvent> buffer;
/**
* 已确认的位点（下游ACK后更新）.
*/
private volatile long ackedOffset = 0;
public EventStore(int bufferSize) {
this.buffer = new ArrayBlockingQueue<>(bufferSize);
log.info("EventStore初始化, 缓冲区大小: {}", bufferSize);
}
/**
* 写入事件（Sink调用）.
* 缓冲区满时阻塞，形成背压传导到Parser.
*/
public void put(RawBinlogEvent event) throws InterruptedException {
buffer.put(event);
}
/**
* 批量拉取事件（MQ Producer调用）.
*
* @param batchSize 最多拉取条数
* @param timeout 等待超时时间
* @return 事件列表
*/
public List<RawBinlogEvent> get(int batchSize, long timeout, TimeUnit unit)
throws InterruptedException {
List<RawBinlogEvent> batch = new ArrayList<>();
// 至少等待一条
RawBinlogEvent first = buffer.poll(timeout, unit);
if (first != null) {
batch.add(first);
// 尽量多取，但不阻塞
buffer.drainTo(batch, batchSize - 1);
}
return batch;
}
/**
* 确认消费（下游ACK后调用）.
*/
public void ack(long offset) {
this.ackedOffset = offset;
log.debug("EventStore ACK, offset: {}", offset);
}
public int size() {
return buffer.size();
}
public long getAckedOffset() {
return ackedOffset;
}
}
10.5 MQ Producer（投递到 Kafka）
package com.example.canal.producer;
import com.alibaba.fastjson.JSON;
import com.example.canal.core.RawBinlogEvent;
import com.example.canal.store.EventStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* 模拟 Canal MQ Producer.
* 职责：从EventStore拉取事件 → 转换为下游格式 → 发送到Kafka.
*/
public class MqProducer implements Runnable {
private static final Logger log = LoggerFactory.getLogger(MqProducer.class);
private final AtomicBoolean running = new AtomicBoolean(false);
private final EventStore eventStore;
private final String topic;
private final int batchSize;
/**
* 模拟 Kafka 发送（真实场景使用 KafkaProducer）.
*/
private final List<String> sentMessages = new ArrayList<>();
public MqProducer(EventStore eventStore, String topic, int batchSize) {
this.eventStore = eventStore;
this.topic = topic;
this.batchSize = batchSize;
}
@Override
public void run() {
running.set(true);
log.info("MqProducer启动, topic: {}, batchSize: {}", topic, batchSize);
while (running.get()) {
try {
// 从Store批量拉取
List<RawBinlogEvent> batch = eventStore.get(batchSize, 1, TimeUnit.SECONDS);
if (batch.isEmpty()) {
continue;
}
// 转换格式 + 发送到Kafka
for (RawBinlogEvent event : batch) {
String message = convertToJson(event);
sendToKafka(topic, event.getDatabase() + "." + event.getTable(), message);
}
// 发送成功后ACK（确认位点）
RawBinlogEvent lastEvent = batch.get(batch.size() - 1);
eventStore.ack(lastEvent.getOffset());
log.debug("MqProducer发送批次完成, size: {}, lastOffset: {}",
batch.size(), lastEvent.getOffset());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
} catch (Exception e) {
log.error("MqProducer发送异常", e);
// 发送失败不ACK，下次重新拉取（At Least Once）
}
}
log.info("MqProducer已停止, 累计发送: {}条", sentMessages.size());
}
/**
* 将RawBinlogEvent转换为Canal标准JSON格式.
*/
private String convertToJson(RawBinlogEvent event) {
Map<String, Object> json = new LinkedHashMap<>();
json.put("database", event.getDatabase());
json.put("table", event.getTable());
json.put("isDdl", false);
json.put("pkNames", Collections.singletonList("id"));
json.put("ts", event.getTimestamp());
// 转换操作类型名称
switch (event.getEventType()) {
case "WRITE_ROWS":
json.put("type", "INSERT");
break;
case "UPDATE_ROWS":
json.put("type", "UPDATE");
break;
case "DELETE_ROWS":
json.put("type", "DELETE");
break;
default:
json.put("type", event.getEventType());
}
// data字段：所有值转为Object（Canal输出全部为String）
json.put("data", event.getRows());
// old字段：UPDATE时的旧值
if ("UPDATE_ROWS".equals(event.getEventType()) && event.getBeforeRows() != null) {
// Canal只输出变更的字段的旧值
List<Map<String, String>> oldData = new ArrayList<>();
for (int i = 0; i < event.getRows().size(); i++) {
Map<String, String> afterRow = event.getRows().get(i);
Map<String, String> beforeRow = event.getBeforeRows().get(i);
Map<String, String> changedFields = new HashMap<>();
for (Map.Entry<String, String> entry : beforeRow.entrySet()) {
String key = entry.getKey();
if (!Objects.equals(entry.getValue(), afterRow.get(key))) {
changedFields.put(key, entry.getValue());
}
}
oldData.add(changedFields);
}
json.put("old", oldData);
} else {
json.put("old", null);
}
return JSON.toJSONString(json);
}
/**
* 模拟发送到Kafka.
*/
private void sendToKafka(String topic, String key, String message) {
// 真实场景：kafkaProducer.send(new ProducerRecord<>(topic, key, message))
sentMessages.add(message);
log.info("发送到Kafka: topic={}, key={}, message={}", topic, key, message);
}
public void stop() {
running.set(false);
}
public List<String> getSentMessages() {
return sentMessages;
}
}
10.6 Canal Server 组装（串联所有组件）
package com.example.canal;
import com.example.canal.core.RawBinlogEvent;
import com.example.canal.parser.EventParser;
import com.example.canal.producer.MqProducer;
import com.example.canal.sink.EventSink;
import com.example.canal.store.EventStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
/**
* 模拟 Canal Server.
* 组装 Parser → Sink → Store → MqProducer 完整链路.
*/
public class CanalServer {
private static final Logger log = LoggerFactory.getLogger(CanalServer.class);
private final EventParser parser;
private final EventSink sink;
private final EventStore store;
private final MqProducer producer;
private final ExecutorService threadPool;
/**
* Parser → Sink 之间的内部队列.
*/
private final BlockingQueue<RawBinlogEvent> parserToSinkQueue;
public CanalServer(String mysqlAddress, String username, String password,
String filterRegex, String kafkaTopic) {
// 初始化各组件
this.parserToSinkQueue = new LinkedBlockingQueue<>(4096);
// Store 使用 Sink 的输出队列（这里简化为直接用Store的put）
this.store = new EventStore(8192);
// Sink 输出到 Store（通过一个适配队列）
BlockingQueue<RawBinlogEvent> sinkToStoreQueue = new LinkedBlockingQueue<RawBinlogEvent>(4096) {
@Override
public void put(RawBinlogEvent event) throws InterruptedException {
store.put(event);
}
};
this.parser = new EventParser(mysqlAddress, username, password, parserToSinkQueue);
this.sink = new EventSink(filterRegex, parserToSinkQueue, sinkToStoreQueue);
this.producer = new MqProducer(store, kafkaTopic, 100);
this.threadPool = Executors.newFixedThreadPool(3, r -> {
Thread t = new Thread(r);
t.setDaemon(true);
return t;
});
}
/**
* 启动Canal Server.
*/
public void start() {
log.info("Canal Server 启动...");
threadPool.submit(parser);
threadPool.submit(sink);
threadPool.submit(producer);
log.info("Canal Server 启动完成 (Parser + Sink + MqProducer)");
}
/**
* 停止Canal Server.
*/
public void stop() {
log.info("Canal Server 停止...");
parser.stop();
sink.stop();
producer.stop();
threadPool.shutdown();
log.info("Canal Server 已停止");
}
/**
* 模拟MySQL产生binlog事件（测试用）.
*/
public void simulateMysqlEvent(RawBinlogEvent event) {
parser.simulateReceiveBinlog(event);
}
public EventParser getParser() { return parser; }
public MqProducer getProducer() { return producer; }
}
10.7 完整演示（Main）
package com.example.canal;
import com.example.canal.core.RawBinlogEvent;
import java.util.HashMap;
import java.util.Map;
/**
* 演示：模拟Canal完整流程.
*
* 流程：MySQL变更 → Parser接收 → Sink过滤 → Store缓存 → Producer发到Kafka
*/
public class CanalDemo {
public static void main(String[] args) throws InterruptedException {
// 1. 创建并启动Canal Server
CanalServer canal = new CanalServer(
"127.0.0.1:3306", // MySQL地址
"canal", // 用户名
"xxx", // 密码
"bbb\\..*", // 只关心 bbb 库的所有表
"binlog-order-topic" // 输出到Kafka的topic
);
canal.start();
// 等待组件启动
Thread.sleep(1000);
// 2. 模拟MySQL产生binlog事件
// 模拟：INSERT INTO bbb.t_order (id, user_id, status, amount)
Map<String, String> insertRow = new HashMap<>();
insertRow.put("id", "1001");
insertRow.put("user_id", "5001");
insertRow.put("status", "CREATED");
insertRow.put("total_amount", "299.00");
insertRow.put("create_time", "2025-02-10 10:00:00");
canal.simulateMysqlEvent(
RawBinlogEvent.insertEvent("bbb", "t_order", insertRow)
);
// 模拟：UPDATE bbb.t_order SET status='PAID' WHERE id=1001
Map<String, String> beforeUpdate = new HashMap<>(insertRow);
Map<String, String> afterUpdate = new HashMap<>(insertRow);
afterUpdate.put("status", "PAID");
canal.simulateMysqlEvent(
RawBinlogEvent.updateEvent("bbb", "t_order", beforeUpdate, afterUpdate)
);
// 模拟：INSERT INTO other_db.t_log (不在过滤范围内，会被Sink过滤掉)
Map<String, String> logRow = new HashMap<>();
logRow.put("id", "9999");
logRow.put("content", "some log");
canal.simulateMysqlEvent(
RawBinlogEvent.insertEvent("other_db", "t_log", logRow)
);
// 3. 等待处理完成
Thread.sleep(3000);
// 4. 查看结果
System.out.println("\n====== 发送到Kafka的消息 ======");
for (String msg : canal.getProducer().getSentMessages()) {
System.out.println(msg);
}
// 预期输出：
// 只有 bbb.t_order 的 INSERT 和 UPDATE 两条消息
// other_db.t_log 被 Sink 过滤掉了
System.out.println("\n====== 当前位点 ======");
System.out.println("Parser位点: " + canal.getParser().getCurrentPosition());
// 5. 停止
canal.stop();
}
}
十一、总结
Canal 核心工作流程
MySQL Master
│ (推送binlog二进制流)
▼
EventParser ──── 伪装Slave，接收+解析二进制binlog
│ (结构化事件对象)
▼
EventSink ────── 正则过滤库表，只留下感兴趣的表
│ (过滤后的事件)
▼
EventStore ───── 环形缓冲区暂存，提供背压能力
│ (批量拉取)
▼
MQ Producer ──── 序列化为JSON，发送到Kafka/RocketMQ
│ (JSON消息)
▼
Kafka Topic ──── 持久化存储，等待下游消费
│
▼
你的应用 ─────── 消费 → 路由 → 业务处理
关键设计决策
| 决策 | Canal 的选择 | 原因 |
|---|---|---|
| 获取数据方式 | 伪装 Slave 而非轮询 | 实时性好、不增加 Master 负担 |
| 内部通信 | 队列（生产者-消费者模式） | 解耦各组件处理速度 |
| 过滤时机 | 在 Sink 层而非 Parser 层 | Parser 职责单一（只管解析），Sink 负责业务过滤 |
| 缓冲设计 | 有界环形缓冲区 | 防止 OOM，提供自然背压 |
| 位点确认 | 下游 ACK 后才更新 | 保证 At Least Once，不丢数据 |
| 高可用 | ZooKeeper 选主 | 简单成熟，位点共享方便 |
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/BADAO_LIUMANG_QIZHI/article/details/162846091
