我们了解到 canal 可以从 MySQL 中感知数据的变化。这是因为它模拟 MySQL slave 的交互协议,伪装自己为 MySQL slave ,从而实现了主从复制。

正是了解到这一点,笔者有两个问题便一直萦绕于心:

  • 它是如何模拟 MySQL slave 交互协议的?
  • 它又是怎么解析 binlog 日志的呢?

今天,笔者准备就着这两个问题,扒拉扒拉 canal 的代码,一探究竟。

一、MySQL 主从复制

在谈 canal 之前,我们有必要再重温下 MySQL 主从复制的原理。

ea577bb5cf57af16962c15e7e46208b3.png

总结上图的流程如下:

  • MySQL master 将数据变更写入二进制日志 (binary log , 其中记录叫做二进制日志事件binary log events);
  • MySQL slave 将 master 的 binary log events 拷贝到它的中继日志 (relay log);
  • MySQL slave 重放 relay log 中的事件,将数据变更反映到自己的数据库。

二、canal 原理

67898ef0394bc6a4f2251b8f6938a4a1.png

上图就很形象的描述了 canal 的角色。它的原理也很简单:

  • canal模拟mysql slave的交互协议,伪装自己为mysql slave,向mysql master发送dump协议;
  • mysql master收到dump请求,开始推送binary log给slave(也就是canal);
  • canal解析binary log对象(原始为byte流);
  • canal将解析后的对象,根据业务场景,分发到比如 MySQL 、RocketMQ 或者 ES 中。

三、源码启动

看完了 MySQL 主从复制和 canal 原理之后,为了方便 debug ,笔者已经在 GitHub Fork 了源码,并导入本地。

可以找到 com.alibaba.otter.canal.deployer.CanalLauncher 类,它就是 canal 独立版本启动的入口类。

在这里,直接运行 main 方法即可运行 canal ,和在 /canal/bin/startup.sh 中效果一样。

事实上,canal 的代码比较多,在架构上又分了很多模块设计,比如事件解析器、事件消费、内存存储、服务实例、元数据、高可用等。

本文不打算面面俱到介绍每一个的实现,那就得正儿八经写一个 canal 系列才行。主要还是为了开头我们提出的那两个问题。

四、如何模拟slave ?

上面我们已经说到, CanalLauncher 是canal 启动的入口类。

运行 main 方法之后, canal 会先做很多准备工作。比如加载配置文件、初始化消息队列、启动 canal Admin、加载Spring配置、注册钩子程序等。

canal 模拟 slave 协议,是在 EventParser 模块中开始进行的。

b77bfe98de26a3df789a6177d19e5e5f.png

在 canal 代码中,整个流程简化如下:

// 开始执行replication// 1. 构造Erosa连接ErosaConnection erosaConnection = buildErosaConnection();// 2. 启动一个心跳线程startHeartBeat(erosaConnection);// 3. 执行dump前的准备工作preDump(erosaConnection);erosaConnection.connect();// 链接// 查询master serverIdlong queryServerId = erosaConnection.queryServerId();if (queryServerId != 0) {    serverId = queryServerId;}// 4. 获取binlog最后的位置信息EntryPosition position = findStartPosition(erosaConnection);final EntryPosition startPosition = position;// 加载元数据processTableMeta(startPosition);// 重新链接,因为在找position过程中可能有状态,需要断开后重建erosaConnection.reconnect();// 4. 开始dump数据erosaConnection.dump(startPosition.getJournalName(),startPosition.getPosition(),sinkHandler);
1、握手、验证

在开始之前,canal 必须先要和 MySQL 服务器建立连接,并完成客户端身份验证。

在 MySQL 中,连接过程协议如下:

dcbcb781a4096dfb970390d5f1447cd2.png

在代码中,我们看一下它的连接方法:

f2fa22b4915f9718244f733ffe577cdb.png

其中, negotiate 方法是握手协议和客户端验证的具体实现。就是按照 MySQL 的协议规范,通过上面创建的 Socket channel 来读写网络数据。

2、dump前的准备

正确连接到 MySQL 后,在开始执行 dump 指令之前,还要初始化一些配置信息。

思路就是通过 MySQL 执行器,执行 SQL 语句,获取信息。

代码就不粘了,不过它们执行的语句如下:

show variables like 'binlog_format'      #获取binlog format格式show variables like 'binlog_row_image'   #获取binlog image格式show variables like 'server_id'          #获取matser serverIdshow master status                       #获取binlog名称和position复制代码
3、注册slave

现在开始调用 erosaConnection.dump(binlogfilename,binlogPosition,func) 方法,来注册slave和发送dump命令。

在使用 COM_BINLOG_DUMP 请求binlog事件之前发送,在主服务器上注册一个从服务器,它的指令是 COM_REGISTER_SLAVE 。

1548e203997201171fdb56b4613abd69.png

注册完之后,就是发送dump请求,它的指令是 COM_BINLOG_DUMP 。

f067a1da0c2ff37ac5aa97591ce43f84.png

在执行完这段代码后,我们通过 show processlist; 查看进程,就可以看到这个dump线程的状态。

iduserhostdbcommandtimestate139canallocalhost:62901nullBinlog Dump3Master has sent all binlog to slave; waiting for more updates

五、如何解析binlog数据 ?

在上面章节中,我们已经看到,MySQL主服务器已经接受了 canal 这个从服务器,那么当canal拿到binlog内容后, 又是怎么解析它的呢?

首先,还记得在配置MySQL服务器的时候,我们将 binlog-format 设置为ROW模式,它是基于行的复制。

binlog中每一个数据变更可以叫做事件,在ROW模式下,有几个主要的事件类型:

事件SQL命令rows 内容TABLE_MAP_EVENTnull定义将要更改的表。WRITE_ROWS_EVENT插入要插入的行数据DELETE_ROWS_EVENT删除被删除的数据UPDATE_ROWS_EVENT更新原数据+要更改的数据

每一次数据的变更,都会触发2个事件,先把要更改的表信息告诉你,然后再告诉你更改的row内容。

比如 TABLE_MAP_EVENT + WRITE_ROWS_EVENT 。

canal在接收到binlog数据后,并不会马上把它解析成我们熟悉的JSON数据,而是在发送的时候才开始。

比如我们选择使用 RocketMQ ,那么在发送之前才开始将binlog里面的byte数组转化为对象。

// 并发构造EntryRowData[] datas = MQMessageUtils.buildMessageData(message, executor);// 串行分区List flatMessages = MQMessageUtils.messageConverter(datas, message.getId());

在这两个方法里,就完成了byte数组到对象的转化。转化成的 FlatMessage 对象,就成了我们在消息队列中消费到的数据结构。

public class FlatMessage implements Serializable {    private long                      id;    private String                    database;    private String                    table;    private List              pkNames;    private Boolean                   isDdl;    private String                    type;    // binlog executeTime    private Long                      es;    // dml build timeStamp    private Long                      ts;    private String                    sql;    private Map      sqlType;    private Map       mysqlType;    private List> data;    private List> old;}

总结

正如本文开头所言,笔者在刚了解到canal机制的时候,确实觉得很不可思议。

咦,它是怎么模拟MySQL slave的呢 ? 总觉得是不是有啥黑科技在里面。。。

事实上,这是源于笔者对MySQL的无知。

MySQL早就制定好了各种接口协议,怎么连接、验证、注册和dump都明明白白的写在那儿啦。

正是应了那句话:花开正好,只待君来~

GitHub 加速计划 / ca / canal
53
10
下载
alibaba/canal: Canal 是由阿里巴巴开源的分布式数据库同步系统,主要用于实现MySQL数据库的日志解析和实时增量数据订阅与消费,广泛应用于数据库变更消息的捕获、数据迁移、缓存更新等场景。
最近提交(Master分支:4 个月前 )
8a4199a7 * 1. Fix compressed OSS binlog data 2. Fix first second data loss caused by dumping from OSS binlog * Fix CI failed test cases 30 天前
79338be0 - String.format is lower than StringBuilder. Benchmark like below: code snippet: String str = String.format("%s-%s-%s", 0, 1, 10); Benchmark Mode Cnt Score Error Units StringBenchmark.append thrpt 46431458.255 ops/s StringBenchmark.format thrpt 985724.313 ops/s StringBenchmark.append avgt ≈ 10⁻⁸ s/op StringBenchmark.format avgt ≈ 10⁻⁶ s/op StringBenchmark.append sample 364232 ≈ 10⁻⁷ s/op StringBenchmark.append:p0.00 sample ≈ 10⁻⁸ s/op StringBenchmark.append:p0.50 sample ≈ 10⁻⁷ s/op StringBenchmark.append:p0.90 sample ≈ 10⁻⁷ s/op StringBenchmark.append:p0.95 sample ≈ 10⁻⁷ s/op StringBenchmark.append:p0.99 sample ≈ 10⁻⁷ s/op StringBenchmark.append:p0.999 sample ≈ 10⁻⁷ s/op StringBenchmark.append:p0.9999 sample ≈ 10⁻⁵ s/op StringBenchmark.append:p1.00 sample 0.001 s/op StringBenchmark.format sample 336220 ≈ 10⁻⁶ s/op StringBenchmark.format:p0.00 sample ≈ 10⁻⁶ s/op StringBenchmark.format:p0.50 sample ≈ 10⁻⁶ s/op StringBenchmark.format:p0.90 sample ≈ 10⁻⁶ s/op StringBenchmark.format:p0.95 sample ≈ 10⁻⁶ s/op StringBenchmark.format:p0.99 sample ≈ 10⁻⁶ s/op StringBenchmark.format:p0.999 sample ≈ 10⁻⁵ s/op StringBenchmark.format:p0.9999 sample ≈ 10⁻⁴ s/op StringBenchmark.format:p1.00 sample 0.001 s/op StringBenchmark.append ss ≈ 10⁻⁶ s/op StringBenchmark.format ss ≈ 10⁻⁵ s/op 30 天前
Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐