Spring是如何保证同一事务获取同一个Connection的？使用Spring的事务同步机制解决：数据库刚插入的记录却查询不到的问题【享学Spring】

java.sql.SQLException: Connection has already been closed

public abstract class DataSourceUtils {
	...
	public static Connection getConnection(DataSource dataSource) throws CannotGetJdbcConnectionException { ... }
	...
	
	// 把definition和connection进行一些准备工作~
	public static Integer prepareConnectionForTransaction(Connection con, @Nullable TransactionDefinition definition) throws SQLException { ...}

	// Reset the given Connection after a transaction,
	// con.setTransactionIsolation(previousIsolationLevel);和con.setReadOnly(false);等等
	public static void resetConnectionAfterTransaction(Connection con, @Nullable Integer previousIsolationLevel) { ... }

	// 该JDBC Connection 是否是当前事务内的链接~
	public static boolean isConnectionTransactional(Connection con, @Nullable DataSource dataSource) { ... }

	// Statement 给他设置超时时间  不传timeout表示不超时
	public static void applyTransactionTimeout(Statement stmt, @Nullable DataSource dataSource) throws SQLException { ... }
	public static void applyTimeout(Statement stmt, @Nullable DataSource dataSource, int timeout) throws SQLException { ... }

	// 此处可能是归还给连接池，也有可能是close~（和连接池参数有关）
	public static void releaseConnection(@Nullable Connection con, @Nullable DataSource dataSource) { ... }
	public static void doReleaseConnection(@Nullable Connection con, @Nullable DataSource dataSource) throws SQLException { ... }

	// 这个是真close
	public static void doCloseConnection(Connection con, @Nullable DataSource dataSource) throws SQLException { ... }
	
	// 如果链接是代理，会拿到最底层的connection
	public static Connection getTargetConnection(Connection con) { ... }
}

// 此处生路而关于DataSource、PlatformTransactionManager事务管理器等的配置
@Slf4j
@Service
public class HelloServiceImpl implements HelloService {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Transactional
    @Override
    public Object hello(Integer id) {
        // 向数据库插入一条记录
        String sql = "insert into user (id,name,age) values (" + id + ",'fsx',21)";
        jdbcTemplate.update(sql);

        // 做其余的事情  可能抛出异常
        System.out.println(1 / 0);
        return "service hello";
    }
}

    @Transactional
    @Override
    public Object hello(Integer id) {
        // 向数据库插入一条记录
        String sql = "insert into user (id,name,age) values (" + id + ",'fsx',21)";
        jdbcTemplate.update(sql);


        // 根据id去查询获取 总数（若查询到了肯定是count=1）
        String query = "select count(1) from user where id = " + id;
        Integer count = jdbcTemplate.queryForObject(query, Integer.class);
        log.info(count.toString());

        return "service hello";
    }

    @Transactional
    @Override
    public Object hello(Integer id) {
        // 向数据库插入一条记录
        String sql = "insert into user (id,name,age) values (" + id + ",'fsx',21)";
        jdbcTemplate.update(sql);


        // 生产环境一般会把些操作交给线程池，此处我只是模拟一下效果而已~
        new Thread(() -> {
            String query = "select count(1) from user where id = " + id;
            Integer count = jdbcTemplate.queryForObject(query, Integer.class);
            log.info(count.toString());
        }).start();


        // 把问题放大
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return "service hello";
    }

@Slf4j
@Service
public class HelloServiceImpl implements HelloService {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Transactional
    @Override
    public Object hello(Integer id) {
        // 向数据库插入一条记录
        String sql = "insert into user (id,name,age) values (" + id + ",'fsx',21)";
        jdbcTemplate.update(sql);


        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
            // 在事务提交之后执行的代码块（方法）  此处使用TransactionSynchronizationAdapter，其实在Spring5后直接使用接口也很方便了~
            @Override
            public void afterCommit() {
                new Thread(() -> {
                    String query = "select count(1) from user where id = " + id;
                    Integer count = jdbcTemplate.queryForObject(query, Integer.class);
                    log.info(count.toString());
                }).start();
            }
        });


        // 把问题放大
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return "service hello";
    }

}

// @since 02.06.2003  它是个抽象类，但是没有任何子类  因为它所有的方法都是静态的
public abstract class TransactionSynchronizationManager {

	// ======保存着一大堆的ThreadLocal 这里就是它的核心存储======

	//  应用代码随事务的声明周期绑定的对象  比如：DataSourceTransactionManager有这么做：
	//TransactionSynchronizationManager.bindResource(obtainDataSource(), txObject.getConnectionHolder());
	// TransactionSynchronizationManager.bindResource(obtainDataSource(), suspendedResources);
	// 简单理解为当前线程的数据存储中心~~~~
	private static final ThreadLocal<Map<Object, Object>> resources = new NamedThreadLocal<>("Transactional resources");

	// 使用的同步器，用于应用扩展
	// 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");

	// 返回的是个只读视图
	public static Map<Object, Object> getResourceMap() {
		Map<Object, Object> map = resources.get();
		return (map != null ? Collections.unmodifiableMap(map) : Collections.emptyMap());
	}

	public static boolean hasResource(Object key) { ... }
	public static Object getResource(Object key) { ... }
	
	// actualKey：确定的key  拆包后的
	@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...
		// 如果ResourceHolder 被标记为了void空白了。此处直接从map里移除掉对应的key 
		// ~~~~~~~并且返回null~~~~~~~~~~~
		if (value instanceof ResourceHolder && ((ResourceHolder) value).isVoid()) {
			map.remove(actualKey);
			// Remove entire ThreadLocal if empty...
			if (map.isEmpty()) {
				resources.remove();
			}
			value = null;
		}
		return value;
	}

	// 逻辑很简单，就是和当前线程绑定一个Map，并且处理ResourceHolder 如果isVoid就抛错
	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 [" + Thread.currentThread().getName() + "]");
		}
	}

	public static Object unbindResource(Object key) throws IllegalStateException { ... }
	public static Object unbindResourceIfPossible(Object key) { ... }
	

	// 同步器是否是激活状态~~~  若是激活状态就可以执行同步器里的相关回调方法了
	public static boolean isSynchronizationActive() {
		return (synchronizations.get() != null);
	}

	// 如果事务已经开启了，就不能再初始化同步器了  而是直接注册
	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<>());
	}

	// 注册同步器TransactionSynchronization   这个非常重要 下面有详细介绍这个接口
	// 注册的时候要求当前线程的事务已经是激活状态的  而不是随便就可以调用的哦~~~
	public static void registerSynchronization(TransactionSynchronization synchronization) throws IllegalStateException {
		Assert.notNull(synchronization, "TransactionSynchronization must not be null");
		if (!isSynchronizationActive()) {
			throw new IllegalStateException("Transaction synchronization is not active");
		}
		synchronizations.get().add(synchronization);
	}


	// 返回的是只读视图  并且，并且支持AnnotationAwareOrderComparator.sort(sortedSynchs); 这样排序~~
	public static List<TransactionSynchronization> getSynchronizations() throws IllegalStateException { ... }
	public static void clearSynchronization() throws IllegalStateException { ... }

	... // 省略name等其余几个属性的get/set方法  因为没有任何逻辑
	// 这个方法列出来，应该下面会解释
	public static void setActualTransactionActive(boolean active) {
		actualTransactionActive.set(active ? Boolean.TRUE : null);
	}
	
	// 清楚所有和当前线程相关的（注意：此处只是clear清除，和当前线程的绑定而已~~~）
	public static void clear() {
		synchronizations.remove();
		currentTransactionName.remove();
		currentTransactionReadOnly.remove();
		currentTransactionIsolationLevel.remove();
		actualTransactionActive.remove();
	}
}

TransactionSynchronizationManager.setActualTransactionActive(status.hasTransaction()); // 相当于表示事务为开启了

TransactionSynchronizationManager.setCurrentTransactionReadOnly(false);
TransactionSynchronizationManager.setActualTransactionActive(false);

TransactionSynchronizationManager.setCurrentTransactionReadOnly(resourcesHolder.readOnly);
TransactionSynchronizationManager.setActualTransactionActive(resourcesHolder.wasActive);

// @since 02.06.2003  实现了java.io.Flushable接口
public interface TransactionSynchronization extends Flushable {

	int STATUS_COMMITTED = 0;
	int STATUS_ROLLED_BACK = 1;
	int STATUS_UNKNOWN = 2;

	// 事务赞提suspend的时候调用此方法
	// 实现这个方法的目的一般是释放掉绑定的resources 
	// TransactionSynchronizationManager#unbindResource
	default void suspend() {
	}
	// 事务恢复时候调用
	// TransactionSynchronizationManager#bindResource
	default void resume() {
	}

	// 将基础会话刷新到数据存储区（如果适用） 比如Hibernate/Jpa的session
	@Override
	default void flush() {
	}

	// 在事务提交之前促发。在AbstractPlatformTransactionManager.processCommit方法里 commit之前触发
	// 事务提交之前，比如flushing SQL statements to the database
	// 请注意：若此处发生了异常，会导致回滚~
	default void beforeCommit(boolean readOnly) {
	}
	// 在beforeCommit之后，在commit/rollback之前执行
	// 它和beforeCommit还有个非常大的区别是：即使beforeCommit抛出异常了  这个也会执行
	default void beforeCompletion() {
	}

	// 这个就非常重要了，它是事务提交（注意事务已经成功提交，数据库已经持久化完成这条数据了）后执行  注意此处是成功提交而没有异常
	// javadoc说了：此处一般可以发短信或者email等操作~~因为事务已经成功提交了
	
	// =====但是但是但是：======
	// 事务虽然已经提交，但事务资源（链接connection）可能仍然是活动的和可访问的。
	// 因此，此时触发的任何数据访问代码仍将“参与”原始事务 允许执行一些清理（不再执行提交操作！）
	// 除非它明确声明它需要在单独的事务中运行。
	default void afterCommit() {
	}

	// 和上面的区别在于：即使抛出异常回滚了  它也会执行的。它的notice同上
	default void afterCompletion(int status) {
	}
}

public abstract class AbstractPlatformTransactionManager implements PlatformTransactionManager, Serializable {
	...
	private void processCommit(DefaultTransactionStatus status) throws TransactionException {
		...
		try {
			prepareForCommit(status);
			triggerBeforeCommit(status);
			triggerBeforeCompletion(status);
			...
			doCommit(status);
			
			// 事务正常提交后  当然triggerAfterCompletion方法上面回滚里有而有个执行 此处不贴出了
			try {
				triggerAfterCommit(status);
			} finally {
				triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
			}
		} finally {
			cleanupAfterCompletion(status);
		}
	}
	...
	// 清楚、回收事务相关的资源~~~  并且恢复底层事务（若需要~）
	private void cleanupAfterCompletion(DefaultTransactionStatus status) {
		status.setCompleted();
		if (status.isNewSynchronization()) {
			TransactionSynchronizationManager.clear();
		}
		if (status.isNewTransaction()) {
			doCleanupAfterCompletion(status.getTransaction());
		}
		if (status.getSuspendedResources() != null) {
			if (status.isDebug()) {
				logger.debug("Resuming suspended transaction after completion of inner transaction");
			}
			Object transaction = (status.hasTransaction() ? status.getTransaction() : null);
			resume(transaction, (SuspendedResourcesHolder) status.getSuspendedResources());
		}
	}
}

public class JdbcTemplate extends JdbcAccessor implements JdbcOperations {
	...
	public <T> T execute(StatementCallback<T> action) throws DataAccessException {
		...
		// dataSource就是此JdbcTemplate所关联的数据源，这个在config配置文件里早就配置好了
		// 显然，这里获取的连接就是事务相关的，和当前想成绑定的connection
		Connection con = DataSourceUtils.getConnection(obtainDataSource());
		...
		finally {
			JdbcUtils.closeStatement(stmt);
			DataSourceUtils.releaseConnection(con, getDataSource());
		}
	}
	...
}

public abstract class TransactionSynchronizationAdapter implements TransactionSynchronization, Ordered {
	@Override
	public int getOrder() {
		return Ordered.LOWEST_PRECEDENCE;
	}
	... // 省略空实现们~~~
}

@Slf4j
@Service
public class HelloServiceImpl implements HelloService {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Transactional
    @Override
    public Object hello(Integer id) {
        // 向数据库插入一条记录
        String sql = "insert into user (id,name,age) values (" + id + ",'fsx',21)";
        jdbcTemplate.update(sql);

         TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
            // 在事务提交之后执行的代码块（方法）  此处使用TransactionSynchronizationAdapter，其实在Spring5后直接使用接口也很方便了~
            @Override
            public void afterCommit() {
                String sql = "insert into user (id,name,age) values (" + (id + 1) + ",'fsx',21)";
                int update = jdbcTemplate.update(sql);
                log.info(update + "");
            }
        });
        return "service hello";
    }

}

public abstract class AbstractPlatformTransactionManager implements PlatformTransactionManager, Serializable {
	...
	private void processCommit(DefaultTransactionStatus status) throws TransactionException {
		...
		try {
			prepareForCommit(status);
			triggerBeforeCommit(status);
			triggerBeforeCompletion(status);
			...
			doCommit(status);
			
			// 事务正常提交后  当然triggerAfterCompletion方法上面回滚里有而有个执行 此处不贴出了
			try {
				triggerAfterCommit(status);
			} finally {
				triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
			}
		} finally {
			cleanupAfterCompletion(status);
		}
	}
	...
	// 清楚、回收事务相关的资源~~~  并且恢复底层事务（若需要~）
	private void cleanupAfterCompletion(DefaultTransactionStatus status) {
		status.setCompleted();
		if (status.isNewSynchronization()) {
			TransactionSynchronizationManager.clear();
		}
		if (status.isNewTransaction()) {
			doCleanupAfterCompletion(status.getTransaction());
		}
		if (status.getSuspendedResources() != null) {
			if (status.isDebug()) {
				logger.debug("Resuming suspended transaction after completion of inner transaction");
			}
			Object transaction = (status.hasTransaction() ? status.getTransaction() : null);
			resume(transaction, (SuspendedResourcesHolder) status.getSuspendedResources());
		}
	}
}

		// 这里最终都会被执行~~~
		// doCleanupAfterCompletion方法在本抽象类是一个空的protected方法
		// 子类可以根据自己的需要，自己去实现事务提交完成后的操作
		if (status.isNewTransaction()) {
			doCleanupAfterCompletion(status.getTransaction());
		}

public class DataSourceTransactionManager extends AbstractPlatformTransactionManager
		implements ResourceTransactionManager, InitializingBean {
	...
	@Override
	protected void doCleanupAfterCompletion(Object transaction) {
		DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction;
		// 释放资源~~ Remove the connection holder from the thread, if exposed.
		if (txObject.isNewConnectionHolder()) {
			TransactionSynchronizationManager.unbindResource(obtainDataSource());
		}
		// Reset connection.
		Connection con = txObject.getConnectionHolder().getConnection();
		try {
			// 这里是关键，在事后会恢复链接的自动提交本能，也就是常用的恢复现场机制嘛~~
			// 显然这个和isMustRestoreAutoCommit属性的值有关，true就会恢复~~~
			if (txObject.isMustRestoreAutoCommit()) {
				con.setAutoCommit(true);
			}
			DataSourceUtils.resetConnectionAfterTransaction(con, txObject.getPreviousIsolationLevel());
		}
		...
		txObject.getConnectionHolder().clear();
	}
	...
}

private static class DataSourceTransactionObject extends JdbcTransactionObjectSupport {

		// 来自父类
		@Nullable
		private ConnectionHolder connectionHolder;
		@Nullable
		private Integer previousIsolationLevel;
		private boolean savepointAllowed = false;

		// 来自本类
		private boolean newConnectionHolder;
		private boolean mustRestoreAutoCommit; // 决定是否要恢复自动提交  默认情况下是false的
		...
		public void setMustRestoreAutoCommit(boolean mustRestoreAutoCommit) {
			this.mustRestoreAutoCommit = mustRestoreAutoCommit;
		}
		public boolean isMustRestoreAutoCommit() {
			return this.mustRestoreAutoCommit;
		}
}

	@Override
	protected void doBegin(Object transaction, TransactionDefinition definition) {
		...
		if (con.getAutoCommit()) {
			txObject.setMustRestoreAutoCommit(true); // 此处设置值为true
			if (logger.isDebugEnabled()) {
				logger.debug("Switching JDBC Connection [" + con + "] to manual commit");
			}
			con.setAutoCommit(false);
		}
		
	}