Canal是阿里巴巴旗下的一款开源项目,利用Java开发。主要用途是基于MySQL数据库增量日志解析,提供增量数据订阅和消费,目前主要支持MySQL。
GitHub地址:
https://github.com/alibaba/canal
在介绍Canal内部原理之前,首先来了解一下MySQL Master/Slave同步原理:
简而言之,Canal是通过模拟成为MySQL的slave,监听MySQL的binlog日志来获取数据。当把MySQL的binlog设置为row模式以后,可以获取到执行的每一个Insert/Update/Delete的脚本,以及修改前和修改后的数据,基于这个特性,Canal就能高效的获取到MySQL数据的变更。
说明:
server代表一个Canal运行实例,对应于一个jvm
instance对应于一个数据队列(1个server对应1..n个instance)
EventParser:数据源接入,模拟slave协议和master进行交互,协议解析
EventSink:Parser和Store连接器,主要进行数据过滤,加工,分发的工作
EventStore:负责存储
MemoryMetaManager:增量订阅和消费信息管理器
整个parser过程大致可分为以下几步:
如果需要同步的master宕机,可以从它的其他slave节点继续同步binlog日志,避免单点故障。
EventSink主要作用如下:
数据过滤:支持通配符的过滤模式,表名,字段内容等
数据路由/分发:解决1:n(1个parser对应多个store的模式)
数据归并:解决n:1(多个parser对应1个store)
数据加工:在进入store之前进行额外的处理,比如join 数据1:n业务
为了合理的利用数据库资源, 一般常见的业务都是按照schema进行隔离,然后在MySQL上层或者dao这一层面上,进行一个数据源路由,屏蔽数据库物理位置对开发的影响,阿里系主要是通过cobar/tddl来解决数据源路由问题。所以,一般一个数据库实例上,会部署多个schema,每个schema会有由1个或者多个业务方关注。
同样,当一个业务的数据规模达到一定的量级后,必然会涉及到水平拆分和垂直拆分的问题,针对这些拆分的数据需要处理时,就需要链接多个store进行处理,消费的位点就会变成多份,而且数据消费的进度无法得到尽可能有序的保证。所以,在一定业务场景下,需要将拆分后的增量数据进行归并处理,比如按照时间戳/全局id进行排序归并。
支持多种存储模式,比如Memory内存模式。采用内存环装的设计来保存消息,借鉴了Disruptor的RingBuffer的实现思路。 RingBuffer设计:
定义了3个cursor:
put:Sink模块进行数据存储的最后一次写入位置(同步写入数据的cursor)
get:数据订阅获取的最后一次提取位置(同步获取的数据的cursor)
ack:数据消费成功的最后一次消费位置
借鉴Disruptor的RingBuffer的实现,将RingBuffer拉直来看:
实现说明:
instance代表了一个实际运行的数据队列,包括了EventPaser、EventSink、EventStore等组件。抽象了CanalInstanceGenerator,主要是考虑配置的管理方式: manager方式:和你自己的内部web console/manager系统进行对接。(目前主要是公司内部使用)
spring方式:基于spring xml + properties进行定义,构建spring配置。
server代表了一个Canal运行实例,为了方便组件化使用,特意抽象了Embeded(嵌入式)/Netty(网络访问)的两种实现。
具体的协议格式,可参见:CanalProtocol.proto。
数据对象格式:EntryProtocol.proto
Entry
Header
logfileName [binlog文件名]
logfileOffset [binlog position]
executeTime [binlog里记录变更发生的时间戳]
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 [column序号]
sqlType [jdbc type]
name [column name]
isKey [是否为主键]
updated [是否发生过变更]
isNull [值是否为null]
value [具体的内容,注意为文本]
针对上述的补充说明: 1.可以提供数据库变更前和变更后的字段内容,针对binlog中没有的name、isKey等信息进行补全 2.可以提供ddl的变更语句
Canal的HA实现机制是依赖zookeeper实现的,主要分为Canal server和Canal client的HA。 Canal server:为了减少对MySQL dump的请求,不同server上的instance要求同一时间只能有一个处于running状态,其他的处于standby状态。
Canal client:为了保证有序性,一份instance同一时间只能由一个Canal client进行get/ack/rollback操作,否则客户端接收无法保证有序。
大致步骤:
Canal Client的方式和Canal server方式类似,也是利用Zookeeper的抢占EPHEMERAL节点的方式进行控制。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。