>作者：幻好
>来源：[恒生LIGHT云社区](https://developer.hs.net/thread/1236)

> 随着系统业务量的不断扩大，都会使用分布式的方式，同时会有非常多的中间件，如redis、消息队列、大数据存储等，但是实际核心的数据存储依然是存储在数据库，多个数据库之前就会存在数据实时同步的问题，为了解决这个问题，需要采用一些数据实时同步中间件来解决问题。

### Canal简介

`Canal`是阿里开源的一款基于 `Mysql`数据库 `binlog`的增量订阅和消费组件，通过它可以订阅数据库的 `binlog`日志，然后进行一些数据消费，如数据镜像、数据异构、数据索引、缓存更新等。相对于消息队列，通过这种机制可以实现数据的有序化和一致性。

### 设计背景

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

目前内部使用的同步，已经支持mysql5.x和oracle部分版本的日志解析。

基于日志增量订阅&消费支持的业务：

1. 数据库镜像
2. 数据库实时备份
3. 多级索引 (卖家和买家各自分库索引)
4. search build
5. 业务cache刷新
6. 价格变化等重要业务消息

### 工作原理

#### mysql主备复制实现

![687474703a2f2f646c2e69746579652e636f6d2f75706c6f61642f6174746163686d656e742f303038302f333038362f34363863316131342d653761642d333239302d396433642d3434616335303161373232372e6a7067.jpg](https://developer.hs.net/storage/attachments/2021/07/25/P2Z7lB7DZVXyJ1zc5AsQFshMOiENdfDxawrcEKOx_thumb.jpg "2358")
从上层来看，复制分成三步：

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

#### canal的工作原理

![687474703a2f2f646c2e69746579652e636f6d2f75706c6f61642f6174746163686d656e742f303038302f333130372f63383762363762612d333934632d333038362d393537372d3964623035626530346339352e6a7067.jpg](https://developer.hs.net/storage/attachments/2021/07/25/CTLxyg6KVz0IZvA15azGbFqXHGxxmGMoB0Wvxg0C_thumb.jpg "2357")

原理相对比较简单：

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

### 底层架构

![687474703a2f2f646c2e69746579652e636f6d2f75706c6f61642f6174746163686d656e742f303038302f333132362f34393535303038352d306364322d333266612d383661362d6636373664623562353937622e6a7067.jpg](https://developer.hs.net/storage/attachments/2021/07/25/TvjEvLbuJtD4sgkIaHOJw12h8N4KOdURAgq8lTc8_thumb.jpg "2356")

说明：

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

`instance`模块：

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

#### EventParser设计

![687474703a2f2f646c2e69746579652e636f6d2f75706c6f61642f6174746163686d656e742f303038302f333134332f37393531633136392d663764662d336362332d616562622d6439323466353733313163622e6a7067.jpg](https://developer.hs.net/storage/attachments/2021/07/25/I5o7k0b51Z8YjzlPU5ILLziRp2bFn89LUGZYblBs_thumb.jpg "2355")

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

1. `Connection`获取上一次解析成功的位置 (如果第一次启动，则获取初始指定的位置或者是当前数据库的`binlog`位点)
2. `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
3. `Mysql`开始推送`Binaly Log`
4. 接收到的`Binaly Log`的通过`Binlog parser`进行协议解析，补充一些特定信息
   // 补充字段名字，字段类型，主键信息，unsigned类型处理
5. 传递给`EventSink`模块进行数据存储，是一个阻塞操作，直到存储成功
6. 存储成功后，定时记录`Binaly Log`位置

#### EventSink设计

![687474703a2f2f646c2e69746579652e636f6d2f75706c6f61642f6174746163686d656e742f303038302f333231342f30613266643637312d643665392d336564372d623131302d3661336237333861336362302e6a7067.jpg](https://developer.hs.net/storage/attachments/2021/07/25/dZUTedEZAyp29fN9qLNZVSBGBQ41zBDUzAm2x4Ry_thumb.jpg "2354")

说明：

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

#### EventStore设计

* 1.目前仅实现了`Memory`内存模式，后续计划增加本地`file`存储，`mixed`混合模式
* 2.借鉴了`Disruptor`的`RingBuffer`的实现思路

`RingBuffer`设计：

![687474703a2f2f646c2e69746579652e636f6d2f75706c6f61642f6174746163686d656e742f303038302f333233372f30363365383438302d313563382d336536362d626264332d3963343464656630396338662e6a7067.jpg](https://developer.hs.net/storage/attachments/2021/07/25/6hYb2qSlpkSoVoDtpe99OuiwFfF36HhmOxJWAa5l_thumb.jpg "2353")

定义了3个cursor

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

#### Instance设计

![687474703a2f2f646c2e69746579652e636f6d2f75706c6f61642f6174746163686d656e742f303038302f333234372f35646531633961662d373739382d336434322d626334332d3563353464383263346462662e6a7067.jpg](https://developer.hs.net/storage/attachments/2021/07/25/y4ZrRqNC2UrLae9eQEK9LBZAa0SYDbzBo6YGB40U_thumb.jpg "2352")

`instance`代表了一个实际运行的数据队列，包括了 `EventPaser`,`EventSink`,`EventStore`等组件。

抽象了 `CanalInstanceGenerator`，主要是考虑配置的管理方式：

* `manager`方式： 和你自己的内部web console/manager系统进行对接。(目前主要是公司内部使用)
* `spring`方式：基于spring xml + properties进行定义，构建spring配置。

#### Server设计

![687474703a2f2f646c2e69746579652e636f6d2f75706c6f61642f6174746163686d656e742f303038302f333235372f66346466333862612d353965322d333938652d623565622d3162626662656363303637362e6a7067.jpg](https://developer.hs.net/storage/attachments/2021/07/25/kxDQbtxKe3UEbSBH8CelWVzYKgDzhdgNGXxh94wt_thumb.jpg "2351")

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

* `Embeded` : 对`latency`和可用性都有比较高的要求，自己又能hold住分布式的相关技术(比如`failover`)
* `Netty` : 基于`netty`封装了一层网络协议，由`canal server`保证其可用性，采用的pull模型，当然`latency`会稍微打点折扣，不过这个也视情况而定。(阿里系的`notify`和`metaq`，典型的`push/pull`模型，目前也逐步的在向`pull`模型靠拢，`push`在数据量大的时候会有一些问题)

### 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:

![687474703a2f2f646c2e69746579652e636f6d2f75706c6f61642f6174746163686d656e742f303038302f333330332f64333230326332362d653935342d333563302d613331392d3537363034313032633537642e6a7067.jpg](https://developer.hs.net/storage/attachments/2021/07/25/9ukwZMrgPAi183echCfB4uMQXPsOkwSECYWeQPbc_thumb.jpg "2350")

大致步骤：

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节点的方式进行控制。

### 应用场景

#### 数据同步

* 微服务开发环境下，为了提高搜索效率，以及搜索的精准度，会大量使用`Redis`、`Mongodb`等`NoSQL`数据库，也会使用大量的`Solr`、`Elasticsearch`等全文检索服务。那么，这个时候，就会有一个问题需要我们来思考和解决：那就是数据同步的问题！
* Canal可以将实时变化的数据库中的数据同步到`Redis`,`Mongodb`或者`Solr`/`Elasticsearch`中。![20200930105843756.png](https://developer.hs.net/storage/attachments/2021/07/25/LmYETqM9Svb2Yvoayq92enGnoklwLzVDWBudR9yx_thumb.png "2349")

#### 数据异构

在大型网站架构中，DB都会采用分库分表来解决容量和性能问题，但分库分表之后带来的新问题。比如不同维度的查询或者聚合查询，此时就会非常棘手。一般我们会通过数据异构机制来解决此问题，数据异构就是将需要join查询的多表按照某一个维度又聚合在一个DB中。让你去查询。canal就是实现数据异构的手段之一。

![1577453-20191109101403531-66775627.png](https://developer.hs.net/storage/attachments/2021/07/25/4b0wW75r3czVjXKtdaaqVAEfwFyULAUV8l9qZLIh_thumb.png "2348")

ITPub博客

认识数据实时同步神器—Canal

最新文章