深入了解 Spring 中的事务（从核心注解和类入手）

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(TransactionManagementConfigurationSelector.class)
public @interface EnableTransactionManagement {
    /**
     * 指定基于目标类代理还是基于接口代理。
     * 默认采用JDK官方的基于接口代理。
     * @return
     */
    boolean proxyTargetClass() default false;

    /**
     * 指定事务通知是如何执行的。默认是通过代理方式执行的。
     * 如果是同一个类中调用的话，请采用AdviceMode.ASPECTJ
     * @return
     */
    AdviceMode mode() default AdviceMode.PROXY;

    /**
     * 指示在特定连接点应用多个通知时事务处理的执行顺序。
     * 默认值是：最低优先级（Integer.MAX_VALUE）
     * @return
     */
    int order() default Ordered.LOWEST_PRECEDENCE;
}

@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface Transactional {
    /**
     * 指定事务管理器的唯一标识
     */
    @AliasFor("transactionManager") String value() default "";

    /**
     * 指定事务管理器的唯一标识
     */
    @AliasFor("value") String transactionManager() default "";

    /**
     * 指定事务的传播行为
     */
    Propagation propagation() default Propagation.REQUIRED;

    /**
     * 指定事务的隔离级别
     */
    Isolation isolation() default Isolation.DEFAULT;

    /**
     * 指定事务的超时时间
     */
    int timeout() default TransactionDefinition.TIMEOUT_DEFAULT;

    /**
     * 指定事务是否只读
     */
    boolean readOnly() default false;

    /**
     * 通过指定异常类的字节码，限定事务在特定情况下回滚
     */
    Class<? extends Throwable>[] rollbackFor() default {};

    /**
     * 通过指定异常类的全限定类名，限定事务在特定情况下回滚
     */
    String[] rollbackForClassName() default {};

    /**
     * 通过指定异常类的字节码，限定事务在特定情况下不回滚
     */
    Class<? extends Throwable>[] noRollbackFor() default {};

    /**
     * 通过指定异常类的全限定类名，限定事务在特定情况下不回滚
     */
    String[] noRollbackForClassName() default {};
}

// 通过 @Import注解，Spring 定义了一个事务管理配置类的导入器。
@Import(TransactionManagementConfigurationSelector.class) 
public @interface EnableTransactionManagement { 
  //其余代码略 
}

@Bean 
@Role(BeanDefinition.ROLE_INFRASTRUCTURE) 
public TransactionAttributeSource transactionAttributeSource() { 
  //创建了一个注解事务的属性解析对象 
  return new AnnotationTransactionAttributeSource(); 
}

public class AnnotationTransactionAttributeSource extends AbstractFallbackTransactionAttributeSource implements Serializable {
    /**
     * 默认构造函数
     */
    public AnnotationTransactionAttributeSource() {
        this(true);
    }

    /**
     * 带参构造（表明是否为public方法）
     */
    public AnnotationTransactionAttributeSource(boolean publicMethodsOnly) {
        this.publicMethodsOnly = publicMethodsOnly;
        //判断是不是jta或者是ejp
        if (jta12Present || ejb3Present) {
            this.annotationParsers = new LinkedHashSet<>(4);
            this.annotationParsers.add(new SpringTransactionAnnotationParser());
            //jta
            if (jta12Present) {
                this.annotationParsers.add(new
                        JtaTransactionAnnotationParser());
            }
            //ejp
            if (ejb3Present) {
                this.annotationParsers.add(new
                        Ejb3TransactionAnnotationParser());
            }
        } else {
            //当都不是的时候，构建一个SpringTransactionAnnotationParser（事务注解解析器）
            this.annotationParsers = Collections.singleton(new
                    SpringTransactionAnnotationParser());
        }
    }
}

public class SpringTransactionAnnotationParser implements TransactionAnnotationParser, Serializable {
    /**
     * 根据传入被注解的元素解析。
     * 可以是Method,Field,Class,Package,Construct等等。
     */
    @Override
    @Nullable
    public TransactionAttribute parseTransactionAnnotation(AnnotatedElement element) {
        AnnotationAttributes attributes = AnnotatedElementUtils.findMergedAnnotationAttributes(element, Transactional.class, false, false);
        if (attributes != null) { //调用根据传入注解属性解析
            return parseTransactionAnnotation(attributes);
        } else {
            return null;
        }
    }

    /**
     * 根据传入注解解析
     */
    public TransactionAttribute parseTransactionAnnotation(Transactional ann) {
        return parseTransactionAnnotation(AnnotationUtils.getAnnotationAttributes(ann, false, false));
    }

    /**
     * 根据传入的注解属性解析
     */
    protected TransactionAttribute parseTransactionAnnotation(AnnotationAttributes
                                                                      attributes) {
        RuleBasedTransactionAttribute rbta = new RuleBasedTransactionAttribute();
        Propagation propagation = attributes.getEnum("propagation");
        rbta.setPropagationBehavior(propagation.value());
        Isolation isolation = attributes.getEnum("isolation");
        rbta.setIsolationLevel(isolation.value());
        rbta.setTimeout(attributes.getNumber("timeout").intValue());
        rbta.setReadOnly(attributes.getBoolean("readOnly"));
        rbta.setQualifier(attributes.getString("value"));
        List<RollbackRuleAttribute> rollbackRules = new ArrayList<>();
        for (Class<?> rbRule : attributes.getClassArray("rollbackFor")) {
            rollbackRules.add(new RollbackRuleAttribute(rbRule));
        }
        for (String rbRule : attributes.getStringArray("rollbackForClassName")) {
            rollbackRules.add(new RollbackRuleAttribute(rbRule));
        }
        for (Class<?> rbRule : attributes.getClassArray("noRollbackFor")) {
            rollbackRules.add(new NoRollbackRuleAttribute(rbRule));
        }
        for (String rbRule : attributes.getStringArray("noRollbackForClassName")) {
            rollbackRules.add(new NoRollbackRuleAttribute(rbRule));
        }
        rbta.setRollbackRules(rollbackRules);
        return rbta;
    }
    //其余代码略
}

public class MyApplicationEvent extends ApplicationEvent {
    public MyApplicationEvent(Object source) {
        super(source);
    }
}

@Component
@Slf4j
public class MyTransactionEventListener {
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void doSomething(MyApplicationEvent event) {
        Map map = (Map) event.getSource();
        log.info("事务提交后执行===>转出账户：" + map.get("sourceName") + ",转入账户：" + map.get("targetName") + ",转账金额：" + map.get("money"));
    }

    @TransactionalEventListener(phase = TransactionPhase.AFTER_ROLLBACK)
    public void otherSomething(MyApplicationEvent event) {
        Map map = (Map) event.getSource();
        log.info("事务回滚后执行===>转出账户：" + map.get("sourceName") + ",转入账户：" + map.get("targetName") + ",转账金额：" + map.get("money"));
    }
}

@Service
public class AccountServiceImpl implements AccountService {
    @Autowired
    private AccountDao accountDao;
    @Autowired
    private ApplicationEventPublisher applicationEventPublisher;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void transfer(String sourceName, String targetName, Double money) {
        try {
            //1.根据名称查询转出账户
            Account source = accountDao.findByName(sourceName);
            //2.根据名称查询转入账户
            Account target = accountDao.findByName(targetName);
            //3.转出账户减钱
            source.setMoney(source.getMoney() - money);
            //4.转入账户加钱
            target.setMoney(target.getMoney() + money);
            accountDao.update(source); 
            // 模拟转账异常
            int i = 1 / 0; 
            //6.更新转入账户
            accountDao.update(target);
        } finally {
            Map<String, Object> map = new HashMap<>();
            map.put("sourceName", sourceName);
            map.put("targetName", targetName);
            map.put("money", money);
            // 发布自定义事件（此处可以发布多个事件，如邮件通知、短信通知等，利用自定义监听器实现代码解耦）
            applicationEventPublisher.publishEvent(new MyApplicationEvent(map));
        }
    }
}

@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@EventListener
public @interface TransactionalEventListener {
    /**
     * 指定事务监听器的执行是在何时。
     * 取值有：
     *      事务提交之前
     *      事务提交之后（默认值）
     *      事务回滚之后
     *      事务执行完成之后
     */
    TransactionPhase phase() default TransactionPhase.AFTER_COMMIT;

    /**
     * 若没有事务的时候，对应的event是否已经执行
     * 默认值为false表示没事务就不执行了
     */
    boolean fallbackExecution() default false;

    /**
     * 指定事件类的字节码
     */
    @AliasFor(annotation = EventListener.class, attribute = "classes") Class<?>[] value() default {};

    /**
     * 它和value属性的作用是一样的
     */
    @AliasFor(annotation = EventListener.class, attribute = "classes") Class<?>[] classes() default {};

    /**
     * 用于指定执行事件处理器的条件。取值是基于Spring的el表达式编写的。
     */
    String condition() default "";
}

// Instantiate all remaining (non-lazy-init) singletons. 
finishBeanFactoryInitialization(beanFactory);//触发其他单例bean的加载

/**
 * 初始化剩余单例bean对象的方法
 */
protected void finishBeanFactoryInitialization(ConfigurableListableBeanFactory beanFactory) {
    // 方法中的其他代码略
    // 初始化剩余单例bean对象.调用的是DefaultListableBeanFactory类中的preInstantiateSingletons方法。
    beanFactory.preInstantiateSingletons();
}

/**
 * 重写接口中的创建监听器方法
 */
@Override
public ApplicationListener<?> createApplicationListener(String beanName, Class<?> type, Method method) {
    return new ApplicationListenerMethodTransactionalAdapter(beanName, type, method);
}

/**
 * 构造函数
 */
public ApplicationListenerMethodTransactionalAdapter(String beanName, Class<?> targetClass, Method method) {
    super(beanName, targetClass, method);
    TransactionalEventListener ann = AnnotatedElementUtils.findMergedAnnotation(method, TransactionalEventListener.class);
    if (ann == null) {
        throw new IllegalStateException("No TransactionalEventListener annotation found on method: "+method);
    }
    this.annotation = ann;
}

public class TransactionTemplate extends DefaultTransactionDefinition implements TransactionOperations, InitializingBean {
    /**
     * 定义日志组件.
     */
    protected final Log logger = LogFactory.getLog(getClass());
    /**
     * 定义事务管理器对象
     */
    @Nullable
    private PlatformTransactionManager transactionManager;

    /*** 默认构造函数 */
    public TransactionTemplate() {
    }

    /*** 通过事务管理器构建事务模板对象 */
    public TransactionTemplate(PlatformTransactionManager transactionManager) {
        this.transactionManager = transactionManager;
    }

    /*** 通过事务管理器和事务定义信息构建模板对象 */
    public TransactionTemplate(PlatformTransactionManager transactionManager, TransactionDefinition transactionDefinition) {
        super(transactionDefinition);
        this.transactionManager = transactionManager;
    }

    /*** 当使用默认构造函数构建事务模板对象时，可以通过此方法注入事务管理器 */
    public void setTransactionManager(@Nullable PlatformTransactionManager transactionManager) {
        this.transactionManager = transactionManager;
    }

    /*** 获取使用的事务管理器对象 */
    @Nullable
    public PlatformTransactionManager getTransactionManager() {
        return this.transactionManager;
    }

    /*** 判断事务管理器是否为空 */
    @Override
    public void afterPropertiesSet() {
        if (this.transactionManager == null) {
            throw new IllegalArgumentException("Property 'transactionManager' is required");
        }
    }

    /*** 编程事务控制的核心方法,重写的是TransactionOperations接口中的方法 */
    @Override
    @Nullable
    public <T> T execute(TransactionCallback<T> action) throws TransactionException {
        Assert.state(this.transactionManager != null, "No PlatformTransactionManager set"); 
        // 判断当前事务管理器是否为CallbackPreferringPlatformTransactionManager类型， 如果是的话，直接使用该接口实现类中的execute方法执行。而无需继续让 PlatformTransactionManager的实现类控制事务，当前坐标环境下它只有一个实现类：WebSphereUowTransactionManager。
        if (this.transactionManager instanceof CallbackPreferringPlatformTransactionManager) {
            return ((CallbackPreferringPlatformTransactionManager) this.transactionManager).execute(this, action);
        } else {
            // 需要借助PlatformTransactionManager的实现类控制事务。
            TransactionStatus status = this.transactionManager.getTransaction(this);
            T result;
            try {
                // 执行TransactionCallback中的doInTransaction方法，此处又是策略模式。
                // spring只提供了一个接口（还有一个抽象实现类），而具体需要事务支持的业务代 码由使用者提供。
                result = action.doInTransaction(status);
            } catch (RuntimeException | Error ex) { 
                // 当doInTransaction执行有异常时事务回滚
                rollbackOnException(status, ex);
                throw ex;
            } catch (Throwable ex) { 
                // 当doInTransaction执行有无法预知的异常时，事务回滚。
                rollbackOnException(status, ex);
                throw new UndeclaredThrowableException(ex, "TransactionCallback threw undeclared checked exception");
            }
            // 没有异常的话，事务提交
            this.transactionManager.commit(status); 
            // 返回执行结果（有可能是结果集，也有可能是影响数据库记录的行数）
            return result;
        }
    }

    /** 出现异常事务回滚 */
    private void rollbackOnException(TransactionStatus status, Throwable ex) throws TransactionException {
        Assert.state(this.transactionManager != null, "No PlatformTransactionManager set");
        logger.debug("Initiating transaction rollback on application exception", ex);
        try {
            // 执行事务管理器中提供的回滚方法。
            this.transactionManager.rollback(status);
        } catch (TransactionSystemException ex2) {
            logger.error("Application exception overridden by rollback exception", ex);
            ex2.initApplicationException(ex);
            throw ex2;
        } catch (RuntimeException | Error ex2) {
            logger.error("Application exception overridden by rollback exception", ex);
            throw ex2;
        }
    }

    /*** 重写equals方法 */
    @Override
    public boolean equals(Object other) {
        return (this == other || (super.equals(other) && (!(other instanceof TransactionTemplate) || getTransactionManager() == ((TransactionTemplate) other).getTransactionManager())));
    }
}

public abstract class DataSourceUtils {
    /*** 获取连接的方法，它本身没有任何操作而是调用了doGetConnection */
    public static Connection getConnection(DataSource dataSource) throws CannotGetJdbcConnectionException {
        try {
            // 真正获取连接的方法
            return doGetConnection(dataSource);
        } catch (SQLException ex) {
            throw new CannotGetJdbcConnectionException("Failed to obtain JDBC Connection: ", ex);
        } catch (IllegalStateException ex) {
            throw new CannotGetJdbcConnectionException("Failed to obtain JDBC Connection: " + ex.getMessage());
        }
    }

    /*** 真正从数据源中获取连接的方法 */
    public static Connection doGetConnection(DataSource dataSource) throws SQLException {
        Assert.notNull(dataSource, "No DataSource specified");
        // 通过事务同步管理器对象获取连接持有者对象
        ConnectionHolder conHolder = (ConnectionHolder) TransactionSynchronizationManager.getResource(dataSource); 
        // 当连接持有者不为null时，并且再满足连接持有者有连接或者是同步的事务其中任何一个条件， 则直接返回连接持有者的连接对象。
        // （synchronizedWithTransaction默认值为false。但是在DataSourceTransactionManager中的doBegin方法中对synchronizedWithTransaction属性赋值为 true了。
        if (conHolder != null && (conHolder.hasConnection() || conHolder.isSynchronizedWithTransaction())) {
            conHolder.requested();
            // 如果ConnectionHandle为null，则返回false，此处取反。就表示 ConnectionHandle为null时，进入if代码块中，给ConnectionHolder设置一个连接
            if (!conHolder.hasConnection()) {
                logger.debug("Fetching resumed JDBC Connection from DataSource");
                conHolder.setConnection(fetchConnection(dataSource));
            }// 返回ConnectionHolder对象中的连接
            return conHolder.getConnection();
        }
        logger.debug("Fetching JDBC Connection from DataSource");
        //如果不满足上面的条件，则从数据源中获取一个连接
        Connection con = fetchConnection(dataSource);
        //判断是否激活了事务同步器
        if (TransactionSynchronizationManager.isSynchronizationActive()) {
            try {
                // 在激活同步的条件下，如果ConnectionHolder为null就创建连接持有者对象
                ConnectionHolder holderToUse = conHolder;
                if (holderToUse == null) { //创建连接持有者对象
                    holderToUse = new ConnectionHolder(con);
                } else {
                    // 直接使用已经存在的连接持有者，并把获取到的连接填充进去
                    holderToUse.setConnection(con);
                }
                holderToUse.requested(); //注册同步器
                TransactionSynchronizationManager.registerSynchronization(new ConnectionSynchronization(holderToUse, dataSource)); //设置snchronizedWithTransaction属性为true
                holderToUse.setSynchronizedWithTransaction(true);

                // 从名称为Transactional resources的ThreadLocal中获取绑定的 Map，并把数据源和ConnectionHolder存入map中
                if (holderToUse != conHolder) {
                    TransactionSynchronizationManager.bindResource(dataSource, holderToUse);
                }
            }
        } catch(RuntimeException ex) {
            // Unexpected exception from external delegation call -> close Connection and rethrow.
            releaseConnection(con, dataSource);
            throw ex;
        }
        // 此时如果激活了事务同步管理器，则返回当前线程的连接。如果没激活，返回的就是数据源中拿 到的连接。
        return con;
    }

    /*** 从数据源中获取一个连接的方法，此时没有和线程绑定 */
    private static Connection fetchConnection(DataSource dataSource) throws SQLException {
        //从数据源中获取一个连接
        Connection con = dataSource.getConnection();
        //如果没有，则表示数据源中没有连接
        if (con == null) {
            throw new IllegalStateException("DataSource returned null from getConnection(): " + dataSource);
        }
        // 返回拿到的连接对象
        return con;
    }
    // 其余代码略
}

public abstract class TransactionSynchronizationManager {
    // 定义了很多ThreadLocal
    // 应用代码随事务的声明周期绑定的对象 比如：DataSourceTransactionManager有这么做：
    // TransactionSynchronizationManager.bindResource(obtainDataSource(), txObject.getConnectionHolder());
    // TransactionSynchronizationManager.bindResource(obtainDataSource(), suspendedResources);
    // 简单理解为当前线程的数据存储中心~~~~
    // 使用的同步器，用于应用扩展
    // TransactionSynchronization同步器是最为重要的一个扩展点~~~
    // 这里是个set 所以每个线程都可以注册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");

    // 判断是否是开启了和线程绑定机制。当开启了之后，我们通过DataSourceUtils获取的连接就是当 前线程的连接了。
    public static boolean isSynchronizationActive() {
        return (synchronizations.get() != null);
    }

    /*** 初始化同步器方法。此方法就是在synchronizations中设置了一个LinkedHashSet * 当然如果事务已经开启了，就不能再初始化同步器了 而是直接注册 */
    public static void initSynchronization() throws IllegalStateException {
        if (isSynchronizationActive()) {
            throw new IllegalStateException("Cannot activate transaction synchronization - already active");
        }
        logger.trace("Initializing transaction synchronization");
        synchronizations.set(new LinkedHashSet<>());
    }// 清除所有和当前线程相关的（注意：此处只是clear清除，和当前线程的绑定而已~~~） 

    public static void clear() {
        synchronizations.remove();
        currentTransactionName.remove();
        currentTransactionReadOnly.remove();
        currentTransactionIsolationLevel.remove();
        actualTransactionActive.remove();
    }
    //其余方法略 
}

public class TransactionAwareDataSourceProxy extends DelegatingDataSource {
    /*** 暴露出来的获取连接的方法 */
    @Override
    public Connection getConnection() throws SQLException {
        return getTransactionAwareConnectionProxy(obtainTargetDataSource());
    }

    /*** 使用JDK的动态代理创建连接的代理对象 */
    protected Connection getTransactionAwareConnectionProxy(DataSource targetDataSource) {
        return (Connection) Proxy.newProxyInstance(ConnectionProxy.class.getClassLoader(), new Class<?>[]{ConnectionProxy.class}, new TransactionAwareInvocationHandler(targetDataSource));
    }

    /*** InvocationHandler的具体实现（增强的部分） */
    private class TransactionAwareInvocationHandler implements InvocationHandler {
        private final DataSource targetDataSource;
        @Nullable
        private Connection target;
        private boolean closed = false;

        public TransactionAwareInvocationHandler(DataSource targetDataSource) {
            this.targetDataSource = targetDataSource;
        }

        @Override
        @Nullable
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { // Invocation on ConnectionProxy interface coming in... 
            if (method.getName().equals("equals")) {
                // Only considered as equal when proxies are identical.
                return proxy == args[0];
            } else if (method.getName().equals("hashCode")) {
                // Use hashCode of Connection proxy.
                return System.identityHashCode(proxy);
            } else if (method.getName().equals("toString")) {
                // Allow for differentiating between the proxy and the raw Connection.
                StringBuilder sb = new StringBuilder("Transaction-aware proxy for target Connection ");
                if (this.target != null) {
                    sb.append("[").append(this.target.toString()).append("]");
                } else {
                    sb.append(" from DataSource [").append(this.targetDataSource).append("]");
                }
                return sb.toString();
            } else if (method.getName().equals("unwrap")) {
                if (((Class<?>) args[0]).isInstance(proxy)) {
                    return proxy;
                }
            } else if (method.getName().equals("isWrapperFor")) {
                if (((Class<?>) args[0]).isInstance(proxy)) {
                    return true;
                }
            } else if (method.getName().equals("close")) {
                // 释放资源仍然使用的是DataSourceUtils中的方法
                DataSourceUtils.doReleaseConnection(this.target, this.targetDataSource);
                this.closed = true;
                return null;
            } else if (method.getName().equals("isClosed")) {
                return this.closed;
            }
            if (this.target == null) {
                if (this.closed) {
                    throw new SQLException("Connection handle already closed");
                }
                if (shouldObtainFixedConnection(this.targetDataSource)) {
                    this.target = DataSourceUtils.doGetConnection(this.targetDataSource);
                }
            }
            Connection actualTarget = this.target;
            if (actualTarget == null) { 
                //获取连接使用的也是DataSourceUtils中的方法
                actualTarget = DataSourceUtils.doGetConnection(this.targetDataSource);
            }
            if (method.getName().equals("getTargetConnection")) {
                return actualTarget;
            }// Invoke method on target Connection. 
            try {
                Object retVal = method.invoke(actualTarget, args);
                // If return value is a Statement, apply transaction timeout. 
                // Applies to createStatement, prepareStatement, prepareCall. 
                if (retVal instanceof Statement) {
                    DataSourceUtils.applyTransactionTimeout((Statement) retVal, this.targetDataSource);
                }
                return retVal;
            } catch (InvocationTargetException ex) {
                throw ex.getTargetException();
            } finally {
                if (actualTarget != this.target) {
                    DataSourceUtils.doReleaseConnection(actualTarget, this.targetDataSource);
                }
            }
        }
    }
    //其余代码略 
}