「从零单排canal 01」 canal 10分钟入门（基于1.1.4版本）

1.简介

canal [kə'næl]，译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据 订阅 和 消费。应该是阿里云DTS（Data Transfer Service）的开源版本。

2.提供的能力

Canal与DTS提供的功能基本类似：

1）基于Mysql的Slave协议实时dump binlog流，解析为事件发送给订阅方。

2）单Canal instance，单DTS数据订阅通道均只支持订阅一个RDS，提供给一个消费者。

3）可使用canal-client客户端进行消息消费。

4）也能够经过简单配置，也能够不须要自行使用canal-client消费，能够选择直接投递到kafka或者RocketMQ集群，用户只须要使用消息队列的consumer消费便可。

5）成功消费消息后须要进行Ack，以确保一致性，服务端则会维护客户端目前的消费位点。

3.工做原理

MySQL的主从复制分红三步：

• master将改变记录到二进制日志(binary log)中（这些记录叫作二进制日志事件，binary log events，能够经过show binlog events进行查看）；
• slave将master的binary log events拷贝到它的中继日志(relay log)；
• slave重作中继日志中的事件，将改变反映它本身的数据。

 

canal 就是模拟了这个过程。

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

 

4. canal 架构

4.1 admin版本总体架构

canal 1.1.4开始支持admin管理，经过canal-admin为canal提供总体配置管理、节点运维等面向运维的功能，提供相对友好的WebUI操做界面，方便更多用户快速和安全的操做，替代了过去繁琐的配置文件管理。

总体部署架构以下。

 

• 多个canal-server能够组成集群模式，每一个instance任务经过zookeeper在集群中实现高可用
• 经过多个集群，能够实现同步资源的物理隔离
• 能够直接抓取消费投递MQ，能够实现生产/消费解耦、消息堆积、消息回溯
• 能够抓取消费投递给canal-client，在用户的服务中进行消息处理，减小中间过程

4.2 canal-server架构

 

说明：

• server表明一个canal-server运行实例，对应于一个jvm
• instance对应于一个数据队列，是真正的变动抓取的实体 （1个server能够对应多个instance)

Instance模块

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

1)EventParser子模块

EventParser模块的类图设计以下

 

每一个EventParser都会关联两个内部组件：CanalLogPositionManager ， CanalHAController

• CanalLogPositionManager：记录binlog最后一次解析成功位置信息，主要是描述下一次canal启动的位点
• CanalHAController：支持Mysql主备，判断当前该连哪一个mysql（基于Heartbeat实现，主库失去心跳则连备库）

EventParser根据HAController获知连到哪里，经过LogPositionManager获知从哪一个位点开始解析，以后便经过Mysql Slave协议拉取binlog进行解析，推入EventSink

2）EventSink子模块

目前只提供了一个带有实际做用的实现：GroupEventSink

GroupEventSink用于将多个instance上的数据进行归并，经常使用于分库后的多数据源归并。

3）EventStore子模块

EventStore的类图以下

 

官方提供的实现类是
MemoryEventStoreWIthBuffer，内部采用的是一个RingBuffer：

 

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

这些位点信息经过MetaManager进行管理。这也解释了为何一个canal instance只能支撑一个消费者：EventStore的RingBuffer只为一个消费者维护信息。

4.3 客户端使用

数据格式已经在前文给出，Canal和DTS客户端均采起：

拉取事件 -> 消费 -> 消费成功后ACK

这样的消费模式，并支持消费不成功时进行rollback，从新消费该数据。

下面是一段简单的客户端调用实例（略去异常处理）：

// 建立CanalConnector， 链接到localhost:11111

CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),11111), destination, "", "");

connector.connect(); // 链接

connector.subscribe(); // 开始订阅binlog

// 开始循环拉取

while (running) {

 Message message = connector.getWithoutAck(1024); // 获取指定数量的数据

 long batchId = message.getId();

 for (Entry entry : message.getEntries()){

 // 对每条消息进行处理

 }

 connector.ack(batchId); // ack

}

5.总结分析

5.1 优势

1）性能优异、功能全面

• canal 1.1.x 版本（release_note）,性能与功能层面有较大的突破,重要提高包括:
• 总体性能测试&优化,提高了150%. #726
• 原生支持prometheus监控 #765
• 原生支持kafka消息投递 #695
• 原生支持aliyun rds的binlog订阅 (解决自动主备切换/oss binlog离线解析) （没法拒绝它的理由！）
• 原生支持docker镜像 #801

2）运维方便

• canal 1.1.4版本，迎来最重要的WebUI能力，引入canal-admin工程，支持面向WebUI的canal动态管理能力，支持配置、任务、日志等在线白屏运维能力
• Standalone的一体化解决方案，无外部服务依赖，运维更简单，在某种程度上也意味着更稳定。
• 开箱即用，节约开发与定制成本。
• 有良好的管理控制平台与监控系统（若是你已经有promethus监控，能够秒接canal监控）

3）多语言支持

• canal 特别设计了 client-server 模式，交互协议使用 protobuf 3.0 , client 端可采用不一样语言实现不一样的消费逻辑
• canal 做为 MySQL binlog 增量获取和解析工具，可将变动记录投递到 MQ 系统中，好比 Kafka/RocketMQ，能够借助于 MQ 的多语言能力

5.2 缺点

• 单instance/订阅通道只支持订阅单个数据库，并只能支持单客户端消费。每当咱们须要新增一个消费端->MySQL的订阅：对于Canal而言，就要给MySQL接一个“Slave”，可能会对主库有必定影响。
• 消息的Schema很弱，全部消息的Schema均相同，客户端须要提早知道各个表消息的Schema与各字段的上下文才能正确消费。

好了，花了10分钟应该对canal有大体了解了，下一期，阿丸计划手把手教你搭建canal集群和admin管理平台，记得关注哦。

 

都看到最后了，原创不易，点个关注，点个赞吧～

知识碎片从新梳理，构建Java知识图谱： github.com/saigu/JavaK…（历史文章查阅很是方便）