背景

工作中有很多场景，比如某些场景为了加快接口响应速度，加入缓存；实现数据持久化的同时，还要满足搜索引擎，需要一份数据多处存储等。思考：以上场景均为一份数据多份存储，这种情况下如何实现数据的同步，且保证数据一致性？
无可厚非，可将数据的多写，嵌入在业务代码里。但有些场景无权限侵入项目代码；且这种实现方式存在代码片段到处分布、事务无法保障等问题。由此，是否可将数据变更的订阅、同步抽象独立出来？canal即是这种思想下的产物组件之一。抽象业务实现如下图：

数据订阅业务流程

目录

目录

canal是什么

数据库增量日志解析、订阅、消费。阿里开源数据库订阅组件

使用场景

● 数据库镜像
● 数据库实时备份
● 索引构建和实时维护(拆分异构索引、倒排索引等)
● 业务 cache 缓存刷新
● 带业务逻辑的增量数据处理

canal工作原理

数据库主从同步

canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议
MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
canal 解析 binary log 对象(原始为 byte 流)

canal实战

1.服务端

1、开启mysql的binlog写入功能，并且配置binlog模式为row
2、分配slave角色账号
以上可联系DBA操作
3、搭建高可用zookeeper集群
4、下载部署包、解压
wget http://canal4mysql.googlecode.com/files/canal.deployer-1.0.1.tar.gz

解压列表

5、配置
/opt/soft/canal/canal.deployer-1.0.24/conf/canal.properties

canal.properties

/opt/soft/canal/canal.deployer-1.0.24/conf/huangye-evaluation/instance.properties

instance.properties

也可搭建canal-admin界面化部署
6、启动
sh /opt/soft/canal/canal.deployer-1.0.24/bin/startup.sh
日志：
tail -f /opt/soft/canal/canal.deployer-1.0.24/logs/canal/canal.log

2.客户端

业务代码

协议解析

3.协议格式

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       [具体的内容，注意为文本]

canal源码解读

canal源码

1.基础模块

● common
公共工具模块一些util、另外还有canal的server、client启动关闭相关的生命周期的基础接口定义。
● protocol
协议模块，主要是canal-sever和canal-client之间的信息传输序列化方式，采用protocobuffer序列化。
● driver
封装数据库建立连接，获取并解析二进制流的方法。获取原始的二进制流，因为单纯的jdbc无法满足获取二进制日志，所以他自己封装，又因为未来可能支持mix，statement模式所以查库的逻辑也一起封装了。
● dbsync
定义了大量的event类对应各种不同格式的二进制日志。
● filter
用于canal-server过滤掉不想关注的某些，某个表，某个字段，的二进制日志等等。还可以可以过滤是否处理ddl、dml等等。
● sink
调用filter进行过滤，将结果交给store进行缓存
● store
维护一个ringbuffer队列将解析后的二进制日志结果缓存，提供给下游消费。为了消息消费可靠，使用ack方式；为了提高性能，使用ringbuffer，消息投递方面是可靠的，顺序的，数组比链表快，不进行垃圾回收
● meta
定义了各种元数据的修改，查询操作，但是修改时可能涉及到原子性操作，需要在对应的store、sink等模块儿里加锁等方式操作。
● parser
调用dbsync解析二进制日志，并且维护二进制日志的位点，下次请求二进制日志时从对应的位点开始。位点分为本地位点储存，集群位点储存。位点有多种实现：canalconfspring下
base：基本的内存存储位点
default：基于zookeeper位点存储
file：本地存储位点
group：为了分库分表设计的位点存储，可以支持配置多个数据源，但是微端存储在内存不建议使用
● instance
调用底层模块，串起来整个流程。一个server对应多个instance，每个instance里有自己的parser、store、sink。在代码里叫instance，在配置文件instance也叫destination。一个server对应多个instance，每个instance里有自己的parser、store、sink。在代码里叫instance，在配置文件instance也叫destination。一个instance封装了一个连接数据库的实例，一个client的注册，zookeeper的交互。
● server
canal-sever的运行实例，抽象出2种实现，基于嵌入式和jetty独立部署的方式，抽象出一种对接kafka的实现。

2.可独立部署模块

● deployer
模拟主从协议，从主库拉取二进制日志，解析二进制日志，并且将解析后的结果存储在本地队列，提供给下游消费，以后称之为canal-server。
● example
从canal-server获取解析后的二进制日志，进行定制化处理，这个工程里可以放一些定制化消费canal-server消息的代码，以后称之为canal-client。
● canal-admin
由于canal-server负责连接mysql，解析二进制日志，canal-server不进行任何定制化的业务编码，仅需简单配置，因此canal-server的工作更偏向于运维。canal-admin就是用于运维在线部署配置canal-server的。

3.对外支持

● client
是连接client-server的组件包，通过这个包可以连接到客户端上面的example就是依赖了client这个模块的一个demo。
● docker
对接docker，用于支持在docker里面部署canal-server。
● prometheus
对接监控，prometheus相当于一个键值对的数据库，canal-server每间隔一段时间将prometheus格式的信息通过http端口传递给prometheus数据库。通过这些数据可以监控canal-server的情况。

canal架构

1.架构

canal架构.jpeg

说明：
server代表一个canal运行实例，对应于一个jvm
instance对应于一个数据队列（1个server对应1..n个instance)

2.canal业务架构

canal业务架构

EventParser：数据源接入，模拟slave协议和master进行交互，协议解析
EventSink：Parser和Store连接器，主要进行数据过滤，加工，分发的工作
EventStore：负责存储
MemoryMetaManager：增量订阅和消费信息管理器

3.EventParse

EventParse

整个parser过程大致可分为以下几步：

1. Connection获取上一次解析成功的log position（如果是第一次启动，则获取初始指定的位置或者是当前数据库的binlog log position）
2. Connection建立连接，向MySQL master发送BINLOG_DUMP请求
3. MySQL开始推送binary Log接收到的binary Log
4. 通过BinlogParser进行协议解析，补充一些特定信息。如补充字段名字、字段类型、主键信息、unsigned类型处理等
5. 将解析后的数据传入到EventSink组件进行数据存储（这是一个阻塞操作，直到存储成功）
6. 定时记录binary Log位置，以便重启后继续进行增量订阅
   如果需要同步的master宕机，可以从它的其他slave节点继续同步binlog日志，避免单点故障。

4.Event Sink设计：

Event Sink设计

EventSink主要作用如下：
数据过滤：支持通配符的过滤模式，表名，字段内容等
数据路由/分发：解决1:n（1个parser对应多个store的模式）
数据归并：解决n:1（多个parser对应1个store）
数据加工：在进入store之前进行额外的处理，比如join
数据1:n业务
为了合理的利用数据库资源， 一般常见的业务都是按照schema进行隔离，然后在MySQL上层或者dao这一层面上，进行一个数据源路由，屏蔽数据库物理位置对开发的影响。所以，一般一个数据库实例上，会部署多个schema，每个schema会有由1个或者多个业务方关注。
数据n:1业务
同样，当一个业务的数据规模达到一定的量级后，必然会涉及到水平拆分和垂直拆分的问题，针对这些拆分的数据需要处理时，就需要链接多个store进行处理，消费的位点就会变成多份，而且数据消费的进度无法得到尽可能有序的保证。所以，在一定业务场景下，需要将拆分后的增量数据进行归并处理，比如按照时间戳/全局id进行排序归并。

5.Event Store设计：

支持多种存储模式，比如Memory内存模式。采用内存环装的设计来保存消息，借鉴了Disruptor的RingBuffer的实现思路。
RingBuffer设计

Event Store设计

定义了3个cursor：
put：Sink模块进行数据存储的最后一次写入位置（同步写入数据的cursor）
get：数据订阅获取的最后一次提取位置（同步获取的数据的cursor）
ack：数据消费成功的最后一次消费位置
借鉴Disruptor的RingBuffer的实现，将RingBuffer拉直来看：

水平结构

实现说明：

1. put/get/ack cursor用于递增，采用long型存储。三者之间的关系为put>=get>=ack
2. buffer的get操作，通过取余或者&操作。(&操作：cusor & (size - 1) , size需要为2的指数，效率比较高)

6.Instance设计

Instance设计

instance代表了一个实际运行的数据队列，包括了EventPaser、EventSink、EventStore等组件。抽象了CanalInstanceGenerator，主要是考虑配置的管理方式：
manager方式：和你自己的内部web console/manager系统进行对接。(目前主要是公司内部使用)
spring方式：基于spring xml + properties进行定义，构建spring配置。

7.Server设计

Server

server代表了一个canal运行实例，为了方便组件化使用，特意抽象了Embeded(嵌入式)/Netty(网络访问)的两种实现。

8.增量订阅/消费设计

订阅/消费

针对上述的补充说明：
1.可以提供数据库变更前和变更后的字段内容，针对binlog中没有的name、isKey等信息进行补全
2.可以提供ddl的变更语句

9.canal HA机制

Canal的HA实现机制是依赖zookeeper实现的，主要分为Canal server和Canal client的HA。

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

10.Canal Server HA架构

csha

大致步骤：

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

踩过的坑

1.消费要幂等

canal-server在请求二进制日志时需要得到上次请求到哪个位置即位点，位点canal又分为了两种方式持久化。
a、本地文件位点：默认的存储位点方式；将位点记录在缓存中，定时刷新到本地的meta.log文件中。
b、集群存储位点：将位点存储于缓存中，定时刷到zookeeper的临时节点里，多个节点间可以共享位点
均需注意消费要幂等。

2.拷贝一个已有的server，需要删掉meta.log和修改tsdb配置

meta.log是本地存储位点方式记录的位点信息，因此拷贝一个server做部署时要删掉旧的位点信息文件，否则server启动不起来
tsdb文件是canal为了加快解析二进制日志，将表的元数据存储到了本地，因此拷贝一个server时这个文件还是旧的会导致server能启动，但是解析二进制日志失败。

3.建议不要使用tsdb，默认是开启的，而且默认缓存15天

为了提升解析二进制的速度，canal将表结构信息进行了tsdb缓存，因此当表结构发生变化时缓存的表结构可能导致解析失败。禁用掉tsdb缓存：
canal.instance.tsdb.enable = false

4.消费注意多线程操作

canal为了保证消息顺序消费，一个canal-server只能有一个canal-client与之对应。对应到api就是getWithoutAck这些api，如果多线程操作这些api会引起非顺序性消费

5.数据库迁移，binlog同步中断找不到位点信息，server卡住不消费

解决：
1） 确认事故时间点，修改配置从事故点开始继续消费
2） 删除meta.log文件或者删除zookeeper记录位置的临时节点（不同版本记录位置方式不同），从当前时刻的 binlog 开始消费，手动补充事故时间段遗漏数据
以上两种方式均需手动重启

6.binlog模式为row

开启其他模式，导致filter规则失效，或者binlog解析eventType类型出错。如Query解析为DML

总结&思考

可使用canal组件的场景，可否用业务+MQ中间件，业务双写模式实现？
业务+MQ中间件
优点：实现简单，不需要引入过多组件；维护成本低
缺点：不能保证数据实时性、一致性、业务控制事务、相同代码片段分布各处、与业务耦合
canal
优点：可保证数据实时性、一致性、解藕
缺点：单独部署，增加维护成本
具体使用哪种方案，可根据具体业务场景选择合适的方案