@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@Documentedpublic @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@Slf4jpublic 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")); }}

@Servicepublic 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@EventListenerpublic @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();}

/** * 重写接口中的创建监听器方法 */@Overridepublic 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); } } } } //其余代码略 }