探究 Canal EventParser 的设计与实现奥妙-阿里云开发者社区

本文将从三个方面深度剖析 EventParser 组件。

从官方文档看 EventParser 的设计思想
从 EventParser 初始化了解内部的是可配置项
从 EventParser 的启动窥探其工作实现原理

温馨提示：本篇篇幅较长，如果耐心阅读一定会有不错的收获，为了提高阅读体验，本文所有源码都是通过截图方式，大家可以重点阅读对应的文字说明，并在文末进行了总结。

1、官方文档看 EventParser

首先我们先从官方文档来看 EventParser 的整体设计，其架构设计图如下所示：

上述图罗列出了 EventParser 的整体工作流程图，其关键步骤如下：

从 Log Position 管理器中获取上一次解析的日志位点。
向 Mysql Master 节点发送 BINLOG_DUMP 请求。
Mysql Master 节点从 Slave 端传入的日志位点开始向从节点推送 binlog 日志。
Slave 接收 binlog 日志，调用 BinlogParser 解析 binlog日志。
将解析后的结构化数据传入到 EventSink 组件。
定时记录解析 binlog 的日志，以便重启后继续进行增量订阅。
上图中还罗列一个HA 特性，即需要同步的 Master 如果宕机，可以从它的其他从节点继续同步 binlog 日志，避免单点故障。

官方文档有助于理解 EventParser 组件的实现原理，但关于如何使用 EventParser 的篇幅较少，故接下来将从源码的角度来反推 EventParser 的特性以及详细的工作实现原理，以便指导我们如何更好的使用 EventParser。

2、源码剖析 EventParser 初始化

从上篇文章我们即可得知，EventParser 组件是 Canal Instance 的四大核心组件之一，那本节的故事就从 CanalInstanceWithManager 的 initEventParser 方法开始。

Step1：获取数据库的连接信息，上面的代码就是集合的基本操作，但从上面的代码可以窥探如何配置数据库相关的地址信息。配置 canal instance 中数据库的地址，用户名密码有如下几种方式（CanalParamter）：

单库场景配置方式一：CanalParmeter 中提供了 masterAddress、masterUsername、masterPassword、standbyAddress、standbyUsername、standbyPassword 6 个属性分别用来指定主库与从库的信息，配置了从库的目的是提供 HA 机制。
单库场景配置方式二：CanalParamter 提供的 List
dbAddresses 方式进行配置，该集合的第一个元素为主库地址、第二个元素为从库地址，其数据库用户名通过 dbUsername、dbPassword 来配置。
多库场景：CanalParmeter 提供了 List< List< DataSourcing>> groupDbAddresses 属性用来设置 mysql 组，例如 MySQL 分库分表。groupDbAddresses 的第一个元素为主库的地址列表，第二个元素为从库的地址列表。

温馨提示：这里的用户名与密码是在对应服务器用于进行 binlog 日志同步的账号信息。

关于多库场景的配置，再详细举例如下：

其对应的初始化代码如下：

CanalInstanceWithManager#initEventParser

Step2：根据配置的 MySQL 构建 EventParser 实例。这里有如下几个关键点：

如果配置的 MySQL 地址是组方式则会创建 GroupEventParser，其内部会维护一个 EventParser 列表。
通过调用 doInitEventParser 方法创建 EventParser 实例。

接下来我们将重点查看 doInitEventParser 的实现细节。

Step3：从这里可以看出 Canal 目前并不支持 Oracle 数据，只支持 MySQL 与本地 binlog 文件(直接根据 binlog 日志文件解析)。

温馨提示：接下来将重点探讨基于 MySQL binlog 日志，并且会忽略与阿里云相关的 RDS 、tsdb 等数据库辅助支持，只关系与开源 MySQL 相关的处理逻辑。

Step3：MySQL 的 binlog 事件解析器实现类为 MysqlEventParser，这里我们重点来阐述一下这些参数的含义：

destination
Canal Instance 实例的名称。
connectionCharset
字符集，解析 binlog 时会将指定的字节数据使用该编码级进行转换，默认为UTF-8。
connectionCharsetNumber
字符集的数字表现形式，UTF8对应的值为 33，该值在与 MySQL 的交互协议包中需要被用到，这里 Canal 处理的不是特别好，最好该属性设置为只读，由 connectionCharset 联动进行设置。
defaultConnectionTimeoutInSeconds
MySQL 默认连接超时时间，因为 Canal 会伪装为 MySQL 服务器的 Slave 节点，需要向 MySQL Master 发送请求，故需要先创建链接，这里就是创建连接的默认超时时间，默认为 30s。
sendBufferSize
用于网络通道发送端缓存区，目前在 Canal 中网络通道的实现类为 BioSocketChannelPool、NettySocketChannelPool，从代码的角度来看，目前这个参数并不会生效，即使用操作系统的默认值。
receiveBufferSize
用于忘了通道接收缓存区大小，目前同 sendBufferSize 参数，并不会生效。
detectingEnable
是否开启心跳检测，默认为开启。
detectingSQL
心跳检测语句，例如 select 1，show master status 等。
detectingIntervalInSeconds
心跳间隔检测，默认为 3s。
slaveId
从服务器的 id，在同一个 MySQL 复制组内不能重复。

Step4：如果设置了 CanalPrameter 的 Listpositions 属性，则将其解析为 EntryPosition 实体，我们来看一下如何表征 binlog 日志的位点信息。

其主要的核心参数如下：

long timestamp
时间戳，用时间戳来表示位置
String journalName
binlog 日志的文件名，例如 mysql-bin.000001。
Long position
使用偏移量来表示具体位点。
long serverId
设置 master 的 id。
String gtid
全局事务ID。

温馨提示：实践指导，CanalParameter 的 List< String> positions 不支持组模式，只能设置一组，即第一个元素为主，第二个元素可以为从节点，该属性非必填。

Step5：继续设置参数，具体看一下各个参数的含义：

fallbackIntervalInSeconds
如果 MySQL 主节点宕机，Canal 支持切换到其从节点继续同步 binlog 日志，但为了数据的完整性，可以设置一个回退时间，即会造成数据重复下发，但尽量不丢失，该值默认为 60s。
profilingEnabled
是否开启性能采集，主要采集的是一批日志经过 EventSink 组件处理到完成存入EventStore 的时间消耗。
filterTableError
是否忽略表过滤异常，默认为 false，表过滤会在后续文章中详细介绍。
parallel
解析、canal 接入 prometheus 采集监控数据是否支持并发，默认为 false。
isGTIDMode
是否开启 gtid 模式

Step6：继续填充解析器相关参数，其重点实现如下：

transactionSize
Canal 提供了一种机制，尝试将一个数据库事务中所有的变更日志一起进行处理，这个为处理缓存事务日志的缓存区长度，默认为 1024。
logPositionManager
初始化日志位点管理器，Canal 提供了基于内存、zookeeper、内存与zookeepr混合管理器等日志位点管理器，这个后续会详细介绍。
AviaterRegexFilter
提供了基于 aviater 的正则表达式，对 table 名称进行过滤。
blackFilter
canal 提供了黑名单配置，提供黑名单正则表达式对 table 名称进行过滤。

Step7：如果解析器是 MySQL 解析器，提供了 HA 机制，即如果 MySQL Master 宕机，Canal 还能主动切换到 MYSQL Slave 节点，继续同步 binlog 日志。

3、EventParser 工作流程详解

上面已经详细介绍了EventParser 的初始化过程，有助于大家对 CanalInstance 相关配置参数的理解，本节将相信介绍 EventParser 的工作流程，其实现代码入口为 EventParser 的 start 方法。本文重点将探究 MySQL binlog 日志的解析，故其实现类为：MysqlEventParser。

MysqlEventParser 的 start 方法代码如下：

主要调用的是其父类的 start 方法。接下来对其进行详细解读。

Step1：创建环形缓存区，其主要的作用是 Canal 在解析 binlog 日志后，会尽量尝试将一个数据库事务所产生的全部变更日志（一个事务所有变更数据）当成一个整体提交给 EventSink 组件，从而 Canal 的消费方能一次将一个事务的数据全部同步，数据的完整性得到了保证。

温馨提示：关于环形缓存区的具体实现细节将在下文详细介绍，这里先简单说一下 Canal 目前无法百分之百保证一个事务的数据就一定是一次消费，如果一个事务产生的变更日志超过了环形缓存区的容量，则会被强制提交消费，一个事务的数据会被分开消费，默认环形缓存区的长度为 1024.

Step2：构建一个 binlog 解析器，该方法在 AbstractEventParser 中为一个抽象方法，具体的实现在其子类中，其代码截图如下：

AbstractMysqlEventParser，在 MySQL binlog 解析的实现类为 LogEventConvert，所处的模块为 parse，该部分是整个 Canal EventParser 的核心，将在后续文章中单独详细介绍。

Step3：启动一个独立的线程来负责 binlog 的解析，其线程包含了 Canal Instance 的 destination、address 等信息，方便利用 jstack 去诊断 binlog 解析相关问题。接下来就是解读该线程的 run 方法，从而探究 binlog 的解析流程。

Step31：首先创建一条到需要解析 binlog 日志的服务器，例如需要同步 192.168.1.166:3306 这个数据库实例的 binlog 日志，那 Canal 首先会使用拥有该库复制权限的账号去创建一条TCP连接，本文并不会详细去介绍这里的实现细节，这里代表一个领域，即需要知晓 MySQL 通讯协议，通过TCP与MySQL建立连接，并按照 MySQL 通讯协议发送命令，例如 select、dump 等请求，这个后续在学完 Canal 等核心组件后，可能会深入学习该部分的内容，这里我重点点出其实现的几个关键要点：

首先创建一条TCP连接，连接到 MySQL 服务器，Canal 提供了 BIO 与 Netty 两种实现方式。
TCP 三次握手后成功建立TCP连接后，需要与 MySQL 进行握手，完成协议约定，客户端登录校验等，例如握手实现代码见：MysqlConnector negotiate。
一言以蔽之，MySqlConnection 的职责就是实现一个 MySQL 客户端。其效果等同于实现我们常用的 SQL 连接客户端，关于这方面的编程其实不难，如果大家有志成为一名数据库中间件方面的技术人员，只需按照 MySQL 官方文档中有关通讯协议即可。

Step32：发送心跳包，这里的关键实现点如下：

利用 Timer 实现定时调度，心跳包发送间隔通过 detectingIntervalInSeconds 指定。
心跳包主要是构建一个 CanalEntry.Entry,其类型为EntryType.HEARTBEAT。
心跳包并不是发送给远端 MySQL 服务器，而是将 Entry 下发到 EventSink 组件。
该心跳包的用意合作，在这里先留一个伏笔，后续文章会一一揭晓。

Step33：执行发送 dump 命令正式从 MySQL 服务器接收 binlog 日志之前的准备工作，具体准备工作如下：

首先再创建一条专属数据库连接，主要用于查找 MySQL 的一些配置信息，统称元数据。
向 MySQL 服务器发送 show variables like 'binlog_format‘ 语句查询服务端配置的 binlog 格式，MySQL 支持 STATEMENT、ROW、MIXED 三种模式。
向 MySQL 服务端发送 show variables like 'binlog_row_image' 语句查询服务器端配置的 binlog_row_image。

扩展阅读：binlog_format 我相信大家都不陌生，对 binlog_row_image 见过的估计比较少，那 binlog_row_image 有何作用呢？

binlog_row_image 主要是在 binlog_format 为 ROW 模式下，控制记录 binlog 事件的方式，binlog 的作用是记录数据的变化，例如 update 请求，需要记录一行记录变化之前的数据以及变化后的数据，在 binlog event 分别用 before 、after 记录变化前后的数据，但有一个问题，是只发生变化的字段的前后值呢，还是记录一行中所有字段修改前后的值呢？故引入了 binlog_row_image，该值支持如下选项：