一. canal概述
canal是Alibaba旗下的一款开源项目,纯Java开发.它是基于数据库增量日志解析,提供增量数据订阅&消费,目前主要支持mysql。
应用场景：
- 1.数据同步，比如：做在线、离线数据库之间的数据同步操作；
- 2.数据消费，比如：需要根据关注的数据库表的变化，做搜索增量；
- 3.数据脱敏，比如：需要将线上动态数据导入到其他地方，做数据脱敏。
二. canal工作原理
1. mysql主备复制实现
(1) master将改变记录到二进制日志(binary log)中（这些记录叫做二进制日志事件，binary log events，可以通过show binlog events进行查看）；
(2) slave将master的binary log events拷贝到它的中继日志(relay log)；
(3) slave重做中继日志中的事件，将改变反映它自己的数据。
2. canal的工作原理
(1) canal模拟mysql salve的交互协议,伪装自己为mysql slave,向mysql master发送dump协议;
(2) mysql master收到dump请求,开始推送binary log给slave(也就是canal);
(3) canal解析binary log对象(原始byte流).
三. canal架构设计
说明:
- server代表一个canal运行实例,对应与一个jvm;
- instance对应于一个数据队列(1个server对应1..n个instance).
instance下的子模块:
- eventParser: 数据源接入,模拟slave协议和master进行交互,协议解析;
- eventSink: parser和store链接器,进行数据的过滤,加工和分发工作;
- eventStore: 数据存储;
- metaManager: 增量订阅&消费信息管理器.
1. EventParser设计
整个parser过程大致可分为几部：
- 1.Connection获取上一次解析成功的位置（如果第一次启动，则获取初始制定的位置或者是当前数据库的binlog位点）;
- 2.Connection建立连接，发生BINLOG_DUMP命令
- 3.Mysql开始推送Binary Log;
- 4.接收到的Binary Log通过Binlog parser进行协议解析，补充一些特定信息;
- 5.传递给EventSink模块进行数据存储，是一个阻塞操作，直到存储成功
存储成功后，定时记录Binary Log位置.
2. EventSink设计
- 数据过滤：支持通配符的过滤模式，表名，字段内容等;
- 数据路由/分发：解决1:n (1个parser对应多个store的模式);
- 数据归并：解决n:1 (多个parser对应1个store);
- 数据加工：在进入store之前进行额外的处理，比如join.
3. EventStore设计
目前canal实现了memory内存、本地file存储以及持久化到zookeeper以保障数据集群共享。memory内存的RingBuffer设计如下图：
定义了3个cursor：
- Put : Sink模块进行数据存储的最后一次写入位置
- Get : 数据订阅获取的最后一次提取位置
- Ack : 数据消费成功的最后一次消费位置
借鉴Disruptor的RingBuffer的实现，将RingBuffer拉直来看：
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 [具体的内容，注意为文本]
四. 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操作,否则客户端接收无法保证有序.
大致步骤：
- 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节点的方式进行控制.
HA配置架构图:
HA配置架构设计图
canal其他链接方式:
1. 单连
2. 两个client+两个instance+1个mysql(其实就是两个单连)
3. 一个server+两个instance+两个mysql+两个client
4. instance的standby配置
五. 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机制.
