[Java知识库]【源码解读】liquibase启动逻辑

在liquibase的实际使用过程中，发现了不少过往没有注意到的问题，秉承自其所以然的思想，借机阅读了下liquibase的源码，深感代码质量和设计还是相当不错的。

1. 前言

最近部门内部终于开始了对于liquibase的试点，在实际的使用过程中发现了不少困惑点，秉承追根溯源的思想，借机对liquibase源码做了简单的了解。

2. 源码解读

liquibase官方对诸如servlet, spring等都进行集成，但不论前期多么花里胡哨，最终的调用都是落在了Liquibase.update()上，本文的解读也由此开始。

2.1 Liquibase.update()

// Liquibase.update()
public void update(Contexts contexts, LabelExpression labelExpression, boolean checkLiquibaseTables)
	throws LiquibaseException {
	// 工厂模式获取执行LockService实例, 
	LockService lockService = LockServiceFactory.getInstance().getLockService(database);
	// 尝试获取数据库锁, 以执行接下来额数据库DDL操作
	//	我们熟悉的"Waiting for changelog lock...." 正是来源于本方法
	// 这一步会将liquibase内置默认表 databasechangeloglock中的 LOCKED 字段设置为 1
	lockService.waitForLock();
	
	// 关于liquibase中的context和label特性, 本文这里不深究. 
	// 它是liquibase为了应对各类部署和执行环境所给出的解决方案, 作用和Sring中的profile, 以及Mybatis中的 enviroment 类似
	// 底部给出了官方提供的对比文档, liquibase官方文档还是挺给力的
	changeLogParameters.setContexts(contexts);
	changeLogParameters.setLabels(labelExpression);

	try {
		// 依然是借助工厂模式ChangeLogParserFactory来获取与用户配置文件匹配的ChangeLogParser实例(例如常见的XMLChangeLogSAXParser, 以及SqlChangeLogParser, YamlChangeLogParser等)
		// 借助上一步获取的ChangeLogParser实例解析用户配置的配置文件, 获得该配置文件在内存中的映射对象DatabaseChangeLog（这个类名也呼应了通常xml配置文件中的根节点 <databaseChangeLog> ）
		// 节点属性解析还得看 DatabaseChangeLog.handleChildNode(ParsedNode node, ResourceAccessor resourceAccessor)
		DatabaseChangeLog changeLog = getDatabaseChangeLog();
		
		// 是否检查liquibase内置表DatabaseChangeLog, 默认为true
		if (checkLiquibaseTables) {
			// 检查内置表DatabaseChangeLog是否存在, 以及确保表中的各个字段存在并且字段类型等都符合预期
			// 这个checkLiquibaseTables方法代码行不多, 涉及逻辑还不少, 因此我们将其下放进行专门讲解
			checkLiquibaseTables(true, changeLog, contexts, labelExpression);
		}

		ChangeLogHistoryServiceFactory.getInstance().getChangeLogService(database).generateDeploymentId();
		// 方法名说明一切; 涉及到的扩展较多, 按照惯例我们依然将对其的讲解下放
		changeLog.validate(database, contexts, labelExpression);
		
		// 这个ChangeLogIterator我们在DatabaseChangeLog.validate()方法中会再次看到
		// 这里的getStandardChangelogIterator 方法, 主要是配置 ShouldRunChangeSetFilter(这个类正是用来判断changeset是否已经执行过了)
		ChangeLogIterator changeLogIterator = getStandardChangelogIterator(contexts, labelExpression, changeLog);
		// 这个 createUpdateVisitor() 方法用来创建UpdateVisitor, 该类具体作用参见下方我们对 ChangeLogIterator 的讲解
		changeLogIterator.run(createUpdateVisitor(), new RuntimeEnvironment(database, contexts, labelExpression));
	} finally {
		database.setObjectQuotingStrategy(ObjectQuotingStrategy.LEGACY);
		try {
			lockService.releaseLock();
		} catch (LockException e) {
			LOG.severe(LogType.LOG, MSG_COULD_NOT_RELEASE_LOCK, e);
		}
		resetServices();
	}
}

// ====================================== Liquibase.checkLiquibaseTables
// 检查内置表DatabaseChangeLog是否存在, 以及确保表中的各个字段存在并且字段类型等都符合预期(不止这些)
public void checkLiquibaseTables(boolean updateExistingNullChecksums, DatabaseChangeLog databaseChangeLog,
								 Contexts contexts, LabelExpression labelExpression) throws LiquibaseException {
	ChangeLogHistoryService changeLogHistoryService =

	// 工厂模式      
	ChangeLogHistoryServiceFactory.getInstance().getChangeLogService(getDatabase());
	// StandardChangeLogHistoryService.init() 正是核心逻辑所在( 检查内置表DatabaseChangeLog是否存在, 以及确保表中的各个字段存在并且字段类型等都符合预期 )	
	changeLogHistoryService.init();
	// 是否更新内置表DatabaseChangeLog中字段MD5SUM值为null的记录 
	// ( 这里为了方便初次接触的同学理解, 所以尽量没有涉及到liquibase中的专有名词, 专业点说就是更新已执行过的changeset的MD5SUM, 这种情况主要出现在有人改了旧有changeset, 这在liquibase是不允许的, 此时的解决方案之一就是去数据库里将内置表DatabaseChangeLog中对应changeset的记录中的MD5SUM值清空, 让liquibase重新生成一个)
	if (updateExistingNullChecksums) {
		// 详细实现参见 StandardChangeLogHistoryService.upgradeChecksums 方法
		//	这里会从数据库中取出已经执行过的changeset, 对比前面读取配置文件得到的changeset集合, 将最新的MD5SUM更新回数据库(以数据库中已存在的changeset为主)
		changeLogHistoryService.upgradeChecksums(databaseChangeLog, contexts, labelExpression);
	}
	// 详细实现参见 StandardLockService.init方法
	//	确保内置表DatabaseChangeLogLock的存在, 并且完成初始化
	LockServiceFactory.getInstance().getLockService(getDatabase()).init();
}		

// ====================================== DatabaseChangeLog.validate
public void validate(Database database, Contexts contexts, LabelExpression labelExpression)
		throws LiquibaseException {

	database.setObjectQuotingStrategy(objectQuotingStrategy);
	// 这里可以看到, 默认Filter为 dbms, context, label
	// 记住这个类型, 一会我们回到主方法 Liquibase.update() 我们还会看到它的身影
	ChangeLogIterator logIterator = new ChangeLogIterator(
			this,
			new DbmsChangeSetFilter(database),
			new ContextChangeSetFilter(contexts),
			new LabelChangeSetFilter(labelExpression)
	);

	ValidatingVisitor validatingVisitor = new ValidatingVisitor(database.getRanChangeSetList());
	// 主要是检查<preConditions>标签是否满足, 这里要是不满足, 整个DatabaseChangeLog就都不要了
	validatingVisitor.validate(database, this);
	// 非常重要的一个方法, 其中
	// 记住这个方法, 一会我们回到主方法 Liquibase.update() 我们还会看到它的身影
	logIterator.run(validatingVisitor, new RuntimeEnvironment(database, contexts, labelExpression));

	for (String message : validatingVisitor.getWarnings().getMessages()) {
		LogService.getLog(getClass()).warning(LogType.LOG, message);
	}
	
	// 校验未全部通过, 抛出异常
	if (!validatingVisitor.validationPassed()) {
		throw new ValidationFailedException(validatingVisitor);
	}
}

2.2 ChangeLogIterator类

在liquibase的启动逻辑中，两次使用到了ChangeLogIterator类：

1. 在liquibase主体启动逻辑Liquibase.update()方法中。
2. 在DatabaseChangeLog.validate()方法中。

ChangeLogIterator的扩展性主要来源于其构造函数传入的ChangeSetFilter数组, 以及其run方法第一个参数ChangeSetVisitor （参见下方的代码讲解）。

// ====================================== ChangeLogIterator.run
public void run(ChangeSetVisitor visitor, RuntimeEnvironment env) throws LiquibaseException {
	Logger log = LogService.getLog(getClass());
	databaseChangeLog.setRuntimeEnvironment(env);
	try (LoggerContext ignored = LogService.pushContext("databaseChangeLog", databaseChangeLog)) {
		// 获取读取配置文件所获得的changeset记录
		List<ChangeSet> changeSetList = new ArrayList<>(databaseChangeLog.getChangeSets());
		if (visitor.getDirection().equals(ChangeSetVisitor.Direction.REVERSE)) {
			Collections.reverse(changeSetList);
		}

		// 迭代
		for (ChangeSet changeSet : changeSetList) {
			boolean shouldVisit = true;
			Set<ChangeSetFilterResult> reasonsAccepted = new HashSet<>();
			Set<ChangeSetFilterResult> reasonsDenied = new HashSet<>();
			// changeSetFilters是通过构造函数传入的
			// 执行过的changeset不再被执行的过滤, 正是在这里完成的. 具体实现类参见ShouldRunChangeSetFilter
			if (changeSetFilters != null) {
				for (ChangeSetFilter filter : changeSetFilters) {
					ChangeSetFilterResult acceptsResult = filter.accepts(changeSet);
					if (acceptsResult.isAccepted()) {
						reasonsAccepted.add(acceptsResult);
					} else {
						// 有一个ChangeSetFilter不满足, 则代表当前changeset不应该被执行
						shouldVisit = false;
						reasonsDenied.add(acceptsResult);
						break;
					}
				}
			}

			try (LoggerContext ignored2 = LogService.pushContext("changeSet", changeSet)) {
				if (shouldVisit && !alreadySaw(changeSet)) {
					// 这里的visit方法
					//	1. 对于ValidatingVisitor的实现来说, 就是判断同一个changeset, 在数据库中的CheckSum 和 配置文件中的是否一致
					//	2. 对于UpdateVisitor的实现来说, 则是调用ChangeSet.execute()方法, 实现内置表databasechangelog的填充, 记录该changeset已执行; 
					visitor.visit(changeSet, databaseChangeLog, env.getTargetDatabase(), reasonsAccepted);
					markSeen(changeSet);
				} else {
					if (visitor instanceof SkippedChangeSetVisitor) {
						((SkippedChangeSetVisitor) visitor).skipped(changeSet, databaseChangeLog, env.getTargetDatabase(), reasonsDenied);
					}
				}
			}
		}
	} finally {
		databaseChangeLog.setRuntimeEnvironment(null);
	}
}

3. ChangeSet类

对于这个ChangeSet，我们要单独解读下。

1. 首先在liquibase中，进行CheckSum的最小维度正是ChangeSet，而不是笔者之前所以为的配置文件级别，而且内置表databasechangelog中的每一条记录也正是一个执行过的ChangeSet。
2. 对于一个ChangeSet的CheckSum的生成规则，可以参考ChangeSet.generateCheckSum()的具体实现
3. 关于ChangeSet的判等标准，可以参见ChangeSet.equals()的具体实现，具体就是通过id, filePath, Author三者共同绝对的。

最后让我们来看看ChangeSet.execute()方法的实现，可以认为是灵魂所在。

/**
 * 这里故意保留了注释
 * This method will actually execute each of the changes in the list against the
 * specified database.
 *
 * @return should change set be marked as ran
 */
public ExecType execute(DatabaseChangeLog databaseChangeLog, ChangeExecListener listener, Database database)
		throws MigrationFailedException {
	if (validationFailed) {
		return ExecType.MARK_RAN;
	}

	long startTime = new Date().getTime();

	ExecType execType = null;

	boolean skipChange = false;

	Executor executor = ExecutorService.getInstance().getExecutor(database);
	try {
		// set object quoting strategy
		database.setObjectQuotingStrategy(objectQuotingStrategy);
		
		// 1. 开启事务
		if (database.supportsDDLInTransaction()) {
			database.setAutoCommit(!runInTransaction);
		}

		// 2. 日志输出打印<comment>注释
		... 

		// 3. 判断<preConditions>
		try {
			if (preconditions != null) {
				preconditions.check(database, databaseChangeLog, this, listener);
			}
		} catch (PreconditionFailedException e) {
			......
		} catch (PreconditionErrorException e) {
			......

			database.rollback();
		} finally {
			database.rollback();
		}
		
		// 4. 执行Change
		// 诸如 addColumn 正是一个 Change
		if (!skipChange) {
			for (Change change : changes) {
				try {
					change.finishInitialization();
				} catch (SetupException se) {
					throw new MigrationFailedException(this, se);
				}
			}

			log.debug(LogType.LOG, "Reading ChangeSet: " + toString());
			for (Change change : getChanges()) {
				if ((!(change instanceof DbmsTargetedChange)) || DatabaseList.definitionMatches(((DbmsTargetedChange) change).getDbms(), database, true)) {
					if (listener != null) {
						listener.willRun(change, this, changeLog, database);
					}
					if (change.generateStatementsVolatile(database)) {
						executor.comment("WARNING The following SQL may change each run and therefore is possibly incorrect and/or invalid:");
					}

					// 执行实际的数据库DDL操作
					database.executeStatements(change, databaseChangeLog, sqlVisitors);
					log.info(LogType.LOG, change.getConfirmationMessage());
					if (listener != null) {
						listener.ran(change, this, changeLog, database);
					}
				} else {
					log.debug(LogType.LOG, "Change " + change.getSerializedObjectName() + " not included for database " + database.getShortName());
				}
			}

			if (runInTransaction) {
				database.commit();
			}
			log.info(LogType.LOG, "ChangeSet " + toString(false) + " ran successfully in " + (new Date().getTime() - startTime + "ms"));
			if (execType == null) {
				execType = ExecType.EXECUTED;
			}
		} else {
			log.debug(LogType.LOG, "Skipping ChangeSet: " + toString());
		}

	} catch (Exception e) {
	   ......
	} finally {
	   ......
	}
	return execType;
}

其中涉及到两个关键：

1. 整个执行过程中，我们可以通过传入的ChangeExecListener实现来扩展自定义逻辑扩展。
2. 层级关系： 一个``

   DatabaseChangeLog    1:n
   	ChangeSet		 1:n
   		Change
   

4. Links

1. Liquibase常见问题总结
2. liquibase常见操作
3. 官方文档 - Understanding Contexts vs. Labels