当前位置：首页 > news >正文

Canal Instance：深入解析与实战指南

news 2026/7/14 22:29:19

1. Canal 简介

Canal 是阿里巴巴开源的一款基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费的中间件。其核心原理是模拟 MySQL Slave 的交互协议，将自己伪装成一个 MySQL 从库，向主库发送 dump 请求，从而实时获取主库的二进制日志（binlog）变更。

一个 Canal Instance 是 Canal 服务中一个独立的数据同步任务单元，负责监听一个特定的 MySQL 数据源，并按照配置的规则解析和投递 binlog 数据。

2. Canal Instance 核心概念

2.1 Instance 的生命周期

一个 Instance 从启动到停止，会经历以下状态：

启动 (Start)：加载配置文件，建立与 MySQL 的连接，开始监听 binlog。
运行 (Running)：持续解析 binlog 事件，并将数据投递给下游消费者。
暂停 (Pause)：临时停止数据投递，但保持与 MySQL 的连接。
停止 (Stop)：断开与 MySQL 的连接，释放所有资源。

2.2 核心配置文件

每个 Instance 对应一个独立的配置文件，通常命名为instance.properties或{destination}.properties。主要配置项包括：

# 数据源配置 canal.instance.master.address=127.0.0.1:3306 canal.instance.dbUsername=canal canal.instance.dbPassword=canal canal.instance.defaultDatabaseName=test_db 过滤规则 canal.instance.filter.regex=.\.. canal.instance.filter.black.regex= 投递配置 canal.instance.mq.topic=example-topic

3. 部署与启动一个 Instance

3.1 单机部署

在 Canal 单机模式下，一个服务进程可以运行多个 Instance。部署步骤如下：

下载与解压：从 GitHub 获取 Canal 发布包。
修改配置文件：在conf目录下，复制example文件夹并重命名为你的 Instance 名称（例如order_instance），然后修改其中的instance.properties。
启动 Canal Server：执行bin/startup.sh（Linux）或bin/startup.bat（Windows）。
验证启动：查看日志文件logs/{destination}/log.log，确认无报错且出现 “start successful” 字样。

3.2 集群部署（HA模式）

为保证高可用，Canal 支持集群部署。多个 Canal Server 可以同时运行同一个 Instance，但同一时刻只有一个处于运行状态（Leader），其他处于待命状态（Standby）。当 Leader 宕机时，ZooKeeper 会协调选举出新的 Leader 接管。

关键配置：

# 启用集群模式 canal.instance.global.mode = spring canal.instance.global.lazy = false canal.instance.global.manager.address = ${zookeeper集群地址} canal.instance.global.spring.xml = classpath:spring/default-instance.xml

4. 客户端消费数据

Canal 支持多种消费方式，最常见的是通过 Java 客户端直连消费。

import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.protocol.Message; import java.net.InetSocketAddress; public class SimpleCanalClient { public static void main(String[] args) { // 1. 创建连接器 CanalConnector connector = CanalConnectors.newSingleConnector( new InetSocketAddress("127.0.0.1", 11111), "example", // Instance 的 destination 名称 "", "" ); try { connector.connect(); connector.subscribe(".*\\..*"); // 订阅所有表 connector.rollback(); // 回滚到未消费的位置 while (true) { // 2. 获取批量消息 Message message = connector.getWithoutAck(100); // 每次获取100条 long batchId = message.getId(); if (batchId != -1) { // 3. 处理消息 message.getEntries().forEach(entry -> { // 解析 entry，处理增删改数据 System.out.println(entry); }); // 4. 确认消费成功 connector.ack(batchId); } Thread.sleep(1000); } } catch (Exception e) { e.printStackTrace(); } finally { connector.disconnect(); } } }