完结篇 | TiDB Binlog 源码阅读系列文章（九）同步数据到下游

上篇文章介绍了用于将 binlog 同步到 MySQL / TiDB 的 Loader package，本文往回退一步，介绍 Drainer 同步到不同下游的机制。

TiDB Binlog（github.com/pingcap/tidb-binlog）用于收集 TiDB 的 binlog，并准实时同步给下游。同步数据这一步重要操作由 Drainer 模块支持，它可以将 binlog 同步到 TiDB / MySQL / Kafka / File （增量备份）等下游组件。

对于 TiDB 和 MySQL 两种类型的下游组件，Drainer 会从 binlog 中还原出对应的 SQL 操作在下游直接执行；
对于 Kafka 和 File（增量备份）两种类型的下游组件，输出约定编码格式的 binlog。用户可以定制后续各种处理流程，如更新搜索引擎索引、清除缓存、增量备份等。TiDB Binlog 自带工具 Reparo 实现了将增量备份数据（下游类型为 File（增量备份））同步到 TiDB / MySQL 的功能。

本文将按以下几个小节介绍 Drainer 如何将收到的 binlog 同步到下游：

Drainer Sync 模块：Drainer 通过 Sync 模块调度整个同步过程，所有的下游相关的同步逻辑统一封装成了 Syncer 接口。
恢复工具 Reparo （读音：reh-PAH-roh）：从下游保存的 File（增量备份）中读取 binlog 同步到 TiDB / MySQL。

Drainer Sync 模块

Syncer

同步机制的核心是 Syncer 接口，定义如下：

// Syncer sync binlog item to downstream
type Syncer interface {
  // Sync the binlog item to downstream
  Sync(item *Item) error
  // will be close if Close normally or meet error, call Error() to check it
  Successes() <-chan *Item
  // Return not nil if fail to sync data to downstream or nil if closed normally
  Error() <-chan error
  // Close the Syncer, no more item can be added by `Sync`
  Close() error
}

其中 Sync 方法表示异步地向下游同步一个 binlog，对应的参数类型是 *Item，这是一个封装了 binlog 的结构体；Successes 方法返回一个 channel，从中可以读取已经成功同步到下游的 Item；Error 方法返回一个 channel，当 Syncer 同步过程出错中断时，会往这个 channel 发送遇到的错误；Close 用于关掉 Syncer，释放资源。

支持的每个下游类型在 drainer/sync 目录下都有一个对应的 Syncer 实现，例如 MySQL 对应的是 mysql.go 里的 MySQLSyncer，Kafka 对应的是 kafka.go 里的 KafkaSyncer。Drainer 启动时，会根据配置文件中指定的下游，找到对应的 Syncer 实现，然后就可以用统一的接口管理整个同步过程了。

Checkpoint

同步进程可能因为各种原因退出，重启后要恢复同步就需要知道上次同步的进度。在 Drainer 里记录同步进度的功能抽象成 Checkpoint 接口，其定义如下：

type CheckPoint interface {
  // Load loads checkpoint information.
  Load() error

  // Save saves checkpoint information.
  Save(int64) error

  // Pos gets position information.
  TS() int64

  // Close closes the CheckPoint and release resources, after closed other methods should not be called again.
  Close() error
}

从以上定义中可以看到，Save 的参数和 TS 的返回结果都是 int64 类型，因为同步的进度是以 TiDB 中单调递增的 commit timestamp 来记录的，它的类型就是 int64。

Drainer 支持不同类型的 Checkpoint 实现，例如 mysql.go 里的 MySQLCheckpoint，默认将 commit timestamp 写到 tidb_binlog 库下的 checkpoint 表。Drainer 会根据下游类型自动选择不同的 Checkpoint 实现，例如 TiDB / MySQL 的下游就会使用 MySQLCheckPoint，File（增量备份）则使用 PbCheckpoint。

在 Syncer 小节，我们看到 Syncer 的 Successes 方法提供了一个 channel 用来接收已经处理完毕的 binlog，收到 binlog 后，我们用 Checkpoint 的 Save 方法保存 binlog 的 commit timestamp 就可以记下同步进度，细节可查看源码中的 handleSuccess 方法。

Translator

Syncer 在收到 binlog 后需要将里面记录的变更转换成适合下游 Syncer 类型的格式，这部分实现在 drainer/translator 包。

以下游是 MySQL / TiDB 的情况为例。MySQLSyncer.Sync 会先调用 TiBinlogToTxn

将 binlog 转换成 loader.Txn 以便接入下层的 loader 模块（loader 接收一个个 loader.Txn 结构并还原成对应的 SQL 批量写入 MySQL / TiDB）。

loader.Txn 定义如下：

// Txn holds transaction info, an DDL or DML sequences
type Txn struct {
  DMLs []*DML
  DDL  *DDL

  // This field is used to hold arbitrary data you wish to include so it
  // will be available when receiving on the Successes channel
  Metadata interface{}
}

Txn 主要有两类：DDL 和 DML。Metadata 目前放的就是传给 Sync 的 *Item 对象。DDL 的情况比较简单，因为 binlog 中已经直接包含了我们要用到的 DDL Query。DML 则需要遍历 binlog 中的一个个行变更，根据它的类型 insert / update / delete 还原成相应的 loader.DML。

Schema

上个小节中，我们提到了对行变更数据的解析，在 binlog 中编码的行变更是没有列信息的，我们需要查到对应版本的列信息才能还原出 SQL 语义。Schema 就是解决这个问题的模块。

在 Drainer 启动时，会调用 loadHistoryDDLJobs 从 TiKV 处查询截至当前时间所有已完成的 DDL Job 记录，按 SchemaVersion 升序排序（可以粗略认为这是一个单调递增地赋给每个 DDL 任务的版本号）。这些记录在 Syncer 中会用于创建一个 Schema 对象。在运行过程中，Drainer 每遇到一条 DDL 也会添加到 Schema 中。

binlog 中带有一个 SchemaVersion 信息，记录这条 binlog 生成的时刻 Schema 版本。在同步 Binlog 前，我们会先用这个 SchemaVersion 信息调用 Schema 的一个方法 handlePreviousDDLJobIfNeed。上一段中我们看到 Schema 从何处收集到有序的 DDL Job 记录，这个方法则是按顺序应用 SchemaVersion 小于等于指定版本的 DDL Job，在 Schema 中维护每个表对应版本的最新结构信息，去掉一些错误代码后实现大致如下：

func (s *Schema) handlePreviousDDLJobIfNeed(version int64) error {
  var i int
  for i = 0; i < len(s.jobs); i++ {
     if s.jobs[i].BinlogInfo.SchemaVersion <= version {
        _, _, _, err := s.handleDDL(s.jobs[i])
        if err != nil {
           return errors.Annotatef(err, "handle ddl job %v failed, the schema info: %s", s.jobs[i], s)
        }
     } else {
        break
     }
  }

  s.jobs = s.jobs[i:]

  return nil
}

对于每个符合条件的 Job，由 handleDDL 方法将其表结构 TableInfo 等信息更新到 Schema 中，其他模块就可以查询到表格当前最新的信息。

恢复工具

我们知道 Drainer 除了可以将 binlog 直接还原到下游数据库以外，还支持同步到其他外部存储系统块，所以我们也提供了相应的工具来处理存储下来的文件，Reparo 是其中之一，用于读取存储在文件系统中的 binlog 文件，写入 TiDB 中。本节简单介绍下 Reparo 的用途与实现，读者可以作为示例了解如何处理同步到文件系统的 binlog 增量备份。

Reparo

Reparo 可以读取同步到文件系统上的 binlog 增量备份并同步到 TiDB。

读取 binlog

当下游设置成 File（增量备份）时，Drainer 会将 Protobuf 编码的 binlog 保存到指定目录，每写满 512 MB 新建一个文件。每个文件有个编号，从 0 开始依次类推。文件名格式定义如下：

// BinlogName creates a binlog file name. The file name format is like binlog-0000000000000001-20181010101010
func BinlogName(index uint64) string {
  currentTime := time.Now()
  return binlogNameWithDateTime(index, currentTime)
}

// binlogNameWithDateTime creates a binlog file name.
func binlogNameWithDateTime(index uint64, datetime time.Time) string {
  return fmt.Sprintf("binlog-%016d-%s", index, datetime.Format(datetimeFormat))
}

文件的前缀都是 “binlog-”，后面跟一个 16 位右对齐的编号和一个时间戳。将目录里的文件按字母顺序排序就可以得到按编号排序的 binlog 文件名。从指定目录获取文件列表的实现如下：

// ReadDir reads and returns all file and dir names from directory
func ReadDir(dirpath string) ([]string, error) {
  dir, err := os.Open(dirpath)
  if err != nil {
     return nil, errors.Trace(err)
  }
  defer dir.Close()

  names, err := dir.Readdirnames(-1)
  if err != nil {
     return nil, errors.Annotatef(err, "dir %s", dirpath)
  }

  sort.Strings(names)

  return names, nil
}

这个函数简单地获取目录里全部文件名，排序后返回。在上层还做了一些过滤来去掉临时文件等。得到文件列表后，Reparo 会用标准库的 bufio.NewReader 逐个打开文件，然后用 Decode 函数读出其中的一条条 binlog：

func Decode(r io.Reader) (*pb.Binlog, int64, error) {
  payload, length, err := binlogfile.Decode(r)
  if err != nil {
     return nil, 0, errors.Trace(err)
  }

  binlog := &pb.Binlog{}
  err = binlog.Unmarshal(payload)
  if err != nil {
     return nil, 0, errors.Trace(err)
  }
  return binlog, length, nil
}

这里先调用了 binlogfile.Decode 从文件中解析出对应 Protobuf 编码的一段二进制数据然后解码出 binlog。

写入 TiDB

得到 binlog 后就可以准备写入 TiDB。Reparo 这部分实现像一个简化版的 Drainer 的 Sync 模块，同样有一个 Syncer 接口以及几个具体实现（除了 mysqlSyncer 还有用于调试的 printSyncer 和 memSyncer），所以就不再介绍。值得一提的是，这里也跟前面很多 MySQL / TiDB 同步相关的模块一样使用了 loader 模块。

小结

本文介绍了 Drainer 是如何实现数据同步的以及 Reparo 如何从文件系统中恢复增量备份数据到 MySQL / TiDB。在 Drainer 中，Syncer 封装了同步到各个下游模块的具体细节，Checkpoint 记录同步进度，Translator 从 binlog 中还原出具体的变更，Schema 在内存中维护每个表对应的表结构定义。

TiDB Binlog 源码阅读系列在此就全部完结了，相信大家通过本系列文章更全面地理解了 TiDB Binlog 的原理和实现细节。我们将继续打磨优化，欢迎大家给我们反馈使用过程中遇到的问题或建议；如果社区小伙伴们想参与 TiDB Binlog 的设计、开发和测试，也欢迎与我们联系 info@pingcap.com，或者在 Repo 中提 issue 讨论。

完结篇 | TiDB Binlog 源码阅读系列文章 （九）同步数据到下游