事务在关系型数据库中是一非常重要的东西，spring中对事务的处理也很简单，也就是加个注解的事。为更进一步理解它是如何实现的，今天我们就来拆解下。
1. 几个核心问题
1. 事务的核心是什么？
简单说其实就是底层数据库的使用方法，比如通常的sql begin; commit; rollback;... 各数据库各自的都会有自己的一套描述，并不完全一样。但总是因为db层面支持事务，在spring中才会支持事务。即spring只是上层应用。
2. 事务相关的代码是否必须写在一块？
就数据库sql操作而言，一般是要求写在一块，一起执行。比如: 开启事务 -> 执行业务1 -> n... -> 提交事务;但实际上可以看出，事务只要声明开始和结束，就可以做一起提交的效果。但是，开启事务后，中间所有的操作，就不能不受事务控制了。除非另外开一个子事务（这就有点复杂了）。那么，spring又如何去实现这种功能呢？
3. 如何将spring和具体的数据库事务关联起来？
比如，数据库有很多种，CRM框架也有很多，如何通过spring的事务框架统一管理起来呢？
4. 如何管理多种不同数据源的事务呢？
更复杂的事务是，我有多个数据源，又怎么去管理呢？能否做到部分事务管理呢？而非统一管理。这里可能更多的是一种api的使用方法了，咱们可以在需要使用时去翻阅官方文档。
2. 来个事务例子
spring中的事务操作很简单，比较常用的spring+mybatis方式，更是简单。只需要在需要实现事务的方法上添加注解即可。但该方法必须是独立的类的第一个方法，不允许是内部方法调用。这个问题的原因是事务需要用到切面完成，而方法内部调用无法走出类被代理而是直达方法。
@RestController @RequestMapping("/hello") @Slf4j public class HelloController { @Resource private UserService userService; // 请求入口 @GetMapping("/transInsert") @ResponseBody public Object transInsert(@RequestParam Long actorId, @RequestParam String firstName, @RequestParam String lastName, @RequestParam Long filmId) { // 事务入口 Object data = userService.insertActorAndFilmActor( actorId, firstName, lastName, filmId); return data; } } @Service @Slf4j public class UserService { @Resource private UserMapper userMapper; // 事务开启 @Transactional(transactionManager = "transactionManager") public Object insertActorAndFilmActor(Long actorId, String firstName, String lastName, Long filmId) { ActorEntity actorEntity = new ActorEntity(); actorEntity.setActorId(actorId); actorEntity.setFirstName(firstName); actorEntity.setLastName(lastName); // 业务操作1 Integer affect = userMapper.insertActor(actorEntity); actorId = actorEntity.getActorId(); FilmActorEntity filmActorEntity = new FilmActorEntity(); filmActorEntity.setActorId(actorId); filmActorEntity.setFilmId(filmId); // 业务操作2 affect += userMapper.insertFilmActor(filmActorEntity); // 退出后业务提交 return affect; } }
效果就是，要么两条数据都插入成功，要么都不成功。其中需要注意的是，springboot中需要开启事务，即在配置类中需要添加如下注解。
@EnableTransactionManagement
...
3. spring事务的实现过程
@Transactional 标识开启事务，当方法调用到这里之后，会先到 TransanctionInterceptor.invoke(); 然后就接入到了事务的管理过程了。我们可以从这里入手。
// org.springframework.transaction.interceptor.TransactionInterceptor#invoke @Override @Nullable public Object invoke(MethodInvocation invocation) throws Throwable { // Work out the target class: may be {@code null}. // The TransactionAttributeSource should be passed the target class // as well as the method, which may be from an interface. Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null); // Adapt to TransactionAspectSupport's invokeWithinTransaction... // 执行事务的核心框架 return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed); } // org.springframework.transaction.interceptor.TransactionAspectSupport#invokeWithinTransaction /** * General delegate for around-advice-based subclasses, delegating to several other template * methods on this class. Able to handle {@link CallbackPreferringPlatformTransactionManager} * as well as regular {@link PlatformTransactionManager} implementations and * {@link ReactiveTransactionManager} implementations for reactive return types. * @param method the Method being invoked * @param targetClass the target class that we're invoking the method on * @param invocation the callback to use for proceeding with the target invocation * @return the return value of the method, if any * @throws Throwable propagated from the target invocation */ @Nullable protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass, final InvocationCallback invocation) throws Throwable { // If the transaction attribute is null, the method is non-transactional. TransactionAttributeSource tas = getTransactionAttributeSource(); final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null); // 获取独立的事务管理器 final TransactionManager tm = determineTransactionManager(txAttr); if (this.reactiveAdapterRegistry != null && tm instanceof ReactiveTransactionManager) { ReactiveTransactionSupport txSupport = this.transactionSupportCache.computeIfAbsent(method, key -> { if (KotlinDetector.isKotlinType(method.getDeclaringClass()) && KotlinDelegate.isSuspend(method)) { throw new TransactionUsageException( "Unsupported annotated transaction on suspending function detected: " + method + ". Use TransactionalOperator.transactional extensions instead."); } ReactiveAdapter adapter = this.reactiveAdapterRegistry.getAdapter(method.getReturnType()); if (adapter == null) { throw new IllegalStateException("Cannot apply reactive transaction to non-reactive return type: " + method.getReturnType()); } return new ReactiveTransactionSupport(adapter); }); return txSupport.invokeWithinTransaction( method, targetClass, invocation, txAttr, (ReactiveTransactionManager) tm); } PlatformTransactionManager ptm = asPlatformTransactionManager(tm); final String joinpointIdentification = methodIdentification(method, targetClass, txAttr); if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) { // Standard transaction demarcation with getTransaction and commit/rollback calls. // 创建事务 TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification); Object retVal; try { // This is an around advice: Invoke the next interceptor in the chain. // This will normally result in a target object being invoked. // 调用业务方法 retVal = invocation.proceedWithInvocation(); } catch (Throwable ex) { // target invocation exception // 抛出异常，事务回滚 completeTransactionAfterThrowing(txInfo, ex); throw ex; } finally { cleanupTransactionInfo(txInfo); } if (retVal != null && vavrPresent && VavrDelegate.isVavrTry(retVal)) { // Set rollback-only in case of Vavr failure matching our rollback rules... TransactionStatus status = txInfo.getTransactionStatus(); if (status != null && txAttr != null) { retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status); } } // 提交事务 commitTransactionAfterReturning(txInfo); return retVal; } else { Object result; final ThrowableHolder throwableHolder = new ThrowableHolder(); // It's a CallbackPreferringPlatformTransactionManager: pass a TransactionCallback in. try { result = ((CallbackPreferringPlatformTransactionManager) ptm).execute(txAttr, status -> { TransactionInfo txInfo = prepareTransactionInfo(ptm, txAttr, joinpointIdentification, status); try { Object retVal = invocation.proceedWithInvocation(); if (retVal != null && vavrPresent && VavrDelegate.isVavrTry(retVal)) { // Set rollback-only in case of Vavr failure matching our rollback rules... retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status); } return retVal; } catch (Throwable ex) { if (txAttr.rollbackOn(ex)) { // A RuntimeException: will lead to a rollback. if (ex instanceof RuntimeException) { throw (RuntimeException) ex; } else { throw new ThrowableHolderException(ex); } } else { // A normal return value: will lead to a commit. throwableHolder.throwable = ex; return null; } } finally { cleanupTransactionInfo(txInfo); } }); } catch (ThrowableHolderException ex) { throw ex.getCause(); } catch (TransactionSystemException ex2) { if (throwableHolder.throwable != null) { logger.error("Application exception overridden by commit exception", throwableHolder.throwable); ex2.initApplicationException(throwableHolder.throwable); } throw ex2; } catch (Throwable ex2) { if (throwableHolder.throwable != null) { logger.error("Application exception overridden by commit exception", throwableHolder.throwable); } throw ex2; } // Check result state: It might indicate a Throwable to rethrow. if (throwableHolder.throwable != null) { throw throwableHolder.throwable; } return result; } }
以上就是整个事务的管理框架，先获取事务配置信息，然后开启事务，执行业务代码，最后进行提交或者回滚事务。清理现场。逻辑非常清晰易懂。
下面让我们更深入理解下，创建事务，提交事务，回滚事务的过程吧。
事务创建与绑定：
// 事务调入 // org.springframework.transaction.interceptor.TransactionAspectSupport#createTransactionIfNecessary /** * Create a transaction if necessary based on the given TransactionAttribute. * <p>Allows callers to perform custom TransactionAttribute lookups through * the TransactionAttributeSource. * @param txAttr the TransactionAttribute (may be {@code null}) * @param joinpointIdentification the fully qualified method name * (used for monitoring and logging purposes) * @return a TransactionInfo object, whether or not a transaction was created. * The {@code hasTransaction()} method on TransactionInfo can be used to * tell if there was a transaction created. * @see #getTransactionAttributeSource() */ @SuppressWarnings("serial") protected TransactionInfo createTransactionIfNecessary(@Nullable PlatformTransactionManager tm, @Nullable TransactionAttribute txAttr, final String joinpointIdentification) { // If no name specified, apply method identification as transaction name. if (txAttr != null && txAttr.getName() == null) { txAttr = new DelegatingTransactionAttribute(txAttr) { @Override public String getName() { return joinpointIdentification; } }; } TransactionStatus status = null; if (txAttr != null) { if (tm != null) { // 首次进入，事务开启 status = tm.getTransaction(txAttr); } else { if (logger.isDebugEnabled()) { logger.debug("Skipping transactional joinpoint [" + joinpointIdentification + "] because no transaction manager has been configured"); } } } return prepareTransactionInfo(tm, txAttr, joinpointIdentification, status); } // org.springframework.transaction.support.AbstractPlatformTransactionManager#getTransaction /** * This implementation handles propagation behavior. Delegates to * {@code doGetTransaction}, {@code isExistingTransaction} * and {@code doBegin}. * @see #doGetTransaction * @see #isExistingTransaction * @see #doBegin */ @Override public final TransactionStatus getTransaction(@Nullable TransactionDefinition definition) throws TransactionException { // Use defaults if no transaction definition given. TransactionDefinition def = (definition != null ? definition : TransactionDefinition.withDefaults()); Object transaction = doGetTransaction(); boolean debugEnabled = logger.isDebugEnabled(); if (isExistingTransaction(transaction)) { // Existing transaction found -> check propagation behavior to find out how to behave. return handleExistingTransaction(def, transaction, debugEnabled); } // Check definition settings for new transaction. if (def.getTimeout() < TransactionDefinition.TIMEOUT_DEFAULT) { throw new InvalidTimeoutException("Invalid transaction timeout", def.getTimeout()); } // No existing transaction found -> check propagation behavior to find out how to proceed. if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_MANDATORY) { throw new IllegalTransactionStateException( "No existing transaction found for transaction marked with propagation 'mandatory'"); } else if (def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRED || def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_REQUIRES_NEW || def.getPropagationBehavior() == TransactionDefinition.PROPAGATION_NESTED) { SuspendedResourcesHolder suspendedResources = suspend(null); if (debugEnabled) { logger.debug("Creating new transaction with name [" + def.getName() + "]: " + def); } try { // 事务开启 return startTransaction(def, transaction, debugEnabled, suspendedResources); } catch (RuntimeException | Error ex) { resume(null, suspendedResources); throw ex; } } else { // Create "empty" transaction: no actual transaction, but potentially synchronization. if (def.getIsolationLevel() != TransactionDefinition.ISOLATION_DEFAULT && logger.isWarnEnabled()) { logger.warn("Custom isolation level specified but no actual transaction initiated; " + "isolation level will effectively be ignored: " + def); } boolean newSynchronization = (getTransactionSynchronization() == SYNCHRONIZATION_ALWAYS); return prepareTransactionStatus(def, null, true, newSynchronization, debugEnabled, null); } } // org.springframework.transaction.support.AbstractPlatformTransactionManager#startTransaction /** * Start a new transaction. */ private TransactionStatus startTransaction(TransactionDefinition definition, Object transaction, boolean debugEnabled, @Nullable SuspendedResourcesHolder suspendedResources) { boolean newSynchronization = (getTransactionSynchronization() != SYNCHRONIZATION_NEVER); DefaultTransactionStatus status = newTransactionStatus( definition, transaction, true, newSynchronization, debugEnabled, suspendedResources); // 具体的事务处理 doBegin(transaction, definition); prepareSynchronization(status, definition); return status; } // org.springframework.jdbc.datasource.DataSourceTransactionManager#doBegin @Override protected void doBegin(Object transaction, TransactionDefinition definition) { DataSourceTransactionObject txObject = (DataSourceTransactionObject) transaction; Connection con = null; try { if (!txObject.hasConnectionHolder() || txObject.getConnectionHolder().isSynchronizedWithTransaction()) { // 初次获取事务的连接，后续需要复用此连接，才能形成事务效果 // 由具体的连接(池)组件决定获取连接的方式 Connection newCon = obtainDataSource().getConnection(); if (logger.isDebugEnabled()) { logger.debug("Acquired Connection [" + newCon + "] for JDBC transaction"); } txObject.setConnectionHolder(new ConnectionHolder(newCon), true); } txObject.getConnectionHolder().setSynchronizedWithTransaction(true); // 多次进入复用db连接 con = txObject.getConnectionHolder().getConnection(); Integer previousIsolationLevel = DataSourceUtils.prepareConnectionForTransaction(con, definition); txObject.setPreviousIsolationLevel(previousIsolationLevel); txObject.setReadOnly(definition.isReadOnly()); // Switch to manual commit if necessary. This is very expensive in some JDBC drivers, // so we don't want to do it unnecessarily (for example if we've explicitly // configured the connection pool to set it already). if (con.getAutoCommit()) { txObject.setMustRestoreAutoCommit(true); if (logger.isDebugEnabled()) { logger.debug("Switching JDBC Connection [" + con + "] to manual commit"); } // 事务开启即是设置 auto commit=0 con.setAutoCommit(false); } prepareTransactionalConnection(con, definition); // 设置开启事务标识，后续看到此标识将走事务逻辑 txObject.getConnectionHolder().setTransactionActive(true); int timeout = determineTimeout(definition); if (timeout != TransactionDefinition.TIMEOUT_DEFAULT) { txObject.getConnectionHolder().setTimeoutInSeconds(timeout); } // Bind the connection holder to the 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); } } // 具体mysql的实现 autocommit=0 // com.mysql.cj.jdbc.ConnectionImpl#setAutoCommit public void setAutoCommit(final boolean autoCommitFlag) throws SQLException { try { synchronized(this.getConnectionMutex()) { this.checkClosed(); if (this.connectionLifecycleInterceptors != null) { IterateBlock<ConnectionLifecycleInterceptor> iter = new IterateBlock<ConnectionLifecycleInterceptor>(this.connectionLifecycleInterceptors.iterator()) { void forEach(ConnectionLifecycleInterceptor each) throws SQLException { if (!each.setAutoCommit(autoCommitFlag)) { this.stopIterating = true; } } }; iter.doForAll(); if (!iter.fullIteration()) { return; } } if ((Boolean)this.autoReconnectForPools.getValue()) { this.autoReconnect.setValue(true); } try { boolean needsSetOnServer = true; if ((Boolean)this.useLocalSessionState.getValue() && this.session.getServerSession().isAutoCommit() == autoCommitFlag) { needsSetOnServer = false; } else if (!(Boolean)this.autoReconnect.getValue()) { needsSetOnServer = this.getSession().isSetNeededForAutoCommitMode(autoCommitFlag); } this.session.getServerSession().setAutoCommit(autoCommitFlag); if (needsSetOnServer) { // 真正的事务开始，即设置 autocommit=0 this.session.execSQL((Query)null, autoCommitFlag ? "SET autocommit=1" : "SET autocommit=0", -1, (NativePacketPayload)null, false, this.nullStatementResultSetFactory, (ColumnDefinition)null, false); } } finally { if ((Boolean)this.autoReconnectForPools.getValue()) { this.autoReconnect.setValue(false); } } } } catch (CJException var12) { throw SQLExceptionsMapping.translateException(var12, this.getExceptionInterceptor()); } }
可以看到，事务的开启有几个重要操作，一个是创建数据库连接，二是设置autocommit=0，三是将连接绑定到当前线程备用。基本上我们就可以推断出其实现原理了。接着往下看。
业务代码的执行：
事务开启之后，就进入了业务代码执行，然后才会有commit或rollback。实际上，业务代码往往是使用CRM框架来进行db操作的，我们此处使用的是 Mybatis 。我们看看其如何和spring进行绑定的。
实际上，mybatis会先获取一个会话session, 然后再获取db连接，最后执行sql语句。
// org.mybatis.spring.SqlSessionTemplate.SqlSessionInterceptor#invoke @Override public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { // 获取会话，建立数据源连接，事务内置 SqlSession sqlSession = getSqlSession(SqlSessionTemplate.this.sqlSessionFactory, SqlSessionTemplate.this.executorType, SqlSessionTemplate.this.exceptionTranslator); try { Object result = method.invoke(sqlSession, args); if (!isSqlSessionTransactional(sqlSession, SqlSessionTemplate.this.sqlSessionFactory)) { // force commit even on non-dirty sessions because some databases require // a commit/rollback before calling close() sqlSession.commit(true); } return result; } catch (Throwable t) { Throwable unwrapped = unwrapThrowable(t); if (SqlSessionTemplate.this.exceptionTranslator != null && unwrapped instanceof PersistenceException) { // release the connection to avoid a deadlock if the translator is no loaded. See issue #22 closeSqlSession(sqlSession, SqlSessionTemplate.this.sqlSessionFactory); sqlSession = null; Throwable translated = SqlSessionTemplate.this.exceptionTranslator .translateExceptionIfPossible((PersistenceException) unwrapped); if (translated != null) { unwrapped = translated; } } throw unwrapped; } finally { if (sqlSession != null) { closeSqlSession(sqlSession, SqlSessionTemplate.this.sqlSessionFactory); } } } } // org.mybatis.spring.SqlSessionUtils#getSqlSession /** * Gets an SqlSession from Spring Transaction Manager or creates a new one if needed. Tries to get a SqlSession out of * current transaction. If there is not any, it creates a new one. Then, it synchronizes the SqlSession with the * transaction if Spring TX is active and <code>SpringManagedTransactionFactory</code> is configured as a transaction * manager. * * @param sessionFactory * a MyBatis {@code SqlSessionFactory} to create new sessions * @param executorType * The executor type of the SqlSession to create * @param exceptionTranslator * Optional. Translates SqlSession.commit() exceptions to Spring exceptions. * @return an SqlSession managed by Spring Transaction Manager * @throws TransientDataAccessResourceException * if a transaction is active and the {@code SqlSessionFactory} is not using a * {@code SpringManagedTransactionFactory} * @see SpringManagedTransactionFactory */ public static SqlSession getSqlSession(SqlSessionFactory sessionFactory, ExecutorType executorType, PersistenceExceptionTranslator exceptionTranslator) { notNull(sessionFactory, NO_SQL_SESSION_FACTORY_SPECIFIED); notNull(executorType, NO_EXECUTOR_TYPE_SPECIFIED); // 首次进入将新创建一个 sqlSessionHolder SqlSessionHolder holder = (SqlSessionHolder) TransactionSynchronizationManager.getResource(sessionFactory); SqlSession session = sessionHolder(executorType, holder); if (session != null) { return session; } LOGGER.debug(() -> "Creating a new SqlSession"); session = sessionFactory.openSession(executorType); // 注册当前的 sessionHolder , 同时也会去拿取数据库连接，从而与spring 事务建立联系 registerSessionHolder(sessionFactory, executorType, exceptionTranslator, session); return session; } // org.mybatis.spring.SqlSessionUtils#registerSessionHolder /** * Register session holder if synchronization is active (i.e. a Spring TX is active). * * Note: The DataSource used by the Environment should be synchronized with the transaction either through * DataSourceTxMgr or another tx synchronization. Further assume that if an exception is thrown, whatever started the * transaction will handle closing / rolling back the Connection associated with the SqlSession. * * @param sessionFactory * sqlSessionFactory used for registration. * @param executorType * executorType used for registration. * @param exceptionTranslator * persistenceExceptionTranslator used for registration. * @param session * sqlSession used for registration. */ private static void registerSessionHolder(SqlSessionFactory sessionFactory, ExecutorType executorType, PersistenceExceptionTranslator exceptionTranslator, SqlSession session) { SqlSessionHolder holder; // 前面设置了事务激活标识，此处即可识别 if (TransactionSynchronizationManager.isSynchronizationActive()) { Environment environment = sessionFactory.getConfiguration().getEnvironment(); if (environment.getTransactionFactory() instanceof SpringManagedTransactionFactory) { LOGGER.debug(() -> "Registering transaction synchronization for SqlSession [" + session + "]"); holder = new SqlSessionHolder(session, executorType, exceptionTranslator); // 把当前sqlSessionFactory 绑定进去，避免后面反复查找性能损耗 TransactionSynchronizationManager.bindResource(sessionFactory, holder); TransactionSynchronizationManager .registerSynchronization(new SqlSessionSynchronization(holder, sessionFactory)); holder.setSynchronizedWithTransaction(true); holder.requested(); } else { if (TransactionSynchronizationManager.getResource(environment.getDataSource()) == null) { LOGGER.debug(() -> "SqlSession [" + session + "] was not registered for synchronization because DataSource is not transactional"); } else { throw new TransientDataAccessResourceException( "SqlSessionFactory must be using a SpringManagedTransactionFactory in order to use Spring transaction synchronization"); } } } else { LOGGER.debug(() -> "SqlSession [" + session + "] was not registered for synchronization because synchronization is not active"); } } // 执行语句之前会获取db连接, 再真正与spring建立绑定 // org.apache.ibatis.executor.SimpleExecutor#prepareStatement private Statement prepareStatement(StatementHandler handler, Log statementLog) throws SQLException { Statement stmt; Connection connection = getConnection(statementLog); stmt = handler.prepare(connection, transaction.getTimeout()); handler.parameterize(stmt); return stmt; } // org.apache.ibatis.executor.BaseExecutor#getConnection protected Connection getConnection(Log statementLog) throws SQLException { Connection connection = transaction.getConnection(); if (statementLog.isDebugEnabled()) { return ConnectionLogger.newInstance(connection, statementLog, queryStack); } else { return connection; } } // org.mybatis.spring.transaction.SpringManagedTransaction#getConnection /** * {@inheritDoc} */ @Override public Connection getConnection() throws SQLException { if (this.connection == null) { // 首次进入会打开连接，事务时会直接承接之前创建的连接，从而使事务生效 openConnection(); } return this.connection; } // org.mybatis.spring.transaction.SpringManagedTransaction#openConnection /** * Gets a connection from Spring transaction manager and discovers if this {@code Transaction} should manage * connection or let it to Spring. * <p> * It also reads autocommit setting because when using Spring Transaction MyBatis thinks that autocommit is always * false and will always call commit/rollback so we need to no-op that calls. */ private void openConnection() throws SQLException { this.connection = DataSourceUtils.getConnection(this.dataSource); this.autoCommit = this.connection.getAutoCommit(); this.isConnectionTransactional = DataSourceUtils.isConnectionTransactional(this.connection, this.dataSource); LOGGER.debug(() -> "JDBC Connection [" + this.connection + "] will" + (this.isConnectionTransactional ? " " : " not ") + "be managed by Spring"); } // org.springframework.jdbc.datasource.DataSourceUtils#getConnection /** * Obtain a Connection from the given DataSource. Translates SQLExceptions into * the Spring hierarchy of unchecked generic data access exceptions, simplifying * calling code and making any exception that is thrown more meaningful. * <p>Is aware of a corresponding Connection bound to the current thread, for example * when using {@link DataSourceTransactionManager}. Will bind a Connection to the * thread if transaction synchronization is active, e.g. when running within a * {@link org.springframework.transaction.jta.JtaTransactionManager JTA} transaction). * @param dataSource the DataSource to obtain Connections from * @return a JDBC Connection from the given DataSource * @throws org.springframework.jdbc.CannotGetJdbcConnectionException * if the attempt to get a Connection failed * @see #releaseConnection */ 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()); } } /** * Actually obtain a JDBC Connection from the given DataSource. * Same as {@link #getConnection}, but throwing the original SQLException. * <p>Is aware of a corresponding Connection bound to the current thread, for example * when using {@link DataSourceTransactionManager}. Will bind a Connection to the thread * if transaction synchronization is active (e.g. if in a JTA transaction). * <p>Directly accessed by {@link TransactionAwareDataSourceProxy}. * @param dataSource the DataSource to obtain Connections from * @return a JDBC Connection from the given DataSource * @throws SQLException if thrown by JDBC methods * @see #doReleaseConnection */ public static Connection doGetConnection(DataSource dataSource) throws SQLException { Assert.notNull(dataSource, "No DataSource specified"); // 一级缓存，使用dataSource绑定 // 二级缓存，使用 SpringManagedTransaction 保存 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)); } return conHolder.getConnection(); } // Else we either got no holder or an empty thread-bound holder here. logger.debug("Fetching JDBC Connection from DataSource"); Connection con = fetchConnection(dataSource); if (TransactionSynchronizationManager.isSynchronizationActive()) { try { // Use same Connection for further JDBC actions within the transaction. // Thread-bound object will get removed by synchronization at transaction completion. ConnectionHolder holderToUse = conHolder; if (holderToUse == null) { holderToUse = new ConnectionHolder(con); } else { holderToUse.setConnection(con); } holderToUse.requested(); TransactionSynchronizationManager.registerSynchronization( new ConnectionSynchronization(holderToUse, dataSource)); holderToUse.setSynchronizedWithTransaction(true); 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; }
很显然，Mybatis和spring是两个独立的东西，但是它们通过DataSource绑定在了一块，通过 DataSourceUtils 将数据库连接复用起来。事务开启时放入的连接，在mybatis使用时，将其取出进行复用。从而保证了操作是在一连接下进行，尽管代码层面看起来是完全分开的。但实际上和我们写一连串的sql是一样的效果。
事务提交:
业务代码执行成功后，就可以进行事务提交了，此时由spring统一进行处理。即获取到db连接，然后commit操作。
// org.springframework.transaction.interceptor.TransactionAspectSupport#commitTransactionAfterReturning /** * Execute after successful completion of call, but not after an exception was handled. * Do nothing if we didn't create a transaction. * @param txInfo information about the current transaction */ protected void commitTransactionAfterReturning(@Nullable TransactionInfo txInfo) { if (txInfo != null && txInfo.getTransactionStatus() != null) { if (logger.isTraceEnabled()) { logger.trace("Completing transaction for [" + txInfo.getJoinpointIdentification() + "]"); } txInfo.getTransactionManager().commit(txInfo.getTransactionStatus()); } } // org.springframework.transaction.support.AbstractPlatformTransactionManager#processCommit /** * Process an actual commit. * Rollback-only flags have already been checked and applied. * @param status object representing the transaction * @throws TransactionException in case of commit failure */ private void processCommit(DefaultTransactionStatus status) throws TransactionException { try { boolean beforeCompletionInvoked = false; try { boolean unexpectedRollback = false; prepareForCommit(status); triggerBeforeCommit(status); triggerBeforeCompletion(status); beforeCompletionInvoked = true; if (status.hasSavepoint()) { if (status.isDebug()) { logger.debug("Releasing transaction savepoint"); } unexpectedRollback = status.isGlobalRollbackOnly(); status.releaseHeldSavepoint(); } else if (status.isNewTransaction()) { if (status.isDebug()) { logger.debug("Initiating transaction commit"); } unexpectedRollback = status.isGlobalRollbackOnly(); doCommit(status); } else if (isFailEarlyOnGlobalRollbackOnly()) { unexpectedRollback = status.isGlobalRollbackOnly(); } // Throw UnexpectedRollbackException if we have a global rollback-only // marker but still didn't get a corresponding exception from commit. if (unexpectedRollback) { throw new UnexpectedRollbackException( "Transaction silently rolled back because it has been marked as rollback-only"); } } catch (UnexpectedRollbackException ex) { // can only be caused by doCommit triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK); throw ex; } catch (TransactionException ex) { // can only be caused by doCommit if (isRollbackOnCommitFailure()) { doRollbackOnCommitException(status, ex); } else { triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN); } throw ex; } catch (RuntimeException | Error ex) { if (!beforeCompletionInvoked) { triggerBeforeCompletion(status); } doRollbackOnCommitException(status, ex); throw ex; } // Trigger afterCommit callbacks, with an exception thrown there // propagated to callers but the transaction still considered as committed. try { triggerAfterCommit(status); } finally { triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED); } } finally { cleanupAfterCompletion(status); } } // org.springframework.jdbc.datasource.DataSourceTransactionManager#doCommit @Override protected void doCommit(DefaultTransactionStatus status) { DataSourceTransactionObject txObject = (DataSourceTransactionObject) status.getTransaction(); Connection con = txObject.getConnectionHolder().getConnection(); if (status.isDebug()) { logger.debug("Committing JDBC transaction on Connection [" + con + "]"); } try { con.commit(); } catch (SQLException ex) { throw new TransactionSystemException("Could not commit JDBC transaction", ex); } } // com.mysql.cj.jdbc.ConnectionImpl#commit public void commit() throws SQLException { try { synchronized(this.getConnectionMutex()) { this.checkClosed(); try { if (this.connectionLifecycleInterceptors != null) { IterateBlock<ConnectionLifecycleInterceptor> iter = new IterateBlock<ConnectionLifecycleInterceptor>(this.connectionLifecycleInterceptors.iterator()) { void forEach(ConnectionLifecycleInterceptor each) throws SQLException { if (!each.commit()) { this.stopIterating = true; } } }; iter.doForAll(); if (!iter.fullIteration()) { return; } } if (this.session.getServerSession().isAutoCommit()) { throw SQLError.createSQLException(Messages.getString("Connection.3"), this.getExceptionInterceptor()); } if ((Boolean)this.useLocalTransactionState.getValue() && !this.session.getServerSession().inTransactionOnServer()) { return; } // 执行 commit 语句 this.session.execSQL((Query)null, "commit", -1, (NativePacketPayload)null, false, this.nullStatementResultSetFactory, (ColumnDefinition)null, false); } catch (SQLException var10) { if ("08S01".equals(var10.getSQLState())) { throw SQLError.createSQLException(Messages.getString("Connection.4"), "08007", this.getExceptionInterceptor()); } throw var10; } finally { this.session.setNeedsPing((Boolean)this.reconnectAtTxEnd.getValue()); } } } catch (CJException var13) { throw SQLExceptionsMapping.translateException(var13, this.getExceptionInterceptor()); } }
事务回滚:
在发生异常后，需要将已经执行的代码进行回滚。否则就会形成数据不一致。
// org.springframework.transaction.interceptor.TransactionAspectSupport#completeTransactionAfterThrowing /** * Handle a throwable, completing the transaction. * We may commit or roll back, depending on the configuration. * @param txInfo information about the current transaction * @param ex throwable encountered */ protected void completeTransactionAfterThrowing(@Nullable TransactionInfo txInfo, Throwable ex) { if (txInfo != null && txInfo.getTransactionStatus() != null) { if (logger.isTraceEnabled()) { logger.trace("Completing transaction for [" + txInfo.getJoinpointIdentification() + "] after exception: " + ex); } if (txInfo.transactionAttribute != null && txInfo.transactionAttribute.rollbackOn(ex)) { try { // 回滚操作 txInfo.getTransactionManager().rollback(txInfo.getTransactionStatus()); } 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; } } else { // We don't roll back on this exception. // Will still roll back if TransactionStatus.isRollbackOnly() is true. try { txInfo.getTransactionManager().commit(txInfo.getTransactionStatus()); } catch (TransactionSystemException ex2) { logger.error("Application exception overridden by commit exception", ex); ex2.initApplicationException(ex); throw ex2; } catch (RuntimeException | Error ex2) { logger.error("Application exception overridden by commit exception", ex); throw ex2; } } } } // 核心回滚方法 // org.springframework.transaction.support.AbstractPlatformTransactionManager#processRollback /** * Process an actual rollback. * The completed flag has already been checked. * @param status object representing the transaction * @throws TransactionException in case of rollback failure */ private void processRollback(DefaultTransactionStatus status, boolean unexpected) { try { boolean unexpectedRollback = unexpected; try { triggerBeforeCompletion(status); if (status.hasSavepoint()) { if (status.isDebug()) { logger.debug("Rolling back transaction to savepoint"); } status.rollbackToHeldSavepoint(); } else if (status.isNewTransaction()) { if (status.isDebug()) { logger.debug("Initiating transaction rollback"); } doRollback(status); } else { // Participating in larger transaction if (status.hasTransaction()) { if (status.isLocalRollbackOnly() || isGlobalRollbackOnParticipationFailure()) { if (status.isDebug()) { logger.debug("Participating transaction failed - marking existing transaction as rollback-only"); } doSetRollbackOnly(status); } else { if (status.isDebug()) { logger.debug("Participating transaction failed - letting transaction originator decide on rollback"); } } } else { logger.debug("Should roll back transaction but cannot - no transaction available"); } // Unexpected rollback only matters here if we're asked to fail early if (!isFailEarlyOnGlobalRollbackOnly()) { unexpectedRollback = false; } } } catch (RuntimeException | Error ex) { triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN); throw ex; } triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK); // Raise UnexpectedRollbackException if we had a global rollback-only marker if (unexpectedRollback) { throw new UnexpectedRollbackException( "Transaction rolled back because it has been marked as rollback-only"); } } finally { cleanupAfterCompletion(status); } } // org.springframework.jdbc.datasource.DataSourceTransactionManager#doRollback @Override protected void doRollback(DefaultTransactionStatus status) { DataSourceTransactionObject txObject = (DataSourceTransactionObject) status.getTransaction(); Connection con = txObject.getConnectionHolder().getConnection(); if (status.isDebug()) { logger.debug("Rolling back JDBC transaction on Connection [" + con + "]"); } try { con.rollback(); } catch (SQLException ex) { throw new TransactionSystemException("Could not roll back JDBC transaction", ex); } } // com.mysql.cj.jdbc.ConnectionImpl#rollback() public void rollback() throws SQLException { try { synchronized(this.getConnectionMutex()) { this.checkClosed(); try { if (this.connectionLifecycleInterceptors != null) { IterateBlock<ConnectionLifecycleInterceptor> iter = new IterateBlock<ConnectionLifecycleInterceptor>(this.connectionLifecycleInterceptors.iterator()) { void forEach(ConnectionLifecycleInterceptor each) throws SQLException { if (!each.rollback()) { this.stopIterating = true; } } }; iter.doForAll(); if (!iter.fullIteration()) { return; } } if (this.session.getServerSession().isAutoCommit()) { throw SQLError.createSQLException(Messages.getString("Connection.20"), "08003", this.getExceptionInterceptor()); } else { try { // 使用 rollback 命令回滚数据 this.rollbackNoChecks(); } catch (SQLException var11) { if (!(Boolean)this.ignoreNonTxTables.getInitialValue() || var11.getErrorCode() != 1196) { throw var11; } } } } catch (SQLException var12) { if ("08S01".equals(var12.getSQLState())) { throw SQLError.createSQLException(Messages.getString("Connection.21"), "08007", this.getExceptionInterceptor()); } else { throw var12; } } finally { this.session.setNeedsPing((Boolean)this.reconnectAtTxEnd.getValue()); } } } catch (CJException var15) { throw SQLExceptionsMapping.translateException(var15, this.getExceptionInterceptor()); } } private void rollbackNoChecks() throws SQLException { try { synchronized(this.getConnectionMutex()) { if (!(Boolean)this.useLocalTransactionState.getValue() || this.session.getServerSession().inTransactionOnServer()) { this.session.execSQL((Query)null, "rollback", -1, (NativePacketPayload)null, false, this.nullStatementResultSetFactory, (ColumnDefinition)null, false); } } } catch (CJException var5) { throw SQLExceptionsMapping.translateException(var5, this.getExceptionInterceptor()); } }
至此，整个事务已操作完成。我们来总结下，整个事务的核心在于，创建一个切面，就业务代码包裹起来，然后，创建连接缓存起来，给业务代码复用。执行完成业务代码后，取回连接进行提交或回滚。然后清理现场。整个事务过程中大量使用了 ThreadLocal 技术进行变量共享。
// org.springframework.transaction.support.TransactionSynchronizationManager#clear /** * Clear the entire transaction synchronization state for the current thread: * registered synchronizations as well as the various transaction characteristics. * @see #clearSynchronization() * @see #setCurrentTransactionName * @see #setCurrentTransactionReadOnly * @see #setCurrentTransactionIsolationLevel * @see #setActualTransactionActive */ public static void clear() { // 此处使用大量的 ThreadLocal 进行线程通信，多方法复用变量 synchronizations.remove(); currentTransactionName.remove(); currentTransactionReadOnly.remove(); currentTransactionIsolationLevel.remove(); actualTransactionActive.remove(); }
4. 事务的初始化时机
实际上，事务只是一个注解，和其他很多的注解一样，会在bean初始的时候一起处理掉，比如此处使用 UserService, 在被化此bean的时候，就会将事务逻辑包裹注入其中。从而在调用时就会先调用事务，再调用业务代码。
// org.springframework.transaction.annotation.AnnotationTransactionAttributeSource#determineTransactionAttribute /** * Determine the transaction attribute for the given method or class. * <p>This implementation delegates to configured * {@link TransactionAnnotationParser TransactionAnnotationParsers} * for parsing known annotations into Spring's metadata attribute class. * Returns {@code null} if it's not transactional. * <p>Can be overridden to support custom annotations that carry transaction metadata. * @param element the annotated method or class * @return the configured transaction attribute, or {@code null} if none was found */ @Nullable protected TransactionAttribute determineTransactionAttribute(AnnotatedElement element) { for (TransactionAnnotationParser parser : this.annotationParsers) { TransactionAttribute attr = parser.parseTransactionAnnotation(element); if (attr != null) { return attr; } } return null; } // org.springframework.transaction.annotation.SpringTransactionAnnotationParser#parseTransactionAnnotation @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; } } // 详细参数配置解析 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; }
初始化完成bean之后，各个地方引用的都是被代理或被切面处理过的实例。所以事务方法可以在任何其他类引用的地方生效。多事务管理器通过 transactionManager 进行指定区分。实际上，它们仅仅是一种逻辑，即可以保证不同的事务管理器之间，使用不同的连接，从而达到多数据管理的效果。