解决方案：数据同步Canal

关键词

伪装为mysql slave（发dump协议，接受binary log）
eventParser（模拟slave协议），eventSink（过滤加工），eventStore （存储）
@CanalEventListener—>adUpdate方法

一、概述

canal是阿里巴巴旗下的一款开源项目，纯Java开发。canal可以用来监控数据库数据的变化，从而获得新增数据，或者修改的数据。

基于数据库增量日志解析
提供增量数据订阅&消费
目前主要支持了MySQL（也支持mariaDB）

早期，阿里巴巴B2B公司因为存在杭州和美国双机房部署，存在跨机房同步的业务需求。不过早期的数据库同步业务，主要是基于trigger的方式获取增量变更，不过从2010年开始，阿里系公司开始逐步的尝试基于数据库的日志解析，获取增量变更进行同步，由此衍生出了增量订阅&消费的业务，从此开启了一段新纪元。

二、原理

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

三、架构

说明：

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

instance模块：

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

EventParser

大致过程

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

Connection获取上一次解析成功的位置 (如果第一次启动，则获取初始指定的位置或者是当前数据库的binlog位点)
Connection建立链接，发送BINLOG_DUMP指令
// 0. write command number
// 1. write 4 bytes bin-log position to start at
// 2. write 2 bytes bin-log flags
// 3. write 4 bytes server id of the slave
// 4. write bin-log file name
Mysql开始推送Binaly Log
接收到的Binaly Log的通过Binlog parser进行协议解析，补充一些特定信息
补充字段名字，字段类型，主键信息，unsigned类型处理
传递给EventSink模块进行数据存储，是一个阻塞操作，直到存储成功
存储成功后，定时记录Binaly Log位置。

EventSink

说明：

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

HA机制设计

canal的ha分为两部分，canal server和canal client分别有对应的ha实现

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

整个HA机制的控制主要是依赖了zookeeper的几个特性，watcher和EPHEMERAL节点(和session生命周期绑定)。

大致步骤：

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方式类似，也是利用zookeeper的抢占EPHEMERAL节点的方式进行控制.

四、环境部署

1. mysql开启binlog模式

（1）查看当前mysql是否开启binlog模式

SHOW VARIABLES LIKE '%log_bin%'

如果log_bin的值为OFF是未开启，为ON是已开启

（2）修改/etc/my.cnf 需要开启binlog模式

[mysqld]
log-bin=mysql-bin
binlog-format=ROW
server_id=1

修改完成之后，重启mysqld的服务

（3）进入mysql

mysql -h localhost -u root -p

（4）创建账号用于测试使用

使用root账号创建用户并授予权限

create user canal@'%' IDENTIFIED by 'canal';GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT,SUPER ON *.* TO 'canal'@'%';FLUSH PRIVILEGES;

2. canal服务端安装配置

（1）下载地址canal

https://github.com/alibaba/canal/releases/tag/canal-1.0.24

（2）下载之后上传到linux系统中，解压缩到指定的目录/usr/local/canal

解压缩之后的目录结构如下：

（3）修改 exmaple下的实例配置

修改如图所示的几个参数

一定要注释掉下面这个参数，这样就会扫描全库

#canal.instance.defaultDatabaseName =

（3）启动服务：

[root@localhost canal]# ./bin/startup.sh

（4）查看日志：

cat /usr/local/canal/logs/canal/canal.log

启动成功

五、通过canal监控数据库中表数据的变更

当用户执行数据库的操作的时候，binlog 日志会被canal捕获到，并解析出数据。我们就可以将解析出来的数据进行相应的逻辑处理。

我们这里使用的一个开源的项目，它实现了springboot与canal的集成。比原生的canal更加优雅。

https://github.com/chenqian56131/spring-boot-starter-canal

使用前需要将starter-canal安装到本地仓库（把提供的依赖拷贝到本地仓库即可）

我们可以参照它提供的canal-test，进行代码实现。

（1）创建工程模块dabing_canal，pom引入依赖

<dependency><groupId>com.xpand</groupId><artifactId>starter-canal</artifactId><version>0.0.1-SNAPSHOT</version>
</dependency>

（2）创建包com.dabing.canal ，包下创建启动类

/*** 数据监控微服务*/
@SpringBootApplication
//开启canalClient
@EnableCanalClient
public class CanalApplication {public static void main(String[] args) {SpringApplication.run(CanalApplication.class,args);}}

（3）添加配置文件application.properties

canal:client:instances:example:host: 192.168.253.128port: 11111

（4）创建com.lagou.canal.listener包，包下创建类

/*** 监控Business库下的表*/
@CanalEventListener
public class BusinessListener {/*** 设置监控点，监控的目标，对应库下的具体表* @param entryType* @param rowData*/@ListenPoint(schema = "dabing_business",table = {"tb_ad"})public void adUpdate(CanalEntry.EntryType entryType, CanalEntry.RowData rowData){System.out.println("tb_ad表中的数据发生变化");System.out.println("变化前的数据");rowData.getBeforeColumnsList().forEach((c)-> System.out.println(c.getName()+":"+c.getValue()));System.out.println("变化后的数据");rowData.getAfterColumnsList().forEach((c)-> System.out.println(c.getName()+":"+c.getValue()));}
}

测试： 启动数据监控微服务，修改dabing_business的tb_ad表，观察控制台输出。