引言

• 在线解析：直接连接MySQL服务器解析binlog文件。

• 离线解析：处理存储在本地的binlog文件。

工具调研

阿里云DMS数据追踪

优点

缺点

mysqlbinlog

优点

缺点

1. 没有列名的补全，只有列对应的位置信息

2. 过滤能力很有限，比如不支持表名过滤，sql类型过滤

#231129 10:00:03 server id 1145804703  end_log_pos 13184 CRC32 0x410c1d23       Table_map: `school`.`bureau_statistics` mapped to number 159
# Columns(BIGINT NOT NULL,
#         BIGINT NOT NULL,
#         BIGINT NOT NULL,
#         BIGINT,
#         INT,
#         INT,
#         INT,
#         INT,
#         INT,
#         INT,
#         DECIMAL(16,2),
#         DATETIME NOT NULL,
#         DATETIME NOT NULL)
# at 13184
#231129 10:00:03 server id 1145804703  end_log_pos 13372 CRC32 0x70f00704       Update_rows_v1: table id 159 flags: STMT_END_F
### UPDATE `school`.`bureau_statistics`
### WHERE
###   @1=430
###   @2=198
###   @3=164325514417812
###   @4=2260
###   @5=43
###   @6=42404
###   @7=2216
###   @8=940
###   @9=35
###   @10=0
###   @11=11695475.00
###   @12='2023-09-12 22:17:13'
###   @13='2023-11-29 09:55:00'
### SET
###   @1=430
###   @2=198
###   @3=164325514417812
###   @4=2260
###   @5=43
###   @6=42404
###   @7=2216
###   @8=940
###   @9=35
###   @10=0
###   @11=11695475.00
###   @12='2023-09-12 22:17:13'
###   @13='2023-11-29 10:00:00'
# at 13372

不过这就是binlog最原始的样子，binlog本身只记录了列的位置信息。因为这对于数据复制和数据恢复来说已经足够了，保存的信息越少，对于存储和传输效率来说肯定越好。

binlog2sql

PyMySQL==0.9.3
wheel==0.29.0
mysql-replication==0.21

优点

1. 支持库表维度过滤

2. 支持sql类型过滤

3. 支持生成回滚SQL

缺点

1. 不支持离线解析

2. 性能较差

3. 社区不活跃

bingo2sql

ERRO[2023-12-04T17:14:26+08:00] 读取表结构文件失败(ddl.sql): 解析失败: [ddl:1273]Unknown collation: 'utf8mb4_0900_ai_ci'  file=bingo2sql.go line=182

优点

缺点

1. 离线表结构读取存在一些小问题

2. 不支持多binlog文件的顺序解析，缺少对云rds的支持

my2sql

优点

缺点

canal2sql

设计初衷

设计目标

1. 支持binlog的在线解析、离线解析

2. 列还原同时支持在线模式和离线模式

3. 支持生成回滚SQL

4. 支持解析范围控制
1. 时间维度
2. position维度

5. 支持对象过滤
1. 库表维度
2. sql类型维度

6. 支持多binlog文件顺序解析

7. 支持云rds binlog文件列表解析

8. 支持分析统计功能
1. 耗时事务统计
2. 大事务统计
3. 表维度/库维度 更新数据量统计
4. qps统计

核心改造点

1. 启动dump线程

2. 连接底层数据源

3. dump binlog

4. 解析binlog

5. 消费binlog

• 在线解析是连接mysql，而离线解析则是连接对应的binlog文件，

• 在线解析是从mysql的实时binlog读取数据，而离线解析是从binlog文件路径读取数据

解析范围控制逻辑收敛

因为canal设计之初就是为了实时监听binlog，所以完全没有考虑终止条件，如果单纯的设计一个离线解析的产品，怎么可能不支持这个功能呢

public class LogEventFilter {
   public LogEvent filter(LogEvent event) {
        if (event == null) {
            return null;
        }
        String logFileName = event.getHeader().getLogFileName();
        if (startTime != null && event.getWhen() < startTime.getTime() / 1000) {
            return null;
        }
        if (endTime != null && event.getWhen() > endTime.getTime() / 1000) {
            shutdownNow();
            return null;
        }
        // binlog 文件需要从头遍历，而online模式可以直接从指定位置读
        if (startFile == null) {
            if (startPosition != null && event.getLogPos() < startPosition) {
                return null;
            }
        } else {
            if (startFile.equals(logFileName) && startPosition != null && event.getLogPos() < startPosition) {
                return null;
            }
        }

        if (endFile == null) {
            if (endPosition != null && event.getLogPos() > endPosition) {
                shutdownNow();
                return null;
            }
        } else {
            if (endFile.equals(logFileName) && endPosition != null && event.getLogPos() > endPosition) {
                shutdownNow();
                return null;
            }
        }
        return event;
    }

    private void shutdownNow() {
        System.exit(1);
    }
}

终止条件支持

public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implements MultiStageCoprocessor {
    private class SinkStoreStage implements EventHandler<MessageEvent>, LifecycleAware {

        public void onEvent(MessageEvent event, long sequence, boolean endOfBatch) throws Exception {
            try {
                LogEvent logEvent = event.getEvent();
                if (event.getEntry() != null) {
                    if (logEventFilter == null || logEventFilter.filter(logEvent) != null) {
                        transactionBuffer.add(event.getEntry());
                    }
                }

                if (connection instanceof MysqlConnection && logEvent.getSemival() == 1) {
                    // semi ack回报
                    ((MysqlConnection) connection).sendSemiAck(logEvent.getHeader().getLogFileName(),
                        logEvent.getHeader().getLogPos());
                }

                // clear for gc
                event.setBuffer(null);
                event.setEvent(null);
                event.setTable(null);
                event.setEntry(null);
                event.setNeedDmlParse(false);

            } catch (Throwable e) {
                exception = new CanalParseException(e);
                throw exception;
            }
        }
    }
}

增加支持通过本地DDL文件还原列

public class TableMetaCache {
    public TableMetaCache(String fileName){

        String createDDLs;
        try {
            createDDLs = FileUtils.readFileToString(new File(fileName));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        MemoryTableMeta memoryTableMeta = new MemoryTableMeta();

        String[] createDDLList = createDDLs.split(";");
        for (String createDDL : createDDLList) {
            memoryTableMeta.apply(DatabaseTableMeta.INIT_POSITION, null, createDDL, null);
        }
        this.tableMetaDB = CacheBuilder.newBuilder().build(new CacheLoader<String, TableMeta>() {

            @Override
            public TableMeta load(String name) throws Exception {
                String[] names = StringUtils.split(name, "`.`");
                String schema = names[0];
                String table = names[1];
                return memoryTableMeta.find(schema, table);
            }
        });

    }
}

    @Override
    protected void preDump(ErosaConnection connection) {
        if (StringUtils.isNotBlank(ddlFile)) {
            tableMetaCache = new TableMetaCache(ddlFile);
        } else {
            metaConnection = buildMysqlConnection();
            try {
                metaConnection.connect();
            } catch (IOException e) {
                throw new CanalParseException(e);
            }
            if (tableMetaTSDB != null && tableMetaTSDB instanceof DatabaseTableMeta) {
                ((DatabaseTableMeta) tableMetaTSDB).setFilter(eventFilter);
                ((DatabaseTableMeta) tableMetaTSDB).setBlackFilter(eventBlackFilter);
                ((DatabaseTableMeta) tableMetaTSDB).setSnapshotInterval(tsdbSnapshotInterval);
                ((DatabaseTableMeta) tableMetaTSDB).setSnapshotExpire(tsdbSnapshotExpire);
                ((DatabaseTableMeta) tableMetaTSDB).init(destination);
            }
            tableMetaCache = new TableMetaCache(metaConnection, tableMetaTSDB);
        }
        ((LogEventConvert) binlogParser).setTableMetaCache(tableMetaCache);
    }

支持多协议离线binlog文件的读取

public class BinlogFileConnection implements ErosaConnection {

    public void dump(String binlogfilename, Long binlogPosition, MultiStageCoprocessor coprocessor) throws IOException {
        try (BinlogFileLogFetcher fetcher = new BinlogFileLogFetcher(bufferSize)) {
            LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
            LogContext context = new LogContext();
            fetcher.open(binlogFile, binlogPosition);
            context.setLogPosition(new LogPosition(binlogfilename, binlogPosition));
            LogEvent event = null;
            while (fetcher.fetch()) {
                event = decoder.decode(fetcher, context);
                if (event == null) {
                    continue;
                }

                if (!coprocessor.publish(event)) {
                    break;
                }
            }
        }
    }
}

public final class BinlogFileLogFetcher extends LogFetcher {
    public void open(String urlStr, final long filePosition) throws FileNotFoundException, IOException {

        URL url = new URL(urlStr);
        URLConnection urlConnection = url.openConnection();
        in = urlConnection.getInputStream();
        // 省略后续代码
    }
}

16:07:14.135 [destination = null , address = rm-bp1yjb2rh8249f1ft90130.mysql.rds.aliyuncs.com/192.168.111.235:3306 , EventParser] ERROR c.g.z.c.p.RemoteBinlogEventWithLocalDDLParser - dump address rm-bp1yjb2rh8249f1ft90130.mysql.rds.aliyuncs.com/192.168.111.235:3306 has an error, retrying. caused by
java.lang.IllegalArgumentException: limit excceed: 3201
	at com.taobao.tddl.dbsync.binlog.LogBuffer.getUint32(LogBuffer.java:532) ~[canal.parse.dbsync-1.1.5.jar:na]
	at com.taobao.tddl.dbsync.binlog.event.LogHeader.processCheckSum(LogHeader.java:299) ~[canal.parse.dbsync-1.1.5.jar:na]
	at com.taobao.tddl.dbsync.binlog.event.LogHeader.<init>(LogHeader.java:228) ~[canal.parse.dbsync-1.1.5.jar:na]
	at com.taobao.tddl.dbsync.binlog.LogDecoder.decode(LogDecoder.java:100) ~[canal.parse.dbsync-1.1.5.jar:na]
	at com.github.zhuchao941.canal2sql.RemoteBinLogConnection.dump(RemoteBinLogConnection.java:94) ~[classes/:na]
	at com.alibaba.otter.canal.parse.inbound.AbstractEventParser$1.run(AbstractEventParser.java:262) ~[canal.parse-1.1.5.jar:na]
	at java.lang.Thread.run(Thread.java:750) [na:1.8.0_333]

解决官方离线解析的Bug

public class LogBuffer {
    public final long getUint32(final int pos) {
        final int position = origin + pos;

        if (pos + 3 >= limit || pos < 0) throw new IllegalArgumentException("limit excceed: "
                                                                            + (pos < 0 ? pos : (pos + 3)));

        byte[] buf = buffer;
        return ((long) (0xff & buf[position])) | ((long) (0xff & buf[position + 1]) << 8)
               | ((long) (0xff & buf[position + 2]) << 16) | ((long) (0xff & buf[position + 3]) << 24);
    }
}

public final class LogHeader {
    private void processCheckSum(LogBuffer buffer) {
        if (checksumAlg != LogEvent.BINLOG_CHECKSUM_ALG_OFF && checksumAlg != LogEvent.BINLOG_CHECKSUM_ALG_UNDEF) {
            crc = buffer.getUint32(eventLen - LogEvent.BINLOG_CHECKSUM_LEN);
        }
    }
}

1. LogDecoder在解码binlog event时，会根据mysql的配置来决定要不要去获取binlog event的校验和CRC（不过貌似只有读取，没有看到用来做校验）

2. 解码是分步进行的，先解码header，再通过header里的event_length解码body

3. 所以对于数据流大小的判断也是分步的，满足header_length了，就可以先解码header

4. 获取校验和的逻辑在解码header的过程中

5. 而校验和在数据流是位于整个binlog event的最后4个字节

1. 自定义一个LogFetcher在上层保证event的完整性

2. 自定义LogDecoder，在下层发现buffer里的event不完整，那就交由上层继续填充

public final class BinlogFileLogFetcher extends LogFetcher {
    public boolean fetch() throws IOException {
        if (in == null) {
            return false;
        }
        // 先判断当前buffer里的数据够不够一次事件
        if (limit < FormatDescriptionLogEvent.LOG_EVENT_HEADER_LEN) {
            // 如果不到一个头就先读一个头
            int length = readFully(in, buffer, origin + limit, FormatDescriptionLogEvent.LOG_EVENT_HEADER_LEN);
            if (length < 0) {
                return false;
            }
            limit += length;
        }
        long eventLen = getUint32(LogEvent.EVENT_LEN_OFFSET);
        if (limit >= eventLen) {
            // 足够一次完整事件，无需从底层读取数据
            return true;
        }
        // 先确保buffer足够
        ensureCapacity((int) eventLen);
        // 从底层读满一个事件
        int length = readFully(in, buffer, origin + limit, (int) (eventLen - FormatDescriptionLogEvent.LOG_EVENT_HEADER_LEN));
        if (length < 0) {
            return false;
        }
        limit += length;
        return true;
    }
}

aliyun模式

public class AliyunBinlogFileConnection implements ErosaConnection {
    public void dump(String binlogfilename, Long binlogPosition, MultiStageCoprocessor coprocessor) throws IOException {
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssZ");
        String[] split = instanceId.split("\\\\|");
        DescribeBinlogFilesResponse describeBinlogFilesResponse = AliyunClient.describeBinlogFiles(new DescribeBinlogFilesRequest().setDBInstanceId(split[0]).setStartTime(format.format(startTime)).setEndTime(format.format(endTime)), ak, sk);
        List<DescribeBinlogFilesResponseBody.DescribeBinlogFilesResponseBodyItemsBinLogFile> binLogFiles = describeBinlogFilesResponse.getBody().getItems().getBinLogFile();
        if (split.length > 1) {
            binLogFiles = binLogFiles.stream().filter(binLogFile -> binLogFile.getHostInstanceID().equals(split[1])).collect(Collectors.toList());
        }
        System.out.println("find binLogFile num:" + binLogFiles.size());
        binLogFiles.sort(Comparator.comparing(o -> o.logBeginTime));
        for (DescribeBinlogFilesResponseBody.DescribeBinlogFilesResponseBodyItemsBinLogFile binlogFile : binLogFiles) {
            System.out.println(String.format("fileName:%s, fileSize:%s, time:%s-%s", binlogFile.getLogFileName(), binlogFile.getFileSize(), binlogFile.getLogBeginTime(), binlogFile.getLogEndTime()));
        }

        for (DescribeBinlogFilesResponseBody.DescribeBinlogFilesResponseBodyItemsBinLogFile logFile : binLogFiles) {
            try (BinlogFileLogFetcher fetcher = new BinlogFileLogFetcher(bufferSize)) {
                LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
                LogContext context = new LogContext();
                String downloadLink = internal ? logFile.intranetDownloadLink : logFile.downloadLink;
                fetcher.open(downloadLink, binlogPosition);
                System.out.println(String.format("# download file %s from %s", logFile.logFileName, downloadLink));
                context.setLogPosition(new LogPosition(binlogfilename, binlogPosition));
                LogEvent event = null;
                while (fetcher.fetch()) {
                    event = decoder.decode(fetcher, context);
                    if (event == null) {
                        continue;
                    }

                    if (!coprocessor.publish(event)) {
                        break;
                    }
                }
            }
        }
    }
}

总结

异常场景记录

1. 在表的最后一列之后增加列，不影响

2. 在其他位置增加列，会导致之前的binlog解析错乱

3. 删除最后一列，不影响

4. 删除其他列，会导致之前的binlog解析错乱

参考

1. MySQL-binlog解析工具

2. 数据同步工具之FlinkCDC/Canal/Debezium对比

3. MySql-Binlog 协议详解 - 报文篇

4. 带你玩转 MySql - 协议篇

5. 解读 MySQL Client/Server Protocol: Connection & Replication

6. 关于binary log那些事——认真码了好长一篇

7. MySQL的半同步是什么？

8. MySQL Binlog 源码入门

9. MySQL如何传输二进制日志

10. 美团DB数据同步到数据仓库的架构与实践