DM 源码阅读系列文章（四）dump/load 全量同步的实现

- [先将所有 `non_innodb_table` 分为 `num_threads` 组，分组方式是遍历这些表，依此将遍历到的表加入到当前数据量最小的分组，尽量保证每个分组内的数据量相近](https://github.com/pingcap/mydumper/blob/9493dd752b9ea8804458e56a955e7f74960fa969/mydumper.c#L1574-L1586)。
- 上述得到的每个分组内会包含一个或多个非 InnoDB 表，如果配置了 `rows-per-file` 选项，会对每张表进行 `chunks` 估算，[对于每一张表，如果估算结果包含多个 chunks，会将子任务进一步按照 `chunks` 进行拆分，分发 `chunks` 数量个子任务](https://github.com/pingcap/mydumper/blob/9493dd752b9ea8804458e56a955e7f74960fa969/mydumper.c#L3033-L3046)，[如果没有 `chunks` 划分，分发为一个独立的子任务](https://github.com/pingcap/mydumper/blob/9493dd752b9ea8804458e56a955e7f74960fa969/mydumper.c#L3047-L3057)。
- 注意，在该模式下，子任务会 [发送到 `queue_less_locking`](https://github.com/pingcap/mydumper/blob/9493dd752b9ea8804458e56a955e7f74960fa969/mydumper.c#L3059)，并在编号为 `num_threads` ~ 2 * `num_threads` 的子线程中处理任务。
    - `less_locking_threads` 任务执行完成之后，[主线程就会 UNLOCK TABLES /* FTWRL */ 释放锁](https://github.com/pingcap/mydumper/blob/9493dd752b9ea8804458e56a955e7f74960fa969/mydumper.c#L1648-L1654)，这样有助于减少锁持有的时间。主线程根据 `conf.unlock_tables` 来判断非 InnoDB 表是否全部导出，[普通工作线程](https://github.com/pingcap/mydumper/blob/9493dd752b9ea8804458e56a955e7f74960fa969/mydumper.c#L639-L641) 或者 [queue_less_locking](https://github.com/pingcap/mydumper/blob/9493dd752b9ea8804458e56a955e7f74960fa969/mydumper.c#L787-L789) 工作线程每次处理完一个非 InnoDB 表任务都会根据 `non_innodb_table_counter` 和 `non_innodb_done` 两个变量判断是否还有没有导出结束的非 InnoDB 表，如果都已经导出结束，就会向异步队列 `conf.unlock_tables` 中发送一条数据，表示可以解锁全局锁。
    - 每个 `less_locking_threads` 处理非 InnoDB 表任务时，会先 [加表锁](https://github.com/pingcap/mydumper/blob/9493dd752b9ea8804458e56a955e7f74960fa969/mydumper.c#L771-L778)，导出数据，最后 [解锁表锁](https://github.com/pingcap/mydumper/blob/9493dd752b9ea8804458e56a955e7f74960fa969/mydumper.c#L803)。

- [遍历每一张非 InnoDB 表，同样对每张表进行 `chunks` 估算，如果包含多个 `chunks`，按照 chunks 个数分发同样的子任务数；如果没有划分 `chunks`，每张表分发一个子任务。所有的任务都分发到 conf->queue 队列。](https://github.com/pingcap/mydumper/blob/9493dd752b9ea8804458e56a955e7f74960fa969/mydumper.c#L1606-L1614)

- 与未开启 `less-locking` 选项的非 InnoDB 表的处理相同，同样是 [按照表分发子任务，如果有 `chunks` 子任务会进一步细分](https://github.com/pingcap/mydumper/blob/9493dd752b9ea8804458e56a955e7f74960fa969/mydumper.c#L1616-L1620)。

```go

type fileJob struct {

   schema    string

   table     string

   dataFile  string

   offset    int64 // 表示读取文件的起始 offset，如果没有 checkpoint 断点信息该值为 0

   info      *tableInfo // 保存原库表，目标库表，列名，insert 语句 column 名字列表等信息

}

```

```go

// worker 工作线程内分发给内部工作协程的任务结构

type dataJob struct {

   sql         string // insert 语句, insert into <table> values (x, y, z), (x2, y2, z2), … (xn, yn, zn);

   schema      string // 目标数据库

   file        string // SQL 文件名

   offset      int64 // 本次导入数据在 SQL 文件的偏移量

   lastOffset  int64 // 上一次已导入数据对应 SQL 文件偏移量

}

// SQL 语句执行协程

doJob := func() {

   for {

       select {

       case <-ctx.Done():

           return

       case job := <-jobQueue:

           sqls := []string{

               fmt.Sprintf("USE `%s`;", job.schema), // 指定插入数据的 schema

               job.sql,

               checkpoint.GenSQL(job.file, job.offset), // 更新 checkpoint 的 SQL 语句

           }

           executeSQLInOneTransaction(sqls) // 在一个事务中执行上述 3 条 SQL 语句

       }

   }

}

// worker 主线程

for {

   select {

   case <-ctx.Done():

       return

   case job := <-fileJobQueue:

       go doJob()

       readDataFileAndDispatchSQLJobs(ctx, dir, job.dataFile, job.offset, job.info)

   }

}

```

- 列值转换：需要对输入 SQL 进行解析拆分为每一个 field，对需要转换的 field 进行转换操作，然后重新拼接起 SQL 语句。详细重写流程见 [reassemble](https://github.com/pingcap/dm/blob/25f95ee08d008fb6469f0b172e432270aaa6be52/loader/convert_data.go#L293) 函数。

- 库表路由：这种场景下只需要 [替换源表到目标表](https://github.com/pingcap/dm/blob/25f95ee08d008fb6469f0b172e432270aaa6be52/loader/loader.go#L263) 即可。