数据库迁移框架Flyway介绍

官方文档

https://flywaydb.org/getstarted/firststeps/api[https://flywaydb.org/getstarted/firststeps/api]

入门示例

Java代码

package foobar;
import org.flywaydb.core.Flyway;
public class App {

public static void main(String[] args) {

Flyway flyway = new Flyway();

// 指定数据源

flyway.setDataSource("jdbc:mysql://localhost/test", "root", "root");

// 开始数据迁移

flyway.migrate();

}

}

在classpath下添加SQL文件 db/migration/V1__Create_person_table.sql

create table PERSON (
    ID int not null,
    NAME varchar(100) not null
);

运行程序，在test数据库会自动创建PERSON表。

后续新增表和字段，只需在db/migration目录下新增SQL文件，格式为V${version}__${name}.sql，version值依次增加，比如V2__name.sql，V3__name.sql。

原理介绍

Flyway的数据库迁移的实现原理是，从classpath或文件系统中找到符合规则的数据库迁移脚本，比如db/migration目录下命名规则为V${version}__${name}.sql的文件，将脚本按照version进行排序，依次执行。执行过的脚本会作为一条记录，存储在schema_version表中。当下次执行迁移时，判断脚本已经执行，则跳过。

MigrationResolver接口负责查找数据库迁移脚本，方法为resolveMigrations()，数据库迁移脚本用ResolvedMigration对象表示。MigrationResolver包含多种实现类，比如SqlMigrationResolver会从classpath下查找sql文件。查询通过Scanner类实现，Location类指定查询路径，sql文件的命名规则需要符合V${version}__${name}.sql，规则中的前缀V、后缀.sql、分隔符__均在FlywayConfiguration接口中定义。另一种实现类JdbcMigrationResolver会从classpath下查找实现JdbcMigration接口的类，类的命名规则需要符合V${version}__${name}。需要扩展自己的实现类，可以继承BaseMigrationResolver。

MetaDataTable接口负责查找已执行的数据库迁移脚本，方法为findAppliedMigrations()，已执行的数据库迁移脚本用AppliedMigration对象表示。MetaDataTable只有一种实现类MetaDataTableImpl，从数据库schema_version表查询所有记录。

ResolvedMigration集合包含了已经执行的AppliedMigration集合，在执行ResolvedMigration前，需要对比AppliedMigration，找到已执行和未执行的ResolvedMigration，对比通过MigrationInfoServiceImpl.refresh()实现。已执行的ResolvedMigration需要校验文件有没有发生变化，有变更则提示错误。未执行的ResolvedMigration依次执行，执行结果记录在schema_version表中。

Flyway的主要方法：

public class Flyway {
   /**数据库迁移*/
   public int migrate();
   /**校验已执行的迁移操作的变更情况*/
   public void validate();
   /**清理数据库*/
   public void clean();
   /**设置数据库的基准版本*/
   public void baseline();
   /**删除执行错误的迁移记录*/
   public void repair();
   /**准备执行环境，并执行Command操作，以上方法都调用了execute()来执行操作*/
   <T> T execute(Command<T> command);
}

接下来我们分析Flyway.migrate()代码执行的主逻辑。

public void migrate() {
  //由Flyway.execute()准备Command.execute()执行所需要的参数
  return execute(new Command<Integer>() {
    public Integer execute(Connection connectionMetaDataTable,
      MigrationResolver migrationResolver,  MetaDataTable metaDataTable, 
      DbSupport dbSupport, Schema[] schemas, FlywayCallback[] flywayCallbacks) {
      //为了简化代码，忽略参数传递
      doMigrate();
}

});

}
private void doMigrate() {

//校验已执行的迁移操作的变更情况

if (validateOnMigrate) {

doValidate(connectionMetaDataTable, dbSupport, migrationResolver,

metaDataTable, schemas, flywayCallbacks, true);

}
//如果尚未进行数据迁移，即schema_version表中不存在数据，

//并且数据库不为空，则插入一条baseline信息

if (!metaDataTable.exists()) {

//数据库不为空

if (!nonEmptySchemas.isEmpty()) {

//插入一条baseline信息

new DbBaseline(connectionMetaDataTable, dbSupport, metaDataTable,

schemas[0], baselineVersion, baselineDescription, flywayCallbacks).baseline();

}

}
//进行数据迁移

DbMigrate dbMigrate = new DbMigrate(connectionUserObjects, dbSupport,

metaDataTable,schemas[0], migrationResolver, ignoreFailedFutureMigration,

Flyway.this);

return dbMigrate.migrate();

}

接下来看DbMigrate.migrate()的代码片段。

MigrationInfoServiceImpl对比ResolvedMigration和AppliedMigration对象，找出需要执行数据库迁移脚本，通过pending()方法返回。最后执行数据库迁移脚本。

public int migrate() {
  int migrationSuccessCount = 0;
  while (true) {
    int count = metaDataTable.lock(new Callable<Integer>() {
  <span class="hljs-meta">@Override</span>
  <span class="hljs-function"><span class="hljs-keyword">public</span> Integer <span class="hljs-title">call</span><span class="hljs-params">()</span> </span>{
    <span class="hljs-comment">//为了简化代码，忽略参数传递</span>
    <span class="hljs-keyword">return</span> doMigrate();
  }
}
<span class="hljs-keyword">if</span> (count == <span class="hljs-number">0</span>) {
  <span class="hljs-comment">// No further migrations available</span>
  <span class="hljs-keyword">break</span>;
}
migrationSuccessCount += count;

}

return migrationSuccessCount;

}
private int doMigrate() {

//收集已经入库的数据库迁移记录，和以文件形式存在的数据库迁移脚本

MigrationInfoServiceImpl infoService = new MigrationInfoServiceImpl(

migrationResolver, metaDataTable, configuration.getTarget(),

configuration.isOutOfOrder(), true, true, true);

infoService.refresh();
//infoService.pending()记录将要执行的数据库迁移脚本

LinkedHashMap<MigrationInfoImpl, Boolean> group =

groupMigrationInfo(infoService.pending());

if (!group.isEmpty()) {

//执行数据库迁移操作

applyMigrations(group);

}

}

DbMigrate.doMigrateGroup() 执行数据库迁移脚本

private void doMigrateGroup() {
  //执行迁移脚本
  migration.getResolvedMigration().getExecutor().execute(connectionUserObjects);
//存入数据库

AppliedMigration appliedMigration = new AppliedMigration(migration.getVersion(),

migration.getDescription(), migration.getType(), migration.getScript(),

migration.getResolvedMigration().getChecksum(), executionTime, true);

metaDataTable.addAppliedMigration(appliedMigration);

}

扩展练习

在一个已经上线的游戏项目中引入Flyway框架，对于已经存在的游戏服，只执行新增的sql语句，对于新搭建的游戏服，需要创建数据库，并执行新增的sql语句。

在这个需求中，需要做的是对于新搭建的游戏服，需要创建数据库。我们可以通过FlywayCallback实现这一点，如果指定数据库为空，则执行初始化数据库的语句。

代码如下：

public static void main(String[] args) {
    final Flyway flyway = new Flyway();
    flyway.setDataSource("jdbc:mysql://localhost/test", "root", "root");
FlywayCallback flywayCallback = <span class="hljs-keyword">new</span> BaseFlywayCallback() {
    
    <span class="hljs-meta">@Override</span>
    <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">beforeMigrate</span><span class="hljs-params">(Connection connection)</span> </span>{
        DbSupport dbSupport = DbSupportFactory.createDbSupport(connection, <span class="hljs-keyword">false</span>);
        <span class="hljs-keyword">if</span>(!hasTable(dbSupport)) {
            initDb(dbSupport);
        }
    }

    <span class="hljs-function"><span class="hljs-keyword">private</span> <span class="hljs-keyword">boolean</span> <span class="hljs-title">hasTable</span><span class="hljs-params">(DbSupport dbSupport)</span> </span>{
        Schema&lt;?&gt; schema = dbSupport.getOriginalSchema();
        Table[] tables = schema.allTables();
        <span class="hljs-keyword">if</span>(tables == <span class="hljs-keyword">null</span> || tables.length == <span class="hljs-number">0</span>) {
            <span class="hljs-keyword">return</span> <span class="hljs-keyword">false</span>;
        }
        <span class="hljs-comment">//忽略表 schema_version</span>
        <span class="hljs-keyword">if</span>(tables.length == <span class="hljs-number">1</span> &amp;&amp; 
            tables[<span class="hljs-number">0</span>].getName().equalsIgnoreCase(<span class="hljs-string">"schema_version"</span>)) {
            <span class="hljs-keyword">return</span> <span class="hljs-keyword">false</span>;
        }
        <span class="hljs-keyword">return</span> <span class="hljs-keyword">true</span>;
    }
    

    <span class="hljs-function"><span class="hljs-keyword">private</span>  <span class="hljs-keyword">void</span> <span class="hljs-title">initDb</span><span class="hljs-params">(DbSupport dbSupport)</span> </span>{
        Scanner scanner = <span class="hljs-keyword">new</span> Scanner(
          Thread.currentThread().getContextClassLoader());
        Resource[] resources = scanner.scanForResources(
          <span class="hljs-keyword">new</span> Location(<span class="hljs-string">"db/init"</span>), <span class="hljs-string">""</span>, <span class="hljs-string">".sql"</span>);
        <span class="hljs-keyword">if</span>(resources == <span class="hljs-keyword">null</span> || resources.length == <span class="hljs-number">0</span>) {
            <span class="hljs-keyword">throw</span> <span class="hljs-keyword">new</span> RuntimeException(
              <span class="hljs-string">"db/init/*.sql not found in the classpath. "</span>);
        }
        <span class="hljs-keyword">for</span>(Resource resource : resources) {
          SqlMigrationExecutor executor = <span class="hljs-keyword">new</span> SqlMigrationExecutor(
              dbSupport, resource, PlaceholderReplacer.NO_PLACEHOLDERS, flyway);
          executor.execute(dbSupport.getJdbcTemplate().getConnection());
        }
    }
};

List&lt;FlywayCallback&gt; callbacks = <span class="hljs-keyword">new</span> ArrayList&lt;FlywayCallback&gt;(
  Arrays.asList(flyway.getCallbacks()));
callbacks.add(flywayCallback);
flyway.setCallbacks(callbacks.toArray(<span class="hljs-keyword">new</span> FlywayCallback[callbacks.size()]));

flyway.setBaselineOnMigrate(<span class="hljs-keyword">true</span>);
flyway.repair();
flyway.migrate();

}

      </div>