canal 源码解析系列-canal的通信数据结构

Posted 2021-09-17 犀牛饲养员

protocol模块 主要定义了client和server的通信协议。canal的数据传包含两部分，一部分是进行binlog订阅时，binlog转换为我们所定义的Message，第二部分是client与server进行传输的TCP协议。这两部分采用的都是采用protobuff格式。

先把一些基础的概念讲清楚，方便我们后面分析核心链路。

先看看Message类，

/**
 * @author zebin.xuzb @ 2012-6-19
 * @version 1.0.0
 *
 * 一个message就是get到的batch packet
 * 包含多个CanalEntry.Entry
 */
public class Message implements Serializable {

    private static final long      serialVersionUID = 1234034768477580009L;
    private long                   id;
    private List<CanalEntry.Entry> entries          = new ArrayList<>();//非raw类型使用这个
    // row data for performance, see:
    // https://github.com/alibaba/canal/issues/726
    private boolean                raw              = true;
    private List<ByteString>       rawEntries       = new ArrayList<>();//raw类型使用这个
    ...

Message封装了canal服务端和客户端的通信协议的数据结构，里面具体存放数据的是entries或者rawEntries，具体使用哪个是由是否raw类型决定的。raw类型相当于是一种为加工的类型，可以存储一些基本的数据结构，比如字节等。

Message的使用实例如下：

while (running) {
            try {
                connector.connect();
                connector.subscribe();
                while (running) {
                    try {
                        List<Message> messages = connector.getListWithoutAck(100L, TimeUnit.MILLISECONDS); // 获取message
                        if (messages == null) {
                            continue;
                        }
                        for (Message message : messages) {
                            long batchId = message.getId();
                            int size = message.getEntries().size();
                            if (batchId == -1 || size == 0) {
                                // try {
                                // Thread.sleep(1000);
                                // } catch (InterruptedException e) {
                                // }
                            } else {
                                // printSummary(message, batchId, size);
                                // printEntry(message.getEntries());
                                logger.info(message.toString());
                            }
                        }
...

CanalEntry又是啥呢？从代码里我们看到CananlEntry包含几种类型：

• CanalEntry.Column - 每个字段的数据结构（列）
• CanalEntry.Entry - 对应一个Event
• CanalEntry.EntryType - 打散后的事件类型，主要用于标识事务的开始，变更数据，结束
• CanalEntry.EventType - 事件类型，insert，update等
• CanalEntry.RowChange - 每行变更数据的数据结构
• CanalEntry.RowData - 对应mysql的一行数据

CanalEntry是基于protobuf定义的。Protobuf是Google开发的一种数据描述语言，类似于XML能够将结构化数据序列化，可用于数据存储、通信协议等方面。这里就不展开讲protobuf了。

Entry包含三个部分：

• header
• entryType
• storeValue

message Entry {
	/**协议头部信息**/
     Header						header 				= 1;
	///**打散后的事件类型**/ [default = ROWDATA]
	oneof entryType_present{
		EntryType					entryType			= 2;
	}

	/**传输的二进制数组**/
	bytes						storeValue			= 3;
}

这几部分分别又包含：

• Header

  • version [协议的版本号,default = 1]
  • logfileName [binlog文件名]
  • logfileOffset [binlog position]
  • serverId [服务端serverId]
  • serverenCode [变更数据的编码]
  • executeTime [变更数据的执行时间]
  • sourceType [变更数据的来源,default = MYSQL]
  • schemaName [变更数据的schemaname]
  • tableName [变更数据的tablename]
  • eventLength [每个event的长度]
  • eventType [insert/update/delete类型,default = UPDATE]
  • props [预留扩展]
  • gtid [当前事务的gitd]

• entryType [事务头BEGIN/事务尾END/数据ROWDATA/HEARTBEAT/GTIDLOG]

• storeValue [byte数据,可展开，对应的类型为RowChange，每行变更数据的数据结构]

  • tableId [tableId,由数据库产生]
  • eventType [数据变更类型,default = UPDATE]
  • isDdl [标识是否是ddl语句，比如create table/drop table]
  • sql [ddl/query的sql语句]
  • props [预留扩展]
  • ddlSchemaName [ddl/query的schemaName，会存在跨库ddl，需要保留执行ddl的当前schemaName]
  • rowDatas [具体insert/update/delete的变更数据，可为多条，1个binlog event事件可对应多条变更，比如批处理]