MySQL数据库同步工具的设计与实现

一、背景

　　在测试过程中，对于不同的测试团队，出于不同的测试目的，我们可能会有多套测试环境。在产品版本迭代过程中，根据业务需求，会对数据库的结构进行一些修改，如：新增表、字段、索引，修改表、字段索引等操作，在一些流程不规范的公司，开发人员不按照规范操作，不及时将这些修改数据库的 SQL 提交到 SVN/Git，当修改后的业务代码部署到新环境时就会引起错误，从而影响测试效率。换个角度再说，就算流程规范的大公司，核心业务都采取分库分表的架构，上千张表难道我们都采用手工执行 SQL 的方式去添加和修改字段吗？这样当然不妥，也许会有同学想到，我们可以采取使用脚本语言的方式批量更新和修改对应数据库，这样也是一种方式，但这种情况的前提是执行人员非常清楚两个数据库的差异，如果执行人员自己也不清楚两个数据库之间的差异呢？可能有的同学又想到可以把源数据库的结构和数据都导入到目标数据库当中，这样就解决了。这样看似可行，但实际不妥。前面我们说了，有多套测试环境，他们的作用可能不一样，举个例子：测试环境用于内部测试，联调环境用于和外部系统的联调，如果我们把测试环境的数据库结构和所有数据都导入联调环境，那么联调环境原有的数据不存在了，无法再和外部进行联调了，所以这也不是一种好的方式。

　　基于以上种种原因，一个数据库结构同步工具貌似是一个比较好的解决方案。

二、实现功能

基于以上的分析，该工具需要实现以下三个功能

• 分析（diff）：分析源数据库和目标数据库结构上的差异，在执行同步和拷贝前建议先执行分析来确定源数据库和目标数据库的差异；
• 同步（sync）：只同步数据库的结构，不同步数据；
• 拷贝（copy）： 对于数据没有要求的情况，可以直接使用拷贝将源数据库的数据库结构和数据全部导入目标数据库；

三、实现思路

具体流程如下：

• 对传入指令进行解析，包括：源数据库和目标数据库的 IP、端口、用户名、密码、数据库名以及执行动作（diff、sync、copy）；
• 分析 db，执行 SQL；
• 分析 db 下的表，执行 SQL；
• 分析表的字段和索引，执行 SQL；

四、分析过程

我们要对数据库的结构进行比对和分析，包括：数据库、数据库下面表、表中的字段和索引，那具体我们应该如何来进行分析和比对呢？

既然我们要做的是MySQL数据库的同步工具，那么我们对 MySQL 数据库就需要有深入一点的了解。在MySQL中，把 INFORMATION_SCHEMA 看作是一个数据库，确切说是信息数据库。其中保存着关于当前 MySQL 服务器所维护的所有其他数据库的信息。如数据库名，数据库的表、表的字段与索引以及访问权限等等。所以我们应该关注的是 INFORMATION_SCHEMA 中的以下几张表：

• SCHEMATA：提供了当前mysql实例中所有数据库的信息。SHOW DATABASES 的结果取之此表；
• TABLES：提供了关于数据库中的表的信息（包括视图）。详细表述了某个表属于哪个 schema，表类型，表引擎，创建时间等信息。SHOW TABLES FROM SCHEMANAME的结果取之此表；
• COLUMNS：提供了表中的列信息。详细表述了某张表的所有列以及每个列的信息。SHOW COLUMNS FROM SCHEMANAME.TABLENAME 的结果取之此表；
• STATISTICS：提供了关于表索引的信息。SHOW INDEX FROM SCHEMANAME.TABLENAME 的结果取之此表；

关键 SQL：

• SELECT * FROM SCHEMATA WHERE SCHEMA_NAME='XXX'；
• SELECT * FROM TABLES WHERE TABLE_SCHEMA='XXX' AND TABLE_NAME='XX'；
• SELECT * FROM COLUMNS WHERE TABLE_SCHEMA='XXX' AND TABLE_NAME='XX'；
• SELECT * FROM STATISTICS WHERE TABLE_SCHEMA='XXX' AND TABLE_NAME='XX'；

五、代码实现

这里只对 sync（同步）做简单介绍：

@Slf4j
public class App {
  /**
   * java -jar xxxx.jar src dst action
   *
   * src: host:port:username:passwd
   *
   * dst: host:port:username:passwd
   *
   * action: sync(同步)|diff(比对)|copy(复制)
   *
   * eg. java -jar day09-1.0.0.jar 127.0.0.1:3366:root:123456 127.0.0.1:3377:root:123456 sync
   *
   * @param args src dst action
   */
  public static void main(String[] args) {
    log.info("db schema sync start, args={}", Arrays.toString(args));
    start(args);
  }

  public static void start(String[] args){

    // 1.校验,参数个数,类型,格式不对校验
    calibration(args);

    // 2.解析,将args 解析成 SyncActionDTO
    SchemaActionDTO actionDTO = parse(args);
    // System.out.println(actionDTO);

    // 3.执行同步/比对/复制
    SchemaHander.doAction(actionDTO);
  }
}

入口类包含三个步骤：校验参数、解析参数、执行操作（doAction）

public class SchemaHander {

  public static void  doAction(SchemaActionDTO actionDTO){
    ConnectDTO src = actionDTO.getSrc();
    ConnectDTO dst = actionDTO.getDst();
    Action action = actionDTO.getAction();

    if (Action.SYNC.equals(action)){
      SyncHander.doSync(src,dst);
    }else if (Action.DIFF.equals(action)){
      DiffHander.doDiff(src,dst);
    }else if (Action.COPY.equals(action)){
      CopyHander.doCopy(src,dst);
    }else {
      throw new IllegalStateException("do not supprt this action");
    }
  }
}

根据接收到的指令的第三个参数从而做对应的操作（diff、sync、copy）

public class SyncHander {

  /**
   * 分析src和dst两个数据库实例
   * @param src
   * @param dst
   */
  public static void doSync(ConnectDTO src,ConnectDTO dst){

    // 1.解析src和dst中的db差异,相同的数据库名和不同的数据库名
    Pair<Set<String>, Set<String>> dbPair = parseDb(src, dst);
    System.out.println("dbPair = " + dbPair);

    // 2.src有,dst无
    DbHander.copyDb(src, dst, dbPair.getLeft());

    // 3.src有,dst有
    DbHander.diffDb(src, dst, dbPair.getRight());

  }
}

解析源数据库和目标数据库的差异，相同的数据库和不同的数据库（不同的指的是src中有二dst中没有）。

src 中有而 dst 中没有的数据库，直接在 dst 中创建数据库、表和索引。

src 中和 dst 中都有的数据库，则进一步分析该数据库中的表的情况。

public class DbHander {
  /**
   * 分析db，src有，dst有
   * @param src
   * @param dst
   * @param target
   */
  public static void diffDb(ConnectDTO src, ConnectDTO dst, Set<String> target){

    for (String db : target) {

      // 解析src和dst中的同名的数据库的差异，返回该数据库中表的差异，相同的表名和不同的表名
      Pair<Set<String>, Set<String>> tablePair = parseTable(src, dst, db);

      // 复制差异表
      TableHandler.copyTable(src, dst, db, tablePair.getLeft());

      // 对比相同表
      TableHandler.diffTable(src, dst, db, tablePair.getRight());
      }
  }
}

套路和分析数据库一样

src 中有而 dst 中没有的表，直接在 dst 中创建。

src 中和 dst 中都有的表，则进一步分析该表的所有字段和字段属性。

public class TableHandler {
/**
   * 分析相同表的字段和索引
   * @param src 源
   * @param dst 目标
   * @param db  数据库
   * @param targetTables 分析的目标表
   */
  public static void diffTable(ConnectDTO src,
      ConnectDTO dst,
      String db,
      Set<String> targetTables) {

    for (String table : targetTables) {

      // 1.分析差异字段
      Pair<Set<String>, Set<String>> columnPair = parseColumn(src, dst, db, table);

      // 2.复制src有,dst无
      ColumnHandler.copyColumn(src, dst, db, table, columnPair.getLeft());

      // 3.分析src有,dst有
      ColumnHandler.diffColumn(src, dst, db, table, columnPair.getRight());

      // 1.分析差异索引
      Pair<Set<String>, Set<String>> indexPair = parseIndex(src, dst, db, table);

      // 2.复制src有,dst无
      IndexHander.copyIndex(src, dst, db, table, indexPair.getLeft());

      // 3.分析src有,dst有
      IndexHander.diffIndex(src, dst, db, table, indexPair.getRight());
    }
  }
}

src 中有而 dst 中没有的字段和索引，直接在 dst 中创建。

src 中和 dst 中都有的字段和索引，则进一步分析。

需要注意的是索引，由于索引分为普通索引、唯一索引、主键索引和组合索引几种类型，所以在生成修改 SQL 时会比较复杂。

public class ColumnHandler {
长度、是否可为空、默认值、注释
   * @param src 源数据库实例
   * @param dst 目标数据库实例
   * @param db  数据库
   * @param table 表
   * @param targetColumns 分析的目标列
   */
  public static void diffColumn(ConnectDTO src,
      ConnectDTO dst,
      String db,
      String table,
      Set<String> targetColumns) {

    for (String column : targetColumns) {

      String queryColumnInfoSql = String.format(
          "select * from COLUMNS where TABLE_SCHEMA='%s' and TABLE_NAME='%s' AND COLUMN_NAME='%s'",
          db, table, column);

      // 1.取出src中的column的几个我们关注的属性,COLUMN_TYPE,COLUMN_COMMENT,IS_NULLABLE,COLUMN_DEFAULT
      Set<ColumnInfoDTO> srcColumnSet = JdbcUtils
          .read(src, ConnectConsts.INFO_SCHEMA_DB_NAME, queryColumnInfoSql)
          .stream()
          .map(entity -> ColumnInfoDTO.builder()
              .columnComment(entity.get("COLUMN_COMMENT").toString())
              .columnDefault(StringUtils.defaultString(String.valueOf(entity.get("COLUMN_DEFAULT")), ""))
              .columnType(entity.get("COLUMN_TYPE").toString())
              .isNullable(entity.get("IS_NULLABLE").toString())
              .build()).collect(Collectors.toSet());

      // 2.取出dst中的column的几个我们关注的属性,COLUMN_TYPE,COLUMN_COMMENT,IS_NULLABLE,COLUMN_DEFAULT
      Set<ColumnInfoDTO> dstColumnSet = JdbcUtils
          .read(dst, ConnectConsts.INFO_SCHEMA_DB_NAME, queryColumnInfoSql)
          .stream()
          .map(entity -> ColumnInfoDTO.builder()
              .columnComment(entity.get("COLUMN_COMMENT").toString())
              .columnDefault(StringUtils.defaultString(String.valueOf(entity.get("COLUMN_DEFAULT")), ""))
              .columnType(entity.get("COLUMN_TYPE").toString())
              .isNullable(entity.get("IS_NULLABLE").toString())
              .build()).collect(Collectors.toSet());

      // 3.逐个去对比,如果不一样,就生成修改SQL,如果一样,就什么都不做
      // 3.1 这个differenceColumn是需要去修改到dst中的
      Set<ColumnInfoDTO> differenceColumn = Sets.difference(srcColumnSet, dstColumnSet)
          .immutableCopy();

      for (ColumnInfoDTO infoDTO : differenceColumn) {
        String sql = String.format("alter table %s modify column %s %s %s %s comment '%s'",
            table,
            column,
            infoDTO.getColumnType(),
            isNullableSet(infoDTO.getIsNullable()),
            isDefaultSet(infoDTO.getColumnDefault()),
            infoDTO.getColumnComment()
        );
        JdbcUtils.write(dst, db, sql);
      }
    }
  }
}

public class IndexHander {
/**
   * 分析相同表相同索引的属性，并修改dst中索引的属性
   * 属性包括索引类型，是否唯一，单索引还是组合索引
   * @param src 源
   * @param dst 目标
   * @param db  db
   * @param table 表
   * @param targeIndexs 分析的目标索引
   */
  public static void diffIndex(ConnectDTO src,
    ConnectDTO dst,
    String db,
    String table,
    Set<String> targeIndexs){

    for (String index : targeIndexs) {
      String queryIndexInfoSql = String.format(
              "select * from STATISTICS where TABLE_SCHEMA='%s' and TABLE_NAME='%s' and INDEX_NAME='%s'",
              db, table, index);

      // 查出该index信息的返回结果,如果是组合索引，一个索引名对应多条记录
      List<Map<String, Object>> entities = JdbcUtils
              .read(src, ConnectConsts.INFO_SCHEMA_DB_NAME, queryIndexInfoSql);

      // 1.取出src中的index的几个我们关注的属性,COLUMN_NAME,NON_UNIQUE,SEQ_IN_INDEX
      Set<IndexInfoDTO> srcIndexSet = JdbcUtils
              .read(src, ConnectConsts.INFO_SCHEMA_DB_NAME, queryIndexInfoSql)
              .stream()
              .map(entity-> IndexInfoDTO.builder()
                  .columnName(entity.get("COLUMN_NAME").toString())
                  .nonUnique(entity.get("NON_UNIQUE").toString())
                  .seqInIndex(entity.get("SEQ_IN_INDEX").toString())
              .build()).collect(Collectors.toSet());

      // 2.取出dst中的index的几个我们关注的属性,COLUMN_NAME,NON_UNIQUE,SEQ_IN_INDEX
      Set<IndexInfoDTO> dstIndexSet = JdbcUtils
              .read(dst, ConnectConsts.INFO_SCHEMA_DB_NAME, queryIndexInfoSql)
              .stream()
              .map(entity-> IndexInfoDTO.builder()
                      .columnName(entity.get("COLUMN_NAME").toString())
                      .nonUnique(entity.get("NON_UNIQUE").toString())
                      .seqInIndex(entity.get("SEQ_IN_INDEX").toString())
                      .build()).collect(Collectors.toSet());

      // 对比，找出名称一样，但是属性不一样的索引。组合索引的比对有问题
      Set<IndexInfoDTO> differenctIndex = Sets.difference(srcIndexSet, dstIndexSet).immutableCopy();

      System.out.println("differenctIndex.size() = " + differenctIndex.size());

      for (IndexInfoDTO infoDTO : differenctIndex) {
        System.out.println("infoDTO = " + infoDTO);
      }

      String sql = null;
      // 单列索引
      if (differenctIndex.size() == 1) {

        // 先删除dst中的索引
        deleteIndex(dst,db,table,index);

        // 再在dst中创建索引
        for (IndexInfoDTO indexDTO : differenctIndex) {
          if ("PRIMARY".equals(index)) {
            sql = String.format("ALTER TABLE %s ADD PRIMARY KEY(%s);", table, indexDTO.getColumnName());
          }else {
            sql = String.format("ALTER TABLE %s ADD %s %s(%s)",
                    table,
                    isNonUnique(indexDTO.getNonUnique()),
                    index,
                    indexDTO.getColumnName());
          }
          JdbcUtils.write(dst,db,sql);
          }
      // 组合索引
      }else if (differenctIndex.size() > 1) {
        // 先删除dst中的索引
        deleteIndex(dst,db,table,index);

        // 再在dst中创建索引
        String[] arrs = getPair(entities).getLeft();
        String nonUnique = getPair(entities).getRight();

        if ("PRIMARY".equals(index)){
          String baseSql = "alter table %s add primary key(";
          String formatSql = formatSql(arrs, baseSql);
          sql = String.format(formatSql, table);
        }else {
          String baseSql = "alter table %s add %s %s(";
          String formatSql = formatSql(arrs, baseSql);
          sql = String.format(formatSql, table, isNonUnique(nonUnique), index);
        }
        JdbcUtils.write(dst,db,sql);
      }
    }
  }
}

这里是分析字段和索引的过程。

以上所有代码，复制数据库、表、字段、索引的代码都没有贴出来，大家可以自己来实现。

另外，最后索引的分析有一个 bug，希望大家可以发现。

六、问题

上面，我们基本实现了这个工具的框架，但是还存在一些问题：

Connection

• 使用连接池,并且基于连接信息做了一个Map<ConnectDTO,DruidDatasource>

SQL 执行

• 使用批量执行SQL；

多线程执行

• 任务分割去从线程池中申请线程,然后去执行；