DTS按业务场景批量迁移阿里云MySQL表实战（下）：迁移管理平台设计与实现

本文是 DTS按业务场景批量迁移阿里云MySQL表实战（上）：技术选型和API对接  的后续，应用状态模式，完成业务系统中的迁移模块。DTS的对接方式可参考前文。
迁移管理平台设计与实现

完成DTS API对接后，就需要考虑如何将DTS和业务系统有机结合实现整套的迁移流程。
出于信息安全角度的考虑，本文删除了大量涉及实际业务的实现代码。
业务约束

从业务出发，最好的体验肯定是用户无感的，即迁移完成后，确认新旧表数据一致，直接切换到新表查询。
显然过于理想化了，如果迁移期间，用户对旧表进行了写入，新表可能会少数据，不能贸然切换，要做数据的对比。如果用户一直在写入，就要一直反复的对比、确认，有增量数据就要删除新表重新迁移，流程复杂。
和业务方沟通，得知对方可以禁止写入正在迁移的公司的表，等迁移完成后再恢复使用。这样流程就简单多了，开始迁移时，将旧表重命名增加特殊的后缀，就能防止用户操作，并确保旧表数据不发生变更。
技术校验

如何判断一个公司是否迁移成功？最严谨的方式是逐表逐行数据对比，但是在使用DTS的情况下并无必要。我采取的比较策略是，在迁移前后：

• 源表目标表数量相同
  
• 对应表数据量相同、数据最后更新时间（如有此列）相同
  

只要满足以上要求，就认为数据是一致的。可以通过SELECT COUNT(*), MAX(updateTime) FROM table_name一次性获取。
迁移状态机

经过分析和简化，迁移状态机如下：

可以发现，每个状态都可以进行“推进”和“回滚”两个动作，很适合使用状态模式来实现。状态模式的实现先放一放，看看几个基本数据结构：
迁移任务

1. @Data
2. public class TableMigrateTask {
3.   // 主键
4.   private Long id;
5.   // 公司id
6.   private Long companyId;
7.   // 原始分库位
8.   private Long oldSchemaId;
9.   // 新分库位
10.   private Long newSchemaId;
11.   // 任务状态
12.   private Integer state;
13.   // DTS任务状态，和阿里云定义一致
14.   private String dtsStatus;
15.   // DTS任务已删除（释放）
16.   private Boolean dtsDeleted;
17.   // DTS实例id
18.   private String dtsInstanceId;
19.   // DTS任务id
20.   private String dtsJobId;
21.   // 失败原因
22.   private String failedReason;
23.   // 状态跳转时的信息，辅助排查问题
24.   private String transitInfo;
25.   // 迁移前表数据统计，json格式，表名、数据量、最后更新时间
26.   private String tableStatisticsSrc;
27.   // 迁移表结果统计，json格式kv结构，表名-数据量
28.   private String tableStatisticsDest;
29. }

复制代码

迁移上下文

迁移状态机实际处理的对象，封装了一些服务，可以视为领域对象（充血模型）。

1. @Setter
2. public class TableMigrateExecuteContext {
4.   // 持有的任务对象
5.   @Getter
6.   private TableMigrateTask tableMigrateTask;
8.   // 当前的状态
9.   @Getter
10.   private TableMigrateState currentState;
12.   private TableMigrateTaskRepository tableMigrateTaskRepository;
14.   private TableMigrateQueryService tableMigrateQueryService;
16.   private TableMigrateService tableMigrateService;
18.   private TableArchiveService tableArchiveService;
20.   private DataSourceHolder dataSourceHolder;
22.   public void createInstanceAndConfigureDtsJob() {
23.       // 调用DTS API创建任务
24.   }
26.   public DescribeDtsJobDetailResponse queryDtsMigJob() {
27.       // 调用DTS API查询
28.   }
30.   public void switchRoute(long oldSchemaId, long newSchemaId) {
31.       // 将分表以外的单表update为newSchemaId
32.   }
34.   public void stopDtsMigJob() {
35.       // 调用DTS API停止
36.   }
38.   public void updateTableMigrateTask(TableMigrateTask modifiedTask) {
39.        // 更新持有的任务(持久化)
40.   }
42.   /**
43.    * 状态推进
44.    *
45.    * @return 返回信息，不成功时非空
46.    */
47.   public String forward() {
48.     return currentState.forward(this);
49.   }
51.   /** 状态回滚 */
52.   public String rollback() {
53.     return currentState.rollback(this);
54.   }
56.   /**
57.    * 重命名旧表
58.    *
59.    * @param forward true-迁移场景，加——migold，反之则不加
60.    * @param ignoreExited 是否忽略已存在的表，仅在初始态的回滚场景可用
61.    */
62.   public void renameOldTableNames(boolean forward, boolean ignoreExited) {
63.       // 注意要考虑源库中是否存在和旧表相同的同名表
64.   }
67.   public void updateDestTableInfo() {
68.      // 更新目标表的统计信息
69.   }
71.   public void archiveNewTables(List<String> needArchiveTables) {
72.       // 归档新表  
73.   }
75.   /**
76.    * 删除表
77.    *
78.    * @param newTable true-新表，false-旧表
79.    */
80.   public void deleteTables(boolean newTable) {
81.       // 批量执行DROP TABLE
82.   }
84.   public void deleteDtsInstanceAndJob() {
85.       // 调用DTS API释放实例
86.   }
87. }

复制代码

工厂类

1. @Component
2. public class TableMigrateContextFactory {
4.   @Resource private TableMigrateTaskRepository tableMigrateTaskRepository;
5.   @Resource private TableMigrateQueryService tableMigrateQueryService;
6.   @Resource private TableMigrateService tableMigrateService;
7.   @Resource private TableArchiveService tableArchiveService;
8.   @Resource private DataSourceHolder dataSourceHolder;
10.   public TableMigrateExecuteContext buildContext(long taskId) {
11.     TableMigrateTask task = tableMigrateTaskRepository.getById(taskId);
12.     if (task == null || task.getStatus() == 0) {
13.       throw new BizException("表迁移任务不存在或已被删除");
14.     }
15.     TableMigrateExecuteContext context = new TableMigrateExecuteContext();
16.     context.setTableMigrateTask(task);
17.     context.setCurrentState(buildState(TableMigrateStateEnum.getByValue(task.getState())));
19.     // 服务注入
20.     context.setTableMigrateTaskRepository(tableMigrateTaskRepository);
21.     context.setTableMigrateQueryService(tableMigrateQueryService);
22.     context.setTableMigrateService(tableMigrateService);
23.     context.setTableArchiveService(tableArchiveService);
24.     context.setDataSourceHolder(dataSourceHolder);
25.     return context;
26.   }
28.   private TableMigrateState buildState(TableMigrateStateEnum stateEnum) {
29.     switch (stateEnum) {
30.       case INIT:
31.         return new MigrateInitState();
32.       case FAILED:
33.         return new MigrateFailedState();
34.       case PROCESSING:
35.         return new MigrateProcessingState();
36.       case NEED_SWITCH:
37.         return new MigrateNeedSwitchState();
38.       case SWITCHED:
39.         return new MigrateSwitchedState();
40.       case FINISH:
41.         return new MigrateFinishState();
42.       default:
43.         throw new BizException("迁移状态非法");
44.     }
45.   }
46. }

复制代码

迁移状态

我在做本次的系统设计时，对状态模式做了一些回顾和参考。迁移状态是状态模式的核心，从设计模式的角度来看，状态模式“允许对象在其内部状态改变时动态调整自身行为，使得对象的表现形态如同修改了其所属类。”
以下是各个类的继承关系：

对应的状态如下：
类名含义说明TableMigrateState接口定义AbstractMigrateState状态抽象类AbstractFinalState终态抽象类终态很多操作都是不支持的，和AbstractMigrateState分开更简洁MigrateInitState初始记录要迁移的统计数据和配置MigrateProcessingState迁移中DTS进行迁移动作的状态MigrateNeedSwitchState迁移完成待切换分库位数据已同步在新表，但还不可以通过业务功能直接访问MigrateSwitchedState分库位已切换待删除旧表数据已同步在新表，且能通过业务功能直接访问MigrateFinishState迁移完成数据已同步在新表且能访问，旧表已删除MigrateFailedState迁移失败回滚，旧表恢复访问，新表如果有则删除状态接口

1. public interface TableMigrateState {
3.   /**
4.    * 前进到下一状态
5.    *
6.    * @param context
7.    * @return 失败的提示信息
8.    */
9.   String forward(TableMigrateExecuteContext context);
11.   /**
12.    * 回滚操作
13.    *
14.    * @param context
15.    * @return 失败的提示信息
16.    */
17.   String rollback(TableMigrateExecuteContext context);
19.   /**
20.    * 获取当前的状态对应枚举
21.    */
22.   TableMigrateStateEnum getState();
24.   /**
25.    * 获取下一个状态
26.    */
27.   TableMigrateState getNextState();
29.   /**
30.    * 获取回滚的状态
31.    */
32.   TableMigrateState getRollbackState();
33. }

复制代码

状态抽象类

1. public abstract class AbstractMigrateState implements TableMigrateState {
3.   @Override
4.   public String forward(TableMigrateExecuteContext context) {
5.     TableMigrateTask task = context.getTableMigrateTask();
7.     // 1. 前置校验
8.     // 根据校验结果，判断是留在当前状态，还是直接回滚到迁移失败状态
10.     // 2. 实际动作，由实现类完成
11.    // 简单起见，在实际动作里的异常都自动回滚
13.     // 3. 状态跳转
14.     transit(context, getNextState());
15.     return null;
16.   }
18.   @Override
19.   public String rollback(TableMigrateExecuteContext context) {
20.     TableMigrateTask task = context.getTableMigrateTask();
21.     // 1. 当前状态校验
22.     checkCurrentState(context);
24.     // 2. 回滚操作，如果发生异常，保持在当前状态
26.     // 3. 状态跳转
27.     transit(context, getRollbackState());
28.     return null;
29.   }
31.   /**
32.    * 前置校验
33.    *
34.    * @param context
35.    */
36.   protected PreCheckResult preCheck(TableMigrateExecuteContext context) {
37.     return checkCurrentState(context);
38.   }
40.   protected PreCheckResult checkCurrentState(TableMigrateExecuteContext context) {
41.       // 检查当前状态是否符合预期，构造检查结果
42.   }
44.   /**
45.    * 改变当前执行上下文状态, 不做其他的业务操作
46.    *
47.    * @param context
48.    * @param nextState
49.    */
50.   private void transit(TableMigrateExecuteContext context, TableMigrateState nextState) {
51.     TableMigrateTask task = context.getTableMigrateTask();
52.     task.setState(nextState.getState().getValue());
53.     context.updateTableMigrateTask(task);
54.     context.setCurrentState(nextState);
55.   }
56.   
57.   protected abstract void doForward(TableMigrateExecuteContext context);
59.   /**
60.    * 回滚操作
61.    *
62.    * <p>需要保证幂等，如果单个回滚操作失败，可以重复执行
63.    *
64.    * @param context
65.    */
66.   protected abstract void doRollback(TableMigrateExecuteContext context);
68.   /**
69.    * 旧表更名
70.    *
71.    * @param context
72.    * @param forward true-迁移场景，旧表加后缀; false-回滚场景，旧表删除后缀
73.    */
74.   protected void renameOldTableNames(
75.       TableMigrateExecuteContext context, boolean forward, boolean ignoreExited) {
76.     context.renameOldTableNames(forward, ignoreExited);
77.   }
79.   /**
80.    * 删除新表
81.    *
82.    * <p>新表的删除，最好不要共用这个方法
83.    *
84.    * @param context
85.    */
86.   protected void deleteNewTables(TableMigrateExecuteContext context) {
87.     context.deleteTables(true);
88.   }
90.   /** 前置校验结果 */
91.   @Data
92.   @AllArgsConstructor
93.   public static class PreCheckResult {
94.     /** 中断，需要回滚 */
95.     public static final int ABORT = -1;
97.     /** 校验通过 */
98.     public static final int PASS = 0;
100.     /** 校验不通过，保持原有状态 */
101.     public static final int NOT_PASS = 1;
103.     private int code;
105.     private String msg;
107.     public static PreCheckResult buildPass() {
108.       return new PreCheckResult(PASS, null);
109.     }
111.     public boolean isPass() {
112.       return this.code == PASS;
113.     }
114.   }
115. }

复制代码

终态抽象类

1. public abstract class AbstractFinalState implements TableMigrateState {
2.   @Override
3.   public String forward(TableMigrateExecuteContext context) {
4.     return "当前状态【" + getState().getValue() + " " + getState().getDes() + "】已是终态，不能进行下一步操作";
5.   }
7.   @Override
8.   public String rollback(TableMigrateExecuteContext context) {
9.     return "当前状态【" + getState().getValue() + " " + getState().getDes() + "】已是终态，不能进行撤销操作";
10.   }
12.   @Override
13.   public TableMigrateState getNextState() {
14.     throw new BizException("当前状态" + getState().getValue() + "已是终态，没有后续状态可跳转");
15.   }
17.   @Override
18.   public TableMigrateState getRollbackState() {
19.     throw new BizException("当前状态" + getState().getValue() + "已是终态，没有后续撤销态可跳转");
20.   }
21. }

复制代码

初始

1. public class MigrateInitState extends AbstractMigrateState {
2.   @Override
3.   protected void doForward(TableMigrateExecuteContext context) {
4.     TableMigrateTask task = context.getTableMigrateTask();
5.     if (task.getOldSchemaId().equals(task.getNewSchemaId())) {
6.       throw new BizException("迁移前后的分库位id相同");
7.     }
9.     // 1. 旧表更名, 直接阻止后续的变更
10.     // 归档不影响RENAME
11.     renameOldTableNames(context, true, false);
13.     // 2. 创建DTS任务并回写到task字段
14.     // 创建失败则直接抛异常，回滚
15.     // 此处DTS任务是直接提交执行的，并不能确定当前实际是哪个状态，因此状态留空
16.     context.createInstanceAndConfigureDtsJob();
17.   }
18.   
19.   @Override
20.   protected void doRollback(TableMigrateExecuteContext context) {
21.     // 初始态回滚时，旧表可能还没有更名
22.     renameOldTableNames(context, false, true);
23.   }
25.   @Override
26.   public TableMigrateStateEnum getState() {
27.     return TableMigrateStateEnum.INIT;
28.   }
30.   @Override
31.   public TableMigrateState getNextState() {
32.     return new MigrateProcessingState();
33.   }
35.   @Override
36.   public TableMigrateState getRollbackState() {
37.     return new MigrateFailedState();
38.   }
39. }

复制代码

迁移中

1. public class MigrateProcessingState extends AbstractMigrateState {
3.   /** DTS-未初始化的状态 */
4.   private static final Set<String> DTS_NOT_INIT =
5.       Sets.newHashSet(
6.           DtsJobStatusEnum.NOT_STARTED.getCode(), DtsJobStatusEnum.NOT_CONFIGURED.getCode());
8.   /** DTS-处理中的状态 */
9.   private static final Set<String> DTS_PROCESSING =
10.       Sets.newHashSet(
11.           DtsJobStatusEnum.PRECHECKING.getCode(),
12.           DtsJobStatusEnum.PRECHECK_PASS.getCode(),
13.           DtsJobStatusEnum.INITIALIZING.getCode(),
14.           DtsJobStatusEnum.SYNCHRONIZING.getCode(),
15.           DtsJobStatusEnum.MIGRATING.getCode(),
16.           DtsJobStatusEnum.SUSPENDING.getCode(),
17.           DtsJobStatusEnum.MODIFYING.getCode(),
18.           DtsJobStatusEnum.RETRYING.getCode(),
19.           DtsJobStatusEnum.UPGRADING.getCode(),
20.           DtsJobStatusEnum.DOWNGRADING.getCode(),
21.           DtsJobStatusEnum.LOCKED.getCode());
23.   /** DTS-失败的状态 */
24.   private static final Set<String> DTS_FAILED =
25.       Sets.newHashSet(
26.           DtsJobStatusEnum.PRECHECK_FAILED.getCode(),
27.           DtsJobStatusEnum.INITIALIZE_FAILED.getCode(),
28.           DtsJobStatusEnum.FAILED.getCode(),
29.           DtsJobStatusEnum.MIGRATION_FAILED.getCode());
31.   @Override
32.   public PreCheckResult preCheck(TableMigrateExecuteContext context) {
33.     checkCurrentState(context);
34.     // 校验DTS任务已完成
35.     TableMigrateTask task = context.getTableMigrateTask();
36.     DescribeDtsJobDetailResponse dtsJobDetailResponse = context.queryDtsMigJob();
37.     if (dtsJobDetailResponse == null || dtsJobDetailResponse.getBody() == null) {
38.       return new PreCheckResult(
39.           PreCheckResult.NOT_PASS,
40.           String.format(
41.               "DTS任务结果查询失败, 返回结果为空。instanceId=%s, jobId=%s",
42.               task.getDtsInstanceId(), task.getDtsJobId()));
43.     }
44.     DescribeDtsJobDetailResponseBody responseBody = dtsJobDetailResponse.getBody();
45.     if (BooleanUtils.isNotTrue(responseBody.getSuccess())) {
46.       return new PreCheckResult(
47.           PreCheckResult.NOT_PASS,
48.           String.format(
49.               "DTS任务结果查询失败, 接口调用结果为失败。instanceId=%s, jobId=%s",
50.               task.getDtsInstanceId(), task.getDtsJobId()));
51.     }
52.     DtsJobStatusEnum dtsJobStatusEnum = DtsJobStatusEnum.getByCode(responseBody.getStatus());
53.     if (dtsJobStatusEnum == null) {
54.       return new PreCheckResult(
55.           PreCheckResult.NOT_PASS,
56.           String.format(
57.               "DTS任务状态非法，请稍后重试。instanceId=%s, jobId=%s, 阿里云状态=%s",
58.               task.getDtsInstanceId(), task.getDtsJobId(), responseBody.getStatus()));
59.     }
60.     task.setDtsStatus(dtsJobStatusEnum.getCode());
61.     if (DTS_NOT_INIT.contains(dtsJobStatusEnum.getCode())) {
62.       return new PreCheckResult(
63.           PreCheckResult.NOT_PASS,
64.           String.format(
65.               "DTS任务状态异常，尚未初始化。instanceId=%s, jobId=%s, 阿里云状态=%s",
66.               task.getDtsInstanceId(), task.getDtsJobId(), responseBody.getStatus()));
67.     }
68.     if (DTS_PROCESSING.contains(dtsJobStatusEnum.getCode())) {
69.       return new PreCheckResult(
70.           PreCheckResult.NOT_PASS,
71.           String.format(
72.               "DTS任务仍在处理中。instanceId=%s, jobId=%s, 阿里云状态=%s",
73.               task.getDtsInstanceId(), task.getDtsJobId(), responseBody.getStatus()));
74.     }
75.     if (DTS_FAILED.contains(dtsJobStatusEnum.getCode())) {
76.       return new PreCheckResult(
77.           PreCheckResult.ABORT,
78.           String.format(
79.               "DTS任务执行失败。instanceId=%s, jobId=%s, 阿里云状态=%s",
80.               task.getDtsInstanceId(), task.getDtsJobId(), responseBody.getStatus()));
81.     }
82.     if (StringUtils.equals(DtsJobStatusEnum.FINISHED.getCode(), dtsJobStatusEnum.getCode())) {
83.       return PreCheckResult.buildPass();
84.     }
85.     return new PreCheckResult(
86.         PreCheckResult.NOT_PASS,
87.         String.format(
88.             "DTS任务状态非法。instanceId=%s, jobId=%s, 阿里云状态=%s",
89.             task.getDtsInstanceId(), task.getDtsJobId(), responseBody.getStatus()));
90.   }
92.   @Override
93.   protected void doForward(TableMigrateExecuteContext context) {
94.     // 已在校验时更新任务状态，并根据DTS执行状态判断要不要回滚
96.     // 1. 将迁移后的表，按照原表的状态进行归档
98.     // 2. 写入迁移后的表统计信息
99.     context.updateDestTableInfo();
100.   }
102.   @Override
103.   protected void doRollback(TableMigrateExecuteContext context) {
104.     // 1. 尝试中止DTS任务，对已完成的DTS任务调用不会抛异常
106.     // 2. 新表删除
107.     deleteNewTables(context);
109.     // 3. 旧表名称还原
110.     renameOldTableNames(context, false, false);
111.   }
113.   @Override
114.   public TableMigrateStateEnum getState() {
115.     return TableMigrateStateEnum.PROCESSING;
116.   }
118.   @Override
119.   public TableMigrateState getNextState() {
120.     return new MigrateNeedSwitchState();
121.   }
123.   @Override
124.   public TableMigrateState getRollbackState() {
125.     return new MigrateFailedState();
126.   }
127. }

复制代码

迁移完成待切换分库位

1. public class MigrateNeedSwitchState extends AbstractMigrateState {
3.   @Override
4.   protected void doForward(TableMigrateExecuteContext context) {
5.     // 单表路由切换
6.     context.switchRoute(
7.         context.getTableMigrateTask().getOldSchemaId(),
8.         context.getTableMigrateTask().getNewSchemaId());
9.   }
11.   @Override
12.   protected void doRollback(TableMigrateExecuteContext context) {
13.     // 1. 新表删除
14.     deleteNewTables(context);
16.     // 2. 旧表名称还原
17.     renameOldTableNames(context, false, false);
19.     // 3. 单表和路由恢复
20.     // 可能已经在forward时做过，因此也做复原
21.     context.switchRoute(
22.         context.getTableMigrateTask().getNewSchemaId(),
23.         context.getTableMigrateTask().getOldSchemaId());
24.   }
26.   @Override
27.   public TableMigrateStateEnum getState() {
28.     return TableMigrateStateEnum.NEED_SWITCH;
29.   }
31.   @Override
32.   public TableMigrateState getNextState() {
33.     return new MigrateSwitchedState();
34.   }
36.   @Override
37.   public TableMigrateState getRollbackState() {
38.     return new MigrateFailedState();
39.   }
40. }

复制代码

分库位已切换待删除旧表

1. public class MigrateSwitchedState extends AbstractMigrateState {
3.   @Override
4.   protected void doForward(TableMigrateExecuteContext context) {
5.     // 删除旧表
6.     context.deleteTables(false);
7.   }
9.   @Override
10.   protected void doRollback(TableMigrateExecuteContext context) {
11.     // 1. 新表删除
12.     deleteNewTables(context);
14.     // 2. 旧表名称还原
15.     renameOldTableNames(context, false, false);
17.     // 3. 单表和路由恢复
18.     context.switchRoute(
19.         context.getTableMigrateTask().getNewSchemaId(),
20.         context.getTableMigrateTask().getOldSchemaId());
21.   }
23.   @Override
24.   public TableMigrateStateEnum getState() {
25.     return TableMigrateStateEnum.SWITCHED;
26.   }
28.   @Override
29.   public TableMigrateState getNextState() {
30.     return new MigrateFinishState();
31.   }
33.   @Override
34.   public TableMigrateState getRollbackState() {
35.     return new MigrateFailedState();
36.   }
37. }

复制代码

迁移完成

1. public class MigrateFinishState extends AbstractFinalState {
2.   @Override
3.   public TableMigrateStateEnum getState() {
4.     return TableMigrateStateEnum.FINISH;
5.   }
6. }

复制代码

迁移失败

1. public class MigrateFailedState extends AbstractFinalState {
2.   @Override
3.   public TableMigrateStateEnum getState() {
4.     return TableMigrateStateEnum.FAILED;
5.   }
6. }

复制代码

操作流程

• 创建迁移任务TableMigrateTask
  
• 通过工厂类，使用任务id创建包含迁移任务TableMigrateTask的迁移上下文TableMigrateExecuteContext
  
• 调用TableMigrateExecuteContext的forward()推进状态、rollback()回滚状态
  

查询相关功能实现

DTS任务列表查询

因为已经把DTS的相关字段持久化了，可以通过业务系统相关的迁移任务表实现分页查询。
不过在“迁移中”跳转到“迁移完成待切换分库位”的过程中，DTS也会经历多个状态，典型的有Prechecking、Migrating、Finished等（见DescribeDtsJobDetail_数据传输_API文档的Status字段说明），可以通过接口获取最新的状态并写入迁移任务表。
如何查询待迁移的表？

回顾一下，要迁移的表分以下三种形式：

• 后缀是公司id，如table_companyId
  
• 后缀是 公司_年份，如table_company_year
  
• 后缀是业务id，如table_bizId
  

对于同一个前缀，以companyId=123为例，第一、三种表都可以精确匹配：

• 第一种表每个公司只有一张，比如table_a_123、table_b_123；
  
• 第三种表每个bizId同一个公司只有一张，比如bizId可以取1、2，那么会存在table_c_1、table_c_2、table_d_1、table_d_2，并且，bizId是有限的，可以从一张bizId_table表获取所有可选值。
  
• 对于第二种表year的取值范围，虽然可以类似bizId一样去找，但是并没有直接的关系表。
  

我想到了两种方案，最后选择了第二种。
SHOW TABLES LIKE查询指定前缀

这是最先考虑到的方案，比较直接，而且在开发、测试环境中运行良好。但是在线上就不行了，将所有表查询一次要数分钟，调用早已超时。我想这应该和线上环境表数量过多导致元数据获取变慢有关，每次查询需要上百ms，累计耗时长达数分钟。
DatabaseMetaData一次性获取所有表后过滤

可以通过DataSource的元数据，一次性获取数据源对应库的所有的表，再将表名进行过滤。经过测试，10W级数据获取全部表的时间在3~7S之间，和方案一相比快很多。
以下代码片段展示了如何获取所有表名，忽略异常处理。

1. String physicalSchemaName = dataSourceHolder.getPhysicalSchemaName(logicalSchemaName);
2. HikariDataSource dataSource = dataSourceHolder.getDataSourceByPhysicalSchemaName(physicalSchemaName);
3. try (Connection connection = dataSource.getConnection()) {
4.     DatabaseMetaData metaData = connection.getMetaData();
5.     ResultSet catalogs = metaData.getCatalogs();
6.     while (catalogs.next()) {
7.         String tableCat = catalogs.getString("TABLE_CAT");
8.         if (!StringUtils.equals(tableCat, physicalSchemaName)) {
9.             // 判断库名是否一致。如果多个库实际上在同一个RDS实例，元数据实际上是这些库的，而非单个库的
10.             continue;
11.         }
13.         // 获取指定数据库中的所有表名
14.         ResultSet tableResultSet =
15.             metaData.getTables(physicalSchemaName, null, "%", new String[] {"TABLE"});
16.         int count = 0;
17.         while (tableResultSet.next()) {
18.           count++;
19.           String physicalTableName = tableResultSet.getString("TABLE_NAME");
20.           // 确定是否是要查的表，判断逻辑此处省略【注1】
21.          }
22.         LoggerUtils.info("本次共查询了" + count + "个表的元数据");
23.       }
24.       LoggerUtils.info(
25.           "获取" + physicalSchemaName + "库的所有表元数据总耗时：" + (System.currentTimeMillis() - t1) + "ms");
26.     } catch (SQLException e) {
27.       LoggerUtils.error("获取分库元数据失败, 发生SQL异常", e);
28.       throw new BizException("获取分库元数据失败, 发生SQL异常", e);
29.     }
30. }

复制代码

对数据库中每个表判定它是否是当前公司的表，对于第一、三种表，可以将后者放到一个HashSet中，每次循环时对比；对于第二种表，字符串前缀匹配无疑要花大量的时间。
为了加速前缀匹配，可以使用经典的数据结构——前缀匹配树。前缀匹配树的代码如下：

1. public class StringPrefixTrie {
2.   private final Node root = new Node();
4.   static class Node {
5.     boolean isEnd;
6.     Map<Character, Node> children = new HashMap<>();
7.   }
9.   /**
10.    * 增加一个待匹配的模式
11.    *
12.    * @param pattern
13.    */
14.   public void addPattern(String pattern) {
15.     Node current = root;
16.     for (char c : pattern.toCharArray()) {
17.       current = current.children.computeIfAbsent(c, k -> new Node());
18.     }
19.     current.isEnd = true;
20.   }
22.   /**
23.    * 是否满足前缀匹配
24.    *
25.    * @param str
26.    * @return
27.    */
28.   public String match(String str) {
29.     Node current = root;
30.     for (int i = 0; i < str.length(); i++) {
31.       current = current.children.get(str.charAt(i));
32.       if (current == null) {
33.         // 没有匹配到任何前缀
34.         return null;
35.       }
36.       if (current.isEnd) {
37.         // 返回匹配到的任意一个前缀
38.         return str.substring(0, i + 1);
39.       }
40.     }
41.     return null;
42.   }
43. }

复制代码

对所有需要前缀匹配的表的前缀调用addPattern()，在循环中先判断是否满足前缀匹配，再判断精准匹配即可。
后续规划

• 批量迁移功能，将迁移批量化、自动化：
  
  
  
  • 批量捞取公司，判断是否需要迁移、迁移成本。理想情况下，数据量少但表多的公司，是迁移到其他库的最佳候选，大大降低源库的表量又节约了复制数据的时间
    
  • 多个公司id提交、创建任务、状态流转
    
  
  
• 自动校验迁移是否成功
  

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！