TransactionSynchronizationManager(事务监听)

TransactionSynchronizationManager是事务同步管理器。我们可以自定义实现TransactionSynchronization类，来监听Spring的事务操作。可以在事务提交之后，回调TransactionSynchronization类的方法。

  1. TransactionSynchronizationManager在源码中的使用
    在SpringCache的自定义CacheManager中。装饰Cache对象使其支持事务操作。即只有在事务提交成功之后，才会进行缓存。

源码位置：org.springframework.cache.transaction.TransactionAwareCacheDecorator#put

@Override
public void put(final Object key, @Nullable final Object value) {
    if (TransactionSynchronizationManager.isSynchronizationActive()) {
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
            @Override
            public void afterCommit() {
                TransactionAwareCacheDecorator.this.targetCache.put(key, value);
            }
        });
    }
    else {
        this.targetCache.put(key, value);
    }
}

【JDBC中的connection详解】中，我们知道connection是线程不安全的，即需要为每一个数据库操作都获取一个Connection对象。事务操作可以看做是一个整体，必须使用同一个Connection进行操作。故在Spring中使用LocalThread(线程上下文)将Connection对象和线程绑定。

在Spring中的org.springframework.transaction.support.TransactionSynchronizationManager类中，便是使用ThreadLocal来为不同的事务线程提供独立的资源副本，并且同时维护这些事务的配置属性和运行状态。

  1. Connection与TransactionSynchronizationManager关系
  2. 请求事务方法时，调用dobegin()将事务信息保存到TransactionSynchronizationManager中：在该方法中主要流程是在数据库连接池中获取一个Connection对象，然后将Connection对象放入到ThreadLocal中。实际上该事务方法的信息均由TransactionSynchronizationManager类管理。

源码：org.springframework.jdbc.datasource.DataSourceTransactionManager#doBegin

protected void doBegin(Object transaction, TransactionDefinition definition) {
DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
Connection con = null;

    try {
        if (!txObject.hasConnectionHolder() ||
                txObject.getConnectionHolder().isSynchronizedWithTransaction()) {
                //在可见的数据源(连接池)中获取Connection对象
            Connection newCon = obtainDataSource().getConnection();
            if (logger.isDebugEnabled()) {
                logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction");
            }
                    
            txObject.setConnectionHolder(new ConnectionHolder(newCon), true);
        }
        
        ...
      //关闭Connection对象的自动提交
       if (con.getAutoCommit()) {
            txObject.setMustRestoreAutoCommit(true);
            if (logger.isDebugEnabled()) {
                logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
            }
            con.setAutoCommit(false);
        }

        //将Connection对象绑定到Thread中
        if (txObject.isNewConnectionHolder()) {
            TransactionSynchronizationManager.bindResource(obtainDataSource(), txObject.getConnectionHolder());
        }
    }

    catch (Throwable ex) {
        if (txObject.isNewConnectionHolder()) {
            DataSourceUtils.releaseConnection(con, obtainDataSource());
            txObject.setConnectionHolder(null, false);
        }
        throw new CannotCreateTransactionException("Could not open JDBC Connection for transaction", ex);
    }
}

public abstract class TransactionSynchronizationManager {
private static final ThreadLocal<Map<Object, Object>> resources =
new NamedThreadLocal<>(“Transactional resources”);

public static void bindResource(Object key, Object value) throws IllegalStateException {
    Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
    Assert.notNull(value, "Value must not be null");
   
    Map<Object, Object> map = resources.get();
    // set ThreadLocal Map if none found
    if (map == null) {
        map = new HashMap<>();
        resources.set(map);
    }
    //将Connection对象绑定到resources 上。
    Object oldValue = map.put(actualKey, value);
    ...
}

}
执行sql语句时，实际上通过org.mybatis.spring.transaction.SpringManagedTransaction类直接获取Connection对象。
源码：org.mybatis.spring.transaction.SpringManagedTransaction#openConnection

private void openConnection() throws SQLException {
//在ThreadLocal中获取Connection对象
this.connection = DataSourceUtils.getConnection(this.dataSource);
this.autoCommit = this.connection.getAutoCommit();
//在ThreadLocal中获取是否开启事务
this.isConnectionTransactional = DataSourceUtils.isConnectionTransactional(this.connection, this.dataSource);
…
}
源码：org.springframework.jdbc.datasource.DataSourceUtils#doGetConnection

public static Connection doGetConnection(DataSource dataSource) throws SQLException {
    Assert.notNull(dataSource, "No DataSource specified");
    //在doBegin()方法中，已经将创建的Connection对象放入到TransactionSynchronizationManager中
    ConnectionHolder conHolder = (ConnectionHolder) TransactionSynchronizationManager.getResource(dataSource);
    if (conHolder != null && (conHolder.hasConnection() || conHolder.isSynchronizedWithTransaction())) {
        conHolder.requested();
        if (!conHolder.hasConnection()) {
            logger.debug("Fetching resumed JDBC Connection from DataSource");
            conHolder.setConnection(fetchConnection(dataSource));
        }
        //直接返回Thread存储的Connection对象。
        return conHolder.getConnection();
    }

    ...

    return con;
}
  1. TransactionSynchronizationManager的结构

public abstract class TransactionSynchronizationManager {

 //线程上下文中保存着【线程池对象：ConnectionHolder】的Map对象。线程可以通过该属性获取到同一个Connection对象。
private static final ThreadLocal<Map<Object, Object>> resources = new NamedThreadLocal<>("Transactional resources");

//事务同步器，是Spring交由程序员进行扩展的代码，每个线程可以注册N个事务同步器。
private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations = new NamedThreadLocal<>("Transaction synchronizations");

// 事务的名称  
private static final ThreadLocal<String> currentTransactionName = new NamedThreadLocal<>("Current transaction name");
// 事务是否是只读  
private static final ThreadLocal<Boolean> currentTransactionReadOnly = new NamedThreadLocal<>("Current transaction read-only status");
// 事务的隔离级别
private static final ThreadLocal<Integer> currentTransactionIsolationLevel = new NamedThreadLocal<>("Current transaction isolation level");
// 事务是否开启   actual：真实的
private static final ThreadLocal<Boolean> actualTransactionActive = new NamedThreadLocal<>("Actual transaction active");

}
在org.springframework.transaction.interceptor.TransactionInterceptor#invoke中，对事务方法进行拦截处理。在createTransactionIfNecessary创建TransactionInfo对象时，会调用AbstractPlatformTransactionManager#prepareSynchronization方法初始化事务同步器。

protected void prepareSynchronization(DefaultTransactionStatus status, TransactionDefinition definition) {
    if (status.isNewSynchronization()) {
        TransactionSynchronizationManager.setActualTransactionActive(status.hasTransaction());
        TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(
                definition.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT ?
                        definition.getIsolationLevel() : null);
        TransactionSynchronizationManager.setCurrentTransactionReadOnly(definition.isReadOnly());
        TransactionSynchronizationManager.setCurrentTransactionName(definition.getName());
        //初始化事务同步器
        TransactionSynchronizationManager.initSynchronization();
    }
}
  1. TransactionSynchronization
    这个类是程序员对事务同步的扩展点：用于事务同步回调的接口。

public interface TransactionSynchronization extends Flushable {

/** Completion status in case of proper commit. */
int STATUS_COMMITTED = 0;

/** Completion status in case of proper rollback. */
int STATUS_ROLLED_BACK = 1;

/** Completion status in case of heuristic mixed completion or system errors. */
int STATUS_UNKNOWN = 2;


/**
 * 事务挂起
 * Supposed to unbind resources from TransactionSynchronizationManager if managing any.
 * @see TransactionSynchronizationManager#unbindResource
 */
default void suspend() {
}

/**
 * 事务恢复
 * Supposed to rebind resources to TransactionSynchronizationManager if managing any.
 * @see TransactionSynchronizationManager#bindResource
 */
default void resume() {
}

/**
 * 将基础会话刷新到数据存储区(如果适用)，比如Hibernate/JPA的Session
 * @see org.springframework.transaction.TransactionStatus#flush()
 */
@Override
default void flush() {
}

/**
 * 在事务提交前触发，此处若发生异常，会导致回滚。
 * @see #beforeCompletion
 */
default void beforeCommit(boolean readOnly) {
}

/**
 * 在beforeCommit之后，commit/rollback之前执行。即使异常，也不会回滚。
 * @see #beforeCommit
 * @see #afterCompletion
 */
default void beforeCompletion() {
}

/**
 * 事务提交后执行。
 */
default void afterCommit() {
}

/**
 * 事务提交/回滚执行
 */
default void afterCompletion(int status) {
}

一般而言，我们在TransactionSynchronization使用最多的是afterCommit和afterCompletion方法。可以在事务执行完毕之后，直接调用afterCommit()方法进行异步通知。

我们在doCommit()方法中提交事务后，在cleanupAfterCompletion对connection进行重置，即我们依旧可以在afterCommit()回调中对数据库进行操作。

private void processCommit(DefaultTransactionStatus status) throws TransactionException {
try {
//提交事务
doCommit(status);
…
try {
//回调所有事务同步器的afterCommit方法。
triggerAfterCommit(status);
}
finally {
//回调所有事务同步器的afterCompletion方法。
triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
}

}
finally {
   //清除TransactionSynchronizationManager的ThreadLocal绑定的数据。
   //解除Thread绑定的resources资源。
   //将Commit设置为自动提交。
  //清理ConnectionHolder资源。
    cleanupAfterCompletion(status);
}

}

作者：小胖学编程
链接：https://www.jianshu.com/p/4b5eb29cc6d9
来源：简书
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。