利用 Canal 构建 MySQL 增量数据同步方案
1. Canal 技术概览
Canal 是阿里巴巴研发的一款开源项目，旨在模拟 MySQL 数据库从库（Slave）的行为。它通过接入 MySQL 主库的 `binlog` 二进制日志，实现对数据变更事件的实时捕获与解析。其核心能力在于将非结构化的 `binlog` 转换为标准化的 JSON 格式对象，从而支持下游系统（如消息队列、搜索引擎或缓存服务）进行高效消费。
选择该方案的显著优势包括：
- 低延迟响应：直接读取磁盘上的 `binlog`，无需查询业务库，同步时延通常控制在毫秒级。
- 高吞吐处理：采用流式处理机制，并支持批量拉取事件，有效降低网络 IO 开销。
- 断点恢复能力：消费者位点信息持久化在 ZooKeeper 或本地文件中，进程重启后可自动从中断处继续消费。
- 灵活性：提供丰富的过滤规则和多种适配器（Adapter），支持对接 Kafka、Elasticsearch、MySQL 等多种目标端。
1.1 核心组件与工作流
Canal 的运作依赖于 MySQL 原有的主从复制协议机制。
- MySQL Master：记录所有数据变动到二进制日志文件。
- Canal Server：伪装成 MySQL Slave，发送 dump 请求接收 Binlog 流。
- Canal Adapter：可选组件，负责将解析后的数据快速投递至异构存储。
- 下游应用：自定义 Client 代码，监听 Server 推送的事件并进行业务逻辑处理。
- 开启 Binlog：执行 `SHOW VARIABLES LIKE 'log_bin';`，状态必须为 ON。
- 指定格式：将 `binlog_format` 设置为
ROW模式，否则无法精确获取行级别变更。 - 权限授权：创建专门账号并授予 `REPLICATION SLAVE`、`REPLICATION CLIENT` 及 `SUPER` 权限。
- JDK 版本：建议 JDK 1.8 及以上版本。
2. 环境部署与配置指南
2.1 前置条件检查
在正式安装前，需确保源数据库满足以下要求：
修改 MySQL 配置文件（如 /etc/my.cnf）：
[mysqld]
log-bin=mysql-bin # 启用日志功能
binlog-format=ROW # 行模式格式
server-id=10 # 确保服务器 ID 全局唯一
初始化专用连接用户：
Create User 'sync_user'@'%' identified by 'StrongPass123!';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT, SUPER ON *.* TO 'sync_user'@'%';
FLUSH PRIVILEGES;
2.2 安装解压
前往官方 Release 页面下载最新版本压缩包。推荐同时下载 Server 和 Adapter 包以覆盖更多场景。
# 创建目录
mkdir -p /opt/canal-server
mkdir -p /opt/canal-adapter
# 解压 Server
tar zxvf canal.deployer-x.x.x.tar.gz -C /opt/canal-server/
# 解压 Adapter
tar zxvf canal.adapter-x.x.x.tar.gz -C /opt/canal-adapter/
2.3 关键配置项
Canal 采用分层配置管理，分为全局参数和实例参数。
全局配置 (canal.properties)
位于 conf/canal.properties，定义服务端的基础设施参数。
canal.id = 2024
canal.instance.master.address = 192.168.1.100:3306
canal.zkServers =
canal.destinations = log_sync_default
实例配置 (instance.properties)
位于 conf/example/instance.properties（文件夹名需与 destinations 对应），定义具体数据源的连接细节。
canal.instance.filter.regex = .*\\..*
canal.instance.dbUsername = sync_user
canal.instance.dbPassword = StrongPass123!
# 这里的 ID 不能与 MySQL server-id 冲突
canal.instance.slaveId = 555
注意：filter.regex允许正则匹配，若只监特定表可写为 test_db\\.user_info。
Adapter 适配层配置
如需直连 ES 等存储，需在 application.yml 中指向正确的 Server 地址，并在 conf/adapter/conf/mysql.yml 下映射字段。
esMapping:
_index: business_orders
_id: order_id
sql: "SELECT id AS order_id, amount FROM t_order"
3. SpringBoot 深度集成实践
除了使用现成的 Adapter，我们也可以在业务项目中引入 Canel Client SDK，自行编写消费逻辑。
3.1 依赖引入
在 pom.xml 中添加如下依赖：
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.1.6</version>
</dependency>
3.2 模型定义与持久层
定义一个代表订单数据的实体类，用于接收转换后的数据。
@Data
@NoArgsConstructor
@AllArgsConstructor
public class BusinessOrder {
private Long orderId; // 订单主键
private String bizNo; // 业务流水号
private BigDecimal value; // 金额
private String currentStatus; // 当前状态
}
对应的数据访问接口示例：
@Mapper
public interface OrderAccess {
@Insert("INSERT INTO business_table(id, biz_no, value, status) VALUES(#{item.orderId}, #{item.bizNo}, #{item.value}, #{item.currentStatus})")
void persist(@Param("item") BusinessOrder item);
@Update("UPDATE business_table SET status=#{item.currentStatus}, value=#{item.value} WHERE biz_no=#{item.bizNo}")
void modify(@Param("item") BusinessOrder item);
}
3.3 监听器实现
创建一个 Spring Component，注入 Connector 进行循环拉取。为了保持代码清晰，将具体的解析逻辑分离出去。
@Component
@Slf4j
public class BinlogSyncTask {
@Resource
private OrderAccess orderAccess;
private static final int BATCH_SIZE = 500;
@PostConstruct
public void init() throws Exception {
CanalConnector connector = CanalConnectors.newClusterConnector(
new ArrayList<>(Arrays.asList("127.0.0.1:11111")),
"log_sync_default", "", ""
);
try {
connector.connect();
connector.subscribe(".*\\..*");
while (true) {
Message msg = connector.getWithoutAck(BATCH_SIZE);
long batchId = msg.getId();
if (msg.getId() == -1 || msg.getEntries().isEmpty()) {
Thread.sleep(1000);
continue;
}
processBatch(msg.getEntries());
connector.ack(batchId);
}
} finally {
connector.disconnect();
}
}
private void processBatch(List<Entry> entryList) {
try {
for (Entry entry : entryList) {
if (entry.getEntryType() != ENTRY_TYPE_ROLLINGFILE &&
entry.getEntryType() != ENTRY_TYPE_TRANSACTIONBEGIN &&
entry.getEntryType() != ENTRY_TYPE_TRANSACTIONEND) {
RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
handleRowChanges(rowChange);
}
}
} catch (Exception e) {
logger.error("解析 binlog 异常", e);
}
}
private void handleRowChanges(RowChange change) {
List<RowData> dataList = change.getRowDatasList();
for (RowData rowData : dataList) {
BusinessOrder record = parseColumns(rowData.getAfterColumnsList());
if (change.getEventType() == EventType.DELETE) {
// 逻辑省略，视业务需求处理删除
} else if (change.getEventType() == EventType.INSERT) {
orderAccess.persist(record);
} else {
orderAccess.modify(record);
}
}
}
private BusinessOrder parseColumns(List<Column> cols) {
BusinessOrder target = new BusinessOrder();
for (Column col : cols) {
String fieldName = col.getName();
Object value = col.getValue();
if ("orderId".equals(fieldName)) {
target.setOrderId(Long.parseLong(value));
} else if ("bizNo".equals(fieldName)) {
target.setBizNo(String.valueOf(value));
} else if ("value".equals(fieldName)) {
target.setValue(new BigDecimal(String.valueOf(value)));
} else if ("currentStatus".equals(fieldName)) {
target.setCurrentStatus(String.valueOf(value));
}
}
return target;
}
}
3.4 验证与监控
完成部署后，可以通过 HTTP 接口触发业务更新，观察控制台输出日志确认同步流程。
插入测试命令：
POST http://localhost:8080/api/orders/create
Content-Type: application/json
{
"bizNo": "ORD_20231001_001",
"value": "999.00",
"currentStatus": "WAIT_PAY"
}
预期日志输出应包含类似以下内容：
================> binlog[mysql-bin.000003:20481], name[test_db, business_table], eventType : INSERT
OrderId: 8821 BizNo: ORD_20231001_001 Value: 999.00 Status: WAIT_PAY
此时检查目标应用数据库，应能发现新增了一条 business_table 记录，且时间与源库操作时间接近。
