1. canal详解

add: 2021-08-24

1.1. canal概述

canal是Alibaba旗下的一款开源项目,纯Java开发.它是基于数据库增量日志解析,提供增量数据订阅&消费,目前主要支持mysql。

1.1.1. 应用场景：

数据同步，比如：做在线、离线数据库之间的数据同步操作；
数据消费，比如：需要根据关注的数据库表的变化，做搜索增量；
数据脱敏，比如：需要将线上动态数据导入到其他地方，做数据脱敏。

1.2. canal工作原理

1.2.1. mysql主备复制实现

master将改变记录到二进制日志(binary log)中（这些记录叫做二进制日志事件，binary log events，可以通过show binlog events进行查看）；
slave将master的binary log events拷贝到它的中继日志(relay log)；
slave重做中继日志中的事件，将改变反映它自己的数据。

1.2.2. canal的工作原理

canal模拟mysql salve的交互协议,伪装自己为mysql slave,向mysql master发送dump协议;
mysql master收到dump请求,开始推送binary log给slave(也就是canal);
canal解析binary log对象(原始byte流).

1.3. canal架构设计

说明:

server代表一个canal运行实例,对应与一个jvm;
instance对应于一个数据队列(1个server对应1..n个instance).

instance下的子模块:

eventParser: 数据源接入,模拟slave协议和master进行交互,协议解析;
eventSink: parser和store链接器,进行数据的过滤,加工和分发工作;
eventStore: 数据存储;
metaManager: 增量订阅&消费信息管理器.

1.3.1. EventParser设计

整个parser过程大致可分为几部：

Connection获取上一次解析成功的位置（如果第一次启动，则获取初始制定的位置或者是当前数据库的binlog位点）;
Connection建立连接，发生BINLOG_DUMP命令
Mysql开始推送Binary Log;
接收到的Binary Log通过Binlog parser进行协议解析，补充一些特定信息;
传递给EventSink模块进行数据存储，是一个阻塞操作，直到存储成功

存储成功后，定时记录Binary Log位置.

1.3.2. EventSink设计

数据过滤：支持通配符的过滤模式，表名，字段内容等;
数据路由/分发：解决1:n (1个parser对应多个store的模式);
数据归并：解决n:1 (多个parser对应1个store);
数据加工：在进入store之前进行额外的处理，比如join.

1.3.3. EventStore设计

目前canal实现了memory内存、本地file存储以及持久化到zookeeper以保障数据集群共享。memory内存的RingBuffer设计如下图：

定义了3个cursor：

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

借鉴Disruptor的RingBuffer的实现，将RingBuffer拉直来看：

1.3.4. 增量订阅、消费设计

get/ack/rollback协议介绍：

Message getWithoutAck(int batchSize)，允许指定batchSize，一次可以获取多条，每次返回的对象为Message，包含的内容为：batch id（唯一标识）和entries（具体的数据对象），具体格式后面会进行介绍。
void rollback(long batchId)，顾命思议，回滚上次的get请求，重新获取数据。基于get获取的batchId进行提交，避免误操作；
void ack(long batchId)，顾命思议，确认已经消费成功，通知server删除数据。基于get获取的batchId进行提交，避免误操作
canal的get/ack/rollback协议和常规的jms协议有所不同，允许get/ack异步处理，比如可以连续调用get多次，后续异步按顺序提交ack/rollback，项目中称之为流式api.

流式api设计的好处：

get/ack异步化，减少因ack带来的网络延迟和操作成本 (99%的状态都是处于正常状态，异常的rollback属于个别情况，没必要为个别的case牺牲整个性能)；
get获取数据后，业务消费存在瓶颈或者需要多进程/多线程消费时，可以不停的轮询get数据，不停的往后发送任务，提高并行化.

数据格式：

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

1.4. canal的HA机制设计

canal的HA机制主要是依赖zookeeper的特性,共分为canal server和canal client两部分:

canal server:为了减少对mysql dump的请求,不同server上的instance要求同一时间只能有一个处于running,其他的处于standby状态.
canal client:为了保证有序性,一份instance同一时间只能由一个canal client进行get/ack/rollback操作,否则客户端接收无法保证有序.

1.4.1. 大致步骤：

canal server要启动某个canal instance时都先向zookeeper进行一次尝试启动判断 (实现：创建EPHEMERAL节点，谁创建成功就允许谁启动)
创建zookeeper节点成功后，对应的canal server就启动对应的canal instance，没有创建成功的canal instance就会处于standby状态
一旦zookeeper发现canal server A创建的节点消失后，立即通知其他的canal server再次进行步骤1的操作，重新选出一个canal server启动instance.
canal client每次进行connect时，会首先向zookeeper询问当前是谁启动了canal instance，然后和其建立链接，一旦链接不可用，会重新尝试connect.
Canal Client的方式和canal server方式类似，也是利用zokeeper的抢占EPHEMERAL节点的方式进行控制.

1.4.2. HA配置架构图:

canal其他链接方式:

单连

两个client+两个instance+1个mysql(其实就是两个单连)

一个server+两个instance+两个mysql+两个client

instance的standby配置

1.5. canal总结

canal原理：模拟mysql slave的交互协议，伪装自己为mysql slave，向mysql master发送dump协议；mysql master收到dump请求，开始推送binary log给slave(也就是canal)；解析binary log对象(原始为byte流)
存在重复消费问题：需要在消费端解决。
canal需要维护EventStore，可以存取在Memory, File, zk.
canal需要维护客户端的状态，同一时刻一个instance只能有一个消费端消费.
支持binlog format 类型:statement, row, mixed. 多次附加功能只能在row下使用，比如otter.
有ACK机制.

以上摘自: 了解canal

Canal详解