使用 Canal 实现 MYSQL 数据的实时变更抓取与同步

Java精选 2022-08-09

收录于合集 #MySQL 84个

今天，给大家介绍一个生产实践更多的实时变更抓取模块canal。

https://github.com/alibaba/canal

实际上，该项目目前已经发布到1.1.5-alpha版本，不仅仅支持数据变更抓取，还支持数据同步、admin配置管理，已经非常完善。

简介

Canal是阿里巴巴的一个开源项目，它同样实现了Mysql变更抓取（Binlog解析），并提供了订阅功能。应该是阿里云DTS（Data Transfer Service）的开源版本。

提供的功能

Canal与DTS提供的功能基本相似：

1）基于Mysql的Slave协议实时dump binlog流，解析为事件发送给订阅方；

事件格式为(省略一些域后的伪码)：

Eevnt {
    Header {
        logfileName     [binlog文件名]
        logfileOffset   [binlog position]
        executeTime     [binlog里记录变更发生的时间戳]
        schemaName      [数据库实例]
        tableName       [表名]
        eventType       [insert/update/delete类型]
    }
    entryType           [事务头BEGIN/事务尾END/数据ROWDATA]
    storeValue          [byte数据,可展开，对应的类型为RowChange]
    isDdl               [是否是ddl变更操作，比如create table/drop table]
    sql                 [具体的ddl sql]
    rowDatas            [具体insert/update/delete的变更数据，可为多条，1个binlog event事件可对应多条变更，比如批处理]
    beforeColumns       [Column类型的数组]
    afterColumns        [Column类型的数组]
}
Column {
    index       [column序号]
    sqlType     [jdbc type]
    name        [column name]
    isKey       [是否为主键]
    updated     [是否发生过变更]
    isNull      [值是否为null]
    value       [具体的值]
}

2）单Canal instance，单DTS数据订阅通道均只支持订阅一个RDS，提供给一个消费者；

3）客户端成功消费消息后需要进行Ack，以确保一致性，服务端则会维护客户端目前的消费位点。

架构浅析

DTS为商业产品，无法获得源码与设计文档，鉴于Canal和DTS提供的功能极为相似，这里分析Canal的架构作为该类产品的代表。

Canal是一个Standalone的Nettty应用（也支持用户采用embedded模式嵌入到自己的服务中），它的整体结构如下：

server：代表一个canal运行实例，对应一个jvm
instance：对应于一个数据队列，一个instance订阅单个Mysql并支撑单个消费者，一个server内可运行多个instance
eventParser：数据源接入，模拟slave协议和master进行交互，协议解析
eventSink：Parser和Store链接器，进行数据过滤，加工，分发的工作
eventStore：数据存储模块
metaManager：客户端消费位点管理模块

EventParser

EventParser模块的类图设计如下

每个EventParser都会关联两个内部组件：CanalLogPositionManager ， CanalHAController

1、CanalLogPositionManager：记录binlog最后一次解析成功位置信息，主要是描述下一次canal启动的位点

2、CanalHAController：支持Mysql主备，判断当前该连哪个mysql（基于Heartbeat实现，主库失去心跳则连备库）

EventParser根据HAController获知连到哪里，通过LogPositionManager获知从哪个位点开始解析，之后便通过Mysql Slave协议拉取binlog进行解析，推入EventSink

EventSink

目前只提供了一个带有实际作用的实现：GroupEventSink

GroupEventSink用于将多个instance上的数据进行归并，常用于分库后的多数据源归并。

EventStore

EventStore的类图如下

官方提供的实现类是MemoryEventStoreWIthBuffer，内部采用的是一个RingBuffer：

Put : Sink模块进行数据存储的最后一次写入位置
Get : 数据订阅获取的最后一次提取位置
Ack : 数据消费成功的最后一次消费位置

这些位点信息通过MetaManager进行管理。这也解释了为什么一个canal instance只能支撑一个消费者：EventStore的RingBuffer只为一个消费者维护信息。

客户端使用

数据格式已经在前文给出，Canal和DTS客户端均采取：

拉取事件 -> 消费 -> 消费成功后ACK

这样的消费模式，并支持消费不成功时进行rollback，重新消费该数据。

下面是一段简单的客户端调用实例（略去异常处理）：

// 创建CanalConnector， 连接到localhost:11111
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),11111), destination, "", "");
connector.connect();    // 连接
connector.subscribe();  // 开始订阅binlog
// 开始循环拉取
while (running) {
    Message message = connector.getWithoutAck(1024); // 获取指定数量的数据
    long batchId = message.getId();
    for (Entry entry : message.getEntries()){
        // 对每条消息进行处理
    }
    connector.ack(batchId); // ack
}

DTS的客户端代码也类似(当然SDK提供的方法名和类名全都不一样)，可以看到得益于Client的高度封装，相当易用。