李鸿星电子商务网站建设,wordpress英文主题 汉化,付费阅读网站代码,千川推广官网前言前面的文章里#xff0c;我们了解到 canal 可以从 MySQL 中感知数据的变化。这是因为它模拟 MySQL slave 的交互协议#xff0c;伪装自己为 MySQL slave #xff0c;从而实现了主从复制。正是了解到这一点#xff0c;笔者有两个问题便一直萦绕于心#xff1a;它是如何…前言前面的文章里我们了解到 canal 可以从 MySQL 中感知数据的变化。这是因为它模拟 MySQL slave 的交互协议伪装自己为 MySQL slave 从而实现了主从复制。正是了解到这一点笔者有两个问题便一直萦绕于心它是如何模拟 MySQL slave 交互协议的它又是怎么解析 binlog 日志的呢今天笔者准备就着这两个问题扒拉扒拉 canal 的代码一探究竟。一、MySQL 主从复制在谈 canal 之前我们有必要再重温下 MySQL 主从复制的原理。总结上图的流程如下MySQL master 将数据变更写入二进制日志 (binary log , 其中记录叫做二进制日志事件binary log events)MySQL slave 将 master 的 binary log events 拷贝到它的中继日志 (relay log)MySQL slave 重放 relay log 中的事件将数据变更反映到自己的数据库。二、canal 原理上图就很形象的描述了 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模块中开始进行的。在 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 中连接过程协议如下在代码中我们看一下它的连接方法其中 negotiate方法是握手协议和客户端验证的具体实现。就是按照 MySQL 的协议规范通过上面创建的 Socketchannel来读写网络数据。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名称和position3、注册slave现在开始调用 erosaConnection.dump(binlogfilename,binlogPosition,func)方法来注册slave和发送dump命令。在使用 COM_BINLOG_DUMP请求binlog事件之前发送在主服务器上注册一个从服务器它的指令是 COM_REGISTER_SLAVE。注册完之后就是发送dump请求它的指令是 COM_BINLOG_DUMP。在执行完这段代码后我们通过 show processlist;查看进程就可以看到这个dump线程的状态。其中 command Binlog Dumpstate Master has sent all binlog to slave; waiting for more updates。五、如何解析binlog数据 ?在上面章节中我们已经看到MySQL主服务器已经接受了 canal 这个从服务器那么当canal拿到binlog内容后 又是怎么解析它的呢首先还记得在配置MySQL服务器的时候我们将 binlog-format设置为ROW模式它是基于行的复制。binlog中每一个数据变更可以叫做事件在ROW模式下有几个主要的事件类型事件SQL命令rows 内容TABLEMAPEVENTnull定义将要更改的表。WRITEROWSEVENT插入要插入的行数据DELETEROWSEVENT删除被删除的数据UPDATEROWSEVENT更新原数据要更改的数据每一次数据的变更都会触发2个事件先把要更改的表信息告诉你然后再告诉你更改的row内容。比如 TABLE_MAP_EVENTWRITE_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都明明白白的写在那儿啦。
