【Spring框架】Spring事务同步

一、什么是Spring事务同步

二、事务同步管理器

2.1 TransactionSynchronizationManager事务同步管理器

2.1.1 资源同步

2.1.2 事务同步

2.1.3 总结

三、事务同步管理器保障事务的原理

四、spring事务为何使用TransactionSynchronizationManager

spring源码实现

五、TransactionSynchronization--(before/after-commit/comple)

一、什么是Spring事务同步

Spring 将 JDBC 的 Connection、Hibernate 的 Session 等访问数据库的连接或者会话对象统称为资源，这些资源在同一时刻是不能多线程共享的。为了让 DAO 或 Service 类可以实现单例模式，Spring的事务同步管理器org.springframework.transaction.support.TransactionSynchronizationManager类利用 ThreadLocal 为不同的事务线程提供了独立的资源副本，并同时维护这些事务的配置属性和运行状态信息。它提供了一些静态方法来操作和获取线程绑定的资源，如bindResource、getResource、unbindResource等。

Spring的事务同步还涉及到事务传播行为，即在一个事务方法中调用另一个事务方法时，后者的事务如何与前者的事务关联。

总的来说Spring事务同步的作用就是在不同的事务线程中保证资源的一致性和事务的正确性。Spring事务同步的主要功能有：

通过ThreadLocal为每个事务线程提供独立的资源副本，如数据库连接、会话对象等，避免线程安全问题。
通过TransactionSynchronizationManager类管理和操作线程绑定的资源，如绑定、获取、解绑等。
通过事务传播行为控制不同事务方法之间的事务关联，如是否使用同一个事务、是否创建新的事务等。
通过事务同步器（TransactionSynchronization）实现事务的扩展功能，如在事务提交前后执行一些额外的操作。

二、事务同步管理器

2.1 TransactionSynchronizationManager事务同步管理器

TransactionSynchronizationManager事务同步管理器，管理每个线程的资源（对于事务，DataSource创建的连接对象connection等称作事务的资源）和事务同步（TransactionSynchronization---用来监听事务操作的回调类，其中定义了在事务执行过程中，进行的拓展操作，如before/after--commit/completion，在getSynchronizations中，对其进行sort排序返回）。

同步分两种，资源的同步和事务的同步。

2.1.1 资源同步

此处就是数据库DataSource的连接connection，保证在一个线程中的事务操作，能够获取同一个connection资源。因此资源存入ThreadLocal<Map<Object, Object>> resources中，保证线程之间的事务操作的隔离（因为获取不同connection，由数据库事务实现spring事务）

2.1.2 事务同步

也就是事务方法同步，synchronizations内的TransactionSynchronization对象的集合，其用来在事务提交前、后，事务完成前后进行的实际操作，其事务操作在各个阶段的执行流程在AbstractPlatformTransactionManager中定义。

2.1.3 总结

TransactionSynchronizationManager 通过 ThreadLocal 对象在当前线程记录了 resources 和 synchronizations属性。resources 是一个 HashMap，用于记录当前参与事务的事务资源，方便进行事务同步，在 DataSourceTransactionManager 的例子中就是以 dataSource 作为 key，保存了数据库连接，这样在同一个线程中，不同的方法调用就可以通过 dataSource 获取相同的数据库连接，从而保证所有操作在一个事务中进行。synchronizations 属性是一个 TransactionSynchronization对象的集合，AbstractPlatformTransactionManager 类中定义了事务操作各个阶段的调用流程

三、事务同步管理器保障事务的原理

spring的事务通过数据库DataSource获取connection来实现，为了使事务方法service.methodA，调用dao.methodB时，仍然能够位于当前事务，如此，能够使得service和dao的调用，在同线程的情况下，都可以获取到相同的connection，就保证了两个操作都在同一个事务。所以需要将connection共享，并考虑使用线程共享变量，threadLocal<Map(datasource,connection)>保存共享的connection。

public abstract class TransactionSynchronizationManager {
    /**…………………………………………………………………………………………………………同步资源……………………………………………………………………………………………… */
    // resource相当于一个(threadID,map(datasource,connectionHolder))的属性，这里用ThreadLocal保存
    // key为DataSource，value为connectionHolder(保存当前threadID的connection)
    // 也就是存储的当前线程ID中不同的数据源DataSource对应的connection，这样能保证在同一个事务线程中，获取相同数据源DataSource的connection都是同一个connection
    private static final ThreadLocal<Map<Object, Object>> resources =
            new NamedThreadLocal<>("Transactional resources");
    
    /**………………………………………………………………………TransactionSynchronization管理…………………………………………………………………………………………… */
    // 当前线程所需要的事务同步器TransactionSynchronization集合
    // TransactionSynchronization用来在事务的执行阶段前后，进行回调操作，如before/after-commit/completion
    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");
    // 当前事务是否active
    private static final ThreadLocal<Boolean> actualTransactionActive =
            new NamedThreadLocal<>("Actual transaction active");
    
    // 获取当前是否存在事务（判断线程共享变量，是否存在TransactionSynchronization
    public static boolean isSynchronizationActive() {
        return (synchronizations.get() != null);
    }
       
    /**……………………………………………………………………获取connection资源……………………………………………………………………………… */
    @Nullable
    public static Object getResource(Object key) {
        Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
        return doGetResource(actualKey);
    }
    @Nullable
    private static Object doGetResource(Object actualKey) {
        Map<Object, Object> map = resources.get();
        if (map == null) {
            return null;
        }
        Object value = map.get(actualKey);
        // Transparently remove ResourceHolder that was marked as void（无效的）...
        if (value instanceof ResourceHolder && ((ResourceHolder) value).isVoid()) {
            map.remove(actualKey);
            // Remove entire ThreadLocal if empty...
            if (map.isEmpty()) {
                resources.remove();
            }
            value = null;
        }
        return value;
    }
    
    //DataSourceTransactionManager.doBegin方法中，将新创建的connection包装成connectionHolder,并存入ThreadLocal<Map<Object, Object>> 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);
        }
        Object oldValue = map.put(actualKey, value);
        // Transparently suppress a ResourceHolder that was marked as void...
        if (oldValue instanceof ResourceHolder && ((ResourceHolder) oldValue).isVoid()) {
            oldValue = null;
        }
        if (oldValue != null) {
            throw new IllegalStateException(
                    "Already value [" + oldValue + "] for key [" + actualKey + "] bound to thread");
        }
    }
    
  /**……………………………………………………………………获取TransactionSynchronization资源……………………………………………………………………………… */
    public static List<TransactionSynchronization> getSynchronizations() throws IllegalStateException {
        Set<TransactionSynchronization> synchs = synchronizations.get();
        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);
            //TransactionSynchronization进行排序
            OrderComparator.sort(sortedSynchs);
            return Collections.unmodifiableList(sortedSynchs);
        }
    }
}

四、spring事务为何使用TransactionSynchronizationManager

一个web项目的主要逻辑模块如下：

spring的事务，通常是在service层的某个方法做完整的事务。service要完成事务，需要如下点：

service在系统中是个单例对象，且service需要通过持有DataSource.connection连接对象，通过数据库事务来实现spring事务；
一个DataSource可以创建多个connection，每个conn需要被一个线程在执行service时候持有，才能在当前调用中完整的通过获取同一个conn实现事务的begin、commit、rollback；
service调用，需要与DAO层交互，因此也要保证DAO也可以获取同一个conn；
Datasource应该单独放在一个类中，以便对于不同的用户线程执行service事务时DataSource.getConnection获取连接。

因此，考虑如上几点需求，需要将不同线程对应的conn存放在ThreadLocal中，能够保证事务执行中，同个线程都可以跨模块获取该conn，保证处于同一事务中。

spring中通过事务同步管理器TransactionSynchronizationManager实现上述推论。

public abstract class TransactionSynchronizationManager {
    /**…………………………………………………………………………………………………………同步资源……………………………………………………………………………………………… */
    //resource相当于一个(threadID,map(datasource,connectionHolder))的属性，这里用ThreadLocal保存
    //key为DataSource，value为connectionHolder(保存当前threadID的connection)
    private static final ThreadLocal<Map<Object, Object>> resources =
            new NamedThreadLocal<>("Transactional resources");
    
    ...
}

spring事务中DataSource，通过DataSourceTransactionManager保存。

spring源码实现

事务开启，通过transactionManager.getTransaction。

// AbstractPlatformTransactionManager.java
public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition)
        throws TransactionException {
    TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults());
    // 获取事务对象DataSourceTransactionObject
    Object transaction = doGetTransaction();
    ...
}
// DataSourceTransactionManager.java
protected Object doGetTransaction() {
    DataSourceTransactionObject txObject = new DataSourceTransactionObject();
    txObject.setSavepointAllowed(isNestedTransactionAllowed());
    ConnectionHolder conHolder =
        //TransactionSynchronizationManager通过DataSource创建connection，包装成ConnectionHolder
        (ConnectionHolder) TransactionSynchronizationManager.getResource(obtainDataSource());
    //设置conHolder入事务对象DataSourceTransactionObject
    txObject.setConnectionHolder(conHolder, false);
    return txObject;
}
 
public abstract class TransactionSynchronizationManager {
    private static final ThreadLocal<Map<Object, Object>> resources =
            new NamedThreadLocal<>("Transactional resources");
 
    public static Object getResource(Object key) {
        Object actualKey = TransactionSynchronizationUtils.unwrapResourceIfNecessary(key);
        Object value = doGetResource(actualKey);
        return value;
    }
 
    private static Object doGetResource(Object actualKey) {
        //从ThreadLocal<Map<Object, Object>> resources中获取当前线程对应的事务连接connection
        Map<Object, Object> map = resources.get();
        if (map == null) {
            return null;
        }
        Object value = map.get(actualKey);
        return value;
    }
}

五、TransactionSynchronization--(before/after-commit/comple)

事务同步器，我们可以自定义实现TransactionSynchronization类，来监听Spring的事务操作，通过TransactionSynchronization类的这些回调方法做一些扩展。

public interface TransactionSynchronization extends Flushable {
    //描述事务当前状态
    int STATUS_COMMITTED = 0;
    int STATUS_ROLLED_BACK = 1;
    int STATUS_UNKNOWN = 2;
    //挂起该事务同步器
    default void suspend() {}
    //恢复事务同步器
    default void resume() {}
    //flush 底层的session到数据库
    @Override
    default void flush() {}
    //事务提交前的回调
    default void beforeCommit(boolean readOnly) {}
    //事务commit/rollback前的回调，用于在事务完成前进行资源清除
    default void beforeCompletion() {}
    //事务提交后的回调，可以用于在事务成功提交后做的进一步的操作，例如：在数据提交到数据库中后，发送确认的短信或邮件
    default void afterCommit() {}
    //在事务commit/rollback之后进行回调，例如可以在事务完成后做一些资源清除
    default void afterCompletion(int status) {}
}

举例：

TransactionSynchronization是注册在TransactionSynchronizationManager内，需要其内的方法判断是否存在事务，是否可以执行事务同步方法。

// 当前事务提交后方可进行异步任务，防止异步任务先于未提交的事务执行
private void callBack(Invoice invoice){
    boolean synchronizationActive = TransactionSynchronizationManager.isSynchronizationActive();
    // 当前存在事务，在事务提交后执行  
    if (synchronizationActive) { 
        TransactionSynchronizationManager.registerSynchronization(
            new TransactionSynchronizationAdapter() {
                @Override
                public void afterCommit() { // 监听事务提交完成
                    // 事务提交后执行异步任务
                    doCall(invoice);
                }
            }
        );
    } else {
        // 当前不存在事务，直接执行
        doCall(invoice);
    }
}