【Spring事务详解】--- 8.beforeCommit、beforeCompletion、afterCommit、afterCompletion实现分析

前言

beforeCommit、beforeCompletion、afterCommit、afterCompletion是Spring为我们在事务提交前、后提供的扩展点，业务在实际使用时可以通过这几个方法在事务提交前、后实现一些额外的逻辑控制，尤其是有些需要在事务提交后才能处理的逻辑，利用afterCommit或者afterCompletion可以非常方便的实现，本文就来简单分析一下Spring是通过什么样的方式来预埋这几个方法的。

Spring事务详解连载

【Spring事务详解】— 1.事务传播的案例演示
【Spring事务详解】— 2.事务应用的注意事项
【Spring事务详解】— 3.事务失效的八种场景
【Spring事务详解】— 4.事务管理器的架构分析
【Spring事务详解】— 5.事务管理器TransactionSynchronizationManager分析
【Spring事务详解】— 6.事务创建的流程分析
【Spring事务详解】— 7.事务提交、回滚的流程分析
【Spring事务详解】— 8.beforeCommit、beforeCompletion、afterCommit、afterCompletion实现分析

TransactionSynchronization

首先这几个方法是定义在TransactionSynchronization接口中的，并且定义的都是default方法，默认没有任何处理。

public interface TransactionSynchronization extends Ordered, Flushable {
	default void beforeCommit(boolean readOnly) {
	}
	
	default void beforeCompletion() {
	}
	
	default void afterCommit() {
	}
	
	default void afterCompletion(int status) {
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13

afterCommit方法分析

以afterCommit方法为例，我们来简单分析一下实现逻辑，其他的3个实现方法大体一致。

1.找到方法调用的入口

TransactionTemplate

如果你使用的是编程式事务，那么TransactionTemplate类的execute方法则是整个事务执行的入口，由于我们只分析afterCommit的实现逻辑，所以我们直接看相关方法即可，在exceute中，就是transactionManager.commit方法。

public <T> T execute(TransactionCallback<T> action) throws TransactionException {
	Assert.state(this.transactionManager != null, "No PlatformTransactionManager set");
	if (this.transactionManager instanceof CallbackPreferringPlatformTransactionManager) {
		return ((CallbackPreferringPlatformTransactionManager) this.transactionManager).execute(this, action);
	}
	else {
		TransactionStatus status = this.transactionManager.getTransaction(this);
		T result;
		try {
			result = action.doInTransaction(status);
		}
		catch (RuntimeException | Error ex) {
			// Transactional code threw application exception -> rollback
			rollbackOnException(status, ex);
			throw ex;
		}
		catch (Throwable ex) {
			// Transactional code threw unexpected exception -> rollback
			rollbackOnException(status, ex);
			throw new UndeclaredThrowableException(ex, "TransactionCallback threw undeclared checked exception");
		}
		// 当前面事务处理完成之后，直接调用事务提交的方法。
		this.transactionManager.commit(status);
		return result;
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

AbstractPlatformTransactionManager

如果是使用声明式事务，那么入口就是commit方法。

事务提交的方法由AbstractPlatformTransactionManager控制，逻辑也很明显，processCommit是真正的执行方法。

@Override
public final void commit(TransactionStatus status) throws TransactionException {
	if (status.isCompleted()) {
		throw new IllegalTransactionStateException(
				"Transaction is already completed - do not call commit or rollback more than once per transaction");
	}
	DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status;
	if (defStatus.isLocalRollbackOnly()) {
		if (defStatus.isDebug()) {
			logger.debug("Transactional code has requested rollback");
		}
		processRollback(defStatus, false);
		return;
	}
	if (!shouldCommitOnGlobalRollbackOnly() && defStatus.isGlobalRollbackOnly()) {
		if (defStatus.isDebug()) {
			logger.debug("Global transaction is marked as rollback-only but transactional code requested commit");
		}
		processRollback(defStatus, true);
		return;
	}
	processCommit(defStatus);
}

// processCommit方法定义了流程处理的主体结构，然后通过doCommit方法，让具体的事务管理器去实现。
private void processCommit(DefaultTransactionStatus status) throws TransactionException {
	try {
		boolean beforeCompletionInvoked = false;
		try {
			boolean unexpectedRollback = false;
			prepareForCommit(status);
			// 这两步就对应这beforeCommit、beforeCompletion的逻辑
			triggerBeforeCommit(status);
			triggerBeforeCompletion(status);
			beforeCompletionInvoked = true;
			if (status.hasSavepoint()) {
				if (status.isDebug()) {
					logger.debug("Releasing transaction savepoint");
				}
				unexpectedRollback = status.isGlobalRollbackOnly();
				status.releaseHeldSavepoint();
			}
			else if (status.isNewTransaction()) {
				if (status.isDebug()) {
					logger.debug("Initiating transaction commit");
				}
				unexpectedRollback = status.isGlobalRollbackOnly();
				// 真正执行事务提交的方法，交给了具体的事务处理器去执行
				doCommit(status);
			}
			else if (isFailEarlyOnGlobalRollbackOnly()) {
				unexpectedRollback = status.isGlobalRollbackOnly();
			}
			// Throw UnexpectedRollbackException if we have a global rollback-only
			// marker but still didn't get a corresponding exception from commit.
			if (unexpectedRollback) {
				throw new UnexpectedRollbackException(
						"Transaction silently rolled back because it has been marked as rollback-only");
			}
		}
		catch (UnexpectedRollbackException ex) {
			// can only be caused by doCommit
			triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);
			throw ex;
		}
		catch (TransactionException ex) {
			// can only be caused by doCommit
			if (isRollbackOnCommitFailure()) {
				doRollbackOnCommitException(status, ex);
			}
			else {
				triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);
			}
			throw ex;
		}
		catch (RuntimeException | Error ex) {
			if (!beforeCompletionInvoked) {
				triggerBeforeCompletion(status);
			}
			doRollbackOnCommitException(status, ex);
			throw ex;
		}
		// Trigger afterCommit callbacks, with an exception thrown there
		// propagated to callers but the transaction still considered as committed.
		try {
			// 当事务提交成功之后，调用triggerAfterCommit方法，afterCommit逻辑入口也在这
			triggerAfterCommit(status);
		}
		finally {
			// afterCompletion的入口。
			triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
		}
	}
	finally {
		cleanupAfterCompletion(status);
	}
}


// 调用TransactionSynchronizationUtils类的triggerAfterCommit方法
private void triggerAfterCommit(DefaultTransactionStatus status) {
	if (status.isNewSynchronization()) {
		TransactionSynchronizationUtils.triggerAfterCommit();
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

TransactionSynchronizationUtils

到了TransactionSynchronizationUtils之后，处理流程就很简单了，获取TransactionSynchronization集合，挨个遍历调用即可。

public static void triggerAfterCommit() {
	invokeAfterCommit(TransactionSynchronizationManager.getSynchronizations());
}

public static void invokeAfterCommit(@Nullable List<TransactionSynchronization> synchronizations) {
	if (synchronizations != null) {
		for (TransactionSynchronization synchronization : synchronizations) {
			// 最终调用到afterCommit方法
			synchronization.afterCommit();
		}
	}
}	
1
2
3
4
5
6
7
8
9
10
11
12

通过以上的流程分析可以看出，Spring就是简单的把执行afterCommit方法放在事务提交后调用。

2.获取TransactionSynchronization集合

根据前面的流程分析可能，synchronizations集合是通过调用TransactionSynchronizationManager.getSynchronizations()方法获取的。

public static void triggerAfterCommit() {
	invokeAfterCommit(TransactionSynchronizationManager.getSynchronizations());
}
1
2
3

getSynchronizations

private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations =
		new NamedThreadLocal<>("Transaction synchronizations");

public static List<TransactionSynchronization> getSynchronizations() throws IllegalStateException {
	// synchronizations是TransactionSynchronizationManager类中定义的一个ThreadLocal全局变量
	Set<TransactionSynchronization> synchs = synchronizations.get();
	// 如果为null，则抛出异常
	if (synchs == null) {
		throw new IllegalStateException("Transaction synchronization is not active");
	}
	// Return unmodifiable snapshot, to avoid ConcurrentModificationExceptions
	// while iterating and invoking synchronization callbacks that in turn
	// might register further synchronizations.
	// 为空，直接返回
	if (synchs.isEmpty()) {
		return Collections.emptyList();
	}
	else {
		// Sort lazily here, not in registerSynchronization.
		// 排序后，返回集合快照
		List<TransactionSynchronization> sortedSynchs = new ArrayList<>(synchs);
		OrderComparator.sort(sortedSynchs);
		return Collections.unmodifiableList(sortedSynchs);
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

3.synchronizations添加元素

很明显，现在问题又变成了synchronizations这个ThreadLocal全局变量是何时被赋值的？

首先在事务开始前，就会通过调用initSynchronization方法，先对其进行初始化，这个方法入口也很容易找到，就在TransactionTemplate的execute方法中，通过调用this.transactionManager.getTransaction(this)方法，完成初始化。

public static void initSynchronization() throws IllegalStateException {
	if (isSynchronizationActive()) {
		throw new IllegalStateException("Cannot activate transaction synchronization - already active");
	}
	synchronizations.set(new LinkedHashSet<>());
}
1
2
3
4
5
6

TransactionSynchronizationManager还提供了registerSynchronization方法可以对其进行添加元素。

public static void registerSynchronization(TransactionSynchronization synchronization)
		throws IllegalStateException {
	Assert.notNull(synchronization, "TransactionSynchronization must not be null");
	Set<TransactionSynchronization> synchs = synchronizations.get();
	if (synchs == null) {
		throw new IllegalStateException("Transaction synchronization is not active");
	}
	synchs.add(synchronization);
}
1
2
3
4
5
6
7
8
9

4.实际使用方式

所以，一般我们自定义的方法都是这样实现的，通过匿名类的方式，直接添加一个TransactionSynchronization类型的元素到synchronizations集合中

TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
    @Override
    public void afterCompletion(int status) {
        // 实现你的业务逻辑，在前面分析的invokeAfterCommit方法中，会调用到这里
    }
});
1
2
3
4
5
6