PolarDB-X 源码解读（八）：Global Binlog 的一生

本篇将对 Binlog 的产生，以及如何通过系统处理并最终生成 Global Binlog 的过程进行分析。

概述：

Global Binlog的一生是指从原始 Binlog 产生，到最终Global Binlog生成期间产生的故事，本文会详细介绍Binlog拉取，数据整形合并，以及最终生成Global Binlog期间经过的关键流程。 Global Binlog涉及到Task和Dumper组件，分别针对Binlog整形合并以及落盘过程，下面会介绍这两个组件对binlog处理的相关代码。

Task组件：

当用户向PolarDB-X写入数据时，最终数据会落盘到底层DN上，DN会产生原始binlog，task组件会拉取到原始binlog，整形合并成Global Binlog后，发送给下游Dumper组件，下面我们通过查看Task的核心代码来了解一下整个原始Binlog的处理流程。

我们可以以代码com.aliyun.polardbx.binlog.canal.core.BinlogEventProcessor (github地址：https://github.com/ApsaraDB/galaxycdc )为入口来查看整个binlog的关键入口class， 查看binlog是如何拉取并投递到下游handler中。

在handler中，会通过一些列的filter，对binlog进行处理。 在com.aliyun.polardbx.binlog.canal.core.handle.DefaultBinlogEventHandle代码中，我们可以看到如下逻辑

通过head对象的doNext方法，逐级遍历所有filter，最终对binlog event进行过滤和格式化处理，并输出逻辑event到下游。head对象为filter链表，初始化逻辑见下面代码。

Extractor 中的filter chain如图：

在BinlogExtractor初始化时，会将一些列Filter加入到列表中，binlog event会按顺序逐个经过这些filter。 com.aliyun.polardbx.binlog.extractor.filter.RtRecordFilter: 会记录当前event下游逻辑处理的rt。 com.aliyun.polardbx.binlog.extractor.filter.TransactionBufferEventFilter：处理事务相关event，并标记出tso事件

当收到commit event时，会尝试执行tryDoNext

事务的推送算法可以在 com.aliyun.polardbx.binlog.extractor.filter .TransactionStorage 类中查看。 com.aliyun.polardbx.binlog.extractor.filter.RebuildEventLogFilter 负责binlog event过滤和整形流程，具体逻辑可以查看handle方法。 整形的关键代码可以查看reformat方法。

Global binlog中，我们只关心QueryEvent、RowEvent(Insert/Update/Delete)和TableMapEvent，所以这里我们只处理了这些event的数据。

com.aliyun.polardbx.binlog.extractor.filter.MinTSOFilter 会对事务进行初步过滤。 数据最终会通过 com.aliyun.polardbx.binlog.extractor.DefaultOutputMergeSourceHandler 推送给下游事务合并代码 com.aliyun.polardbx.binlog.merge.MergeSource 中，sourceId唯一标记了当前数据流id，queue保存事务索引txnKey。

com.aliyun.polardbx.binlog.merge.LogEventMerger 中，会将所有DN对应的MergeSource的queue遍历提取出来

com.aliyun.polardbx.binlog.merge.MergeController 会保存对应的sourceId和数据，确保每个sourceId只会收到一条数据，下游在拉取数据时，会将保存在优先队列的数据pop出来，push到 com.aliyun.polardbx.binlog.collect.LogEventCollector 中，最终通过ringBuffer提供给com.aliyun.polardbx.binlog.transmit.LogEventTransmitter 来投递给下游Dumper组件。

Dumper组件

当Global Binlog系统启动时，优先启动Task组件，监听端口。Dumper组件会尝试连接Task。 Dumper组件收到Task推送的Global Binlog后，会对binlog进行最后的细节处理，并且把处理好的结果写入磁盘。下面我们从Dumper组件消费Task推送数据的入口来分析，整个binlog处理核心流程。 在 com.aliyun.polardbx.binlog.dumper.dump.logfile.LogFileGenerator 类的start方法中，该方法会启动grpc连接Task端口。

代码里实现了 com.aliyun.polardbx.binlog.rpc.TxnMessageReceiver 接口，该接口会消费上游推送过来的所有数据，并在consume方法中，处理相关Global Binlog position等相关信息，并最终通过 com.aliyun.polardbx.binlog.dumper.dump.logfile.BinlogFile 写入到磁盘中。

private void consume(TxnMessage message, MessageType processType) throws IOException, InterruptedException {
            ...
        switch (processType) {
        case BEGIN:
             ...
            break;
        case DATA:
             ...
            break;
        case END:
             ...
            break;
        case TAG:
            currentToken = message.getTxnTag().getTxnMergedToken();
            if (currentToken.getType() == TxnType.META_DDL) {
                ...
            } else if (currentToken.getType() == TxnType.META_DDL_PRIVATE) {
                ...
            } else if (currentToken.getType() == TxnType.META_SCALE) {
                ...
            } else if (currentToken.getType() == TxnType.META_HEARTBEAT) {
                ...
            } else if (currentToken.getType() == TxnType.META_CONFIG_ENV_CHANGE) {
                ...
            }
            break;
        default:
            throw new PolardbxException("invalid message type for logfile generator: " + processType);
        }
    }

在consome方法中会逐个处理事务和数据相关的事件，如果事件打标了系统之间交互的tag，会针对相应的tag做一定的逻辑处理。

小结：

本文对binlog的拉取、整形处理和最终落盘涉及到的关键流程进行了简单梳理。

文章来源：阿里云数据库