Spring 多线程事务处理
作者：快乐非自愿限量之名
- 2024-03-12 福建
本文字数：20489 字
阅读完需：约 67 分钟
一、背景
本文主要介绍了 spring 多线程事务的解决方案，心急的小伙伴可以跳过上面的理论介绍分析部分直接看最终解决方案。
在我们日常的业务活动中，经常会出现大规模的修改插入操作，比如在 3.0 的活动赛事创建，涉及到十几张表的插入（一张表可能插入一行或者多行数据），由于单线程模型的关系，所有的 sql 都是串行，即后面的 sql 必须都要等到前面的 sql 执行完成才能继续。但是在很多场景下，sql 的执行顺序并不影响业务的结果，面对这样的场景，我们很自然的想到了使用异步的方式去处理，可是我们同时又希望整个创建操作是事务性的，即要全部成功，要么全部失败，但是单纯的使用异步线程并不能达到我们理想的效果。
这个时候，我们需要一种多线程下保证事务的解决方案。
代码片段，大量的同步保存操作
public void much(){ //业务操作1 doBusiness1(); //业务操作2 doBusiness2(); //业务操作3 doBusiness3(); //业务操作4 doBusiness4(); }
private void doBusiness1() { //执行sql1 //执行sql2 //执行sql3 //执行sql4 }复制代码
每个业务操作可以是相关联的，也有可能是完全无关的，但如果做成异步的话我们就无法保证事务，怎么去解决这个问题呢？
二、理论先行
1.事务介绍
我们先确定 spring 事务的本质是什么，spring 本身不支持事务，spring 实现事务只是对我们原有的业务逻辑做了一层包装，他替我们决定了什么时候开启事务，什么情况下应该向数据库提交，什么时候回滚，及实现我们设置的一些事务参数，包括回滚的条件，传播类型等。
我们所熟知的 spring 事务有两种主流的解决方式，一种是声明式事务，一种是编程式事务。
先来讲我们最常用的声明式事务。
1.1 声明式事务
声明式事务就是我们最常用的 @Transactional 注解，通常我们只需要在我们想交由 spring 控制的事务方法上加上注解即可，这个注解有一些重要的参数，由于不是本文重点，就不在此展开。这是一个经典的 spring 的 aop 实现，为了弄清楚在加上 @Transactional 注解后 spring 到底为我们做了什么，我们可以从两方面入手，一是 spring 如何给我们生成相应的代理对象，二是这个代理对象为我们做了什么。
事务的开始是由 @EnableTransactionManagement 注解产生，这个注解在运行时会导入 TransactionManagementConfigurationSelector 这个类，这个类本质上是一个 ImportSelector，他根据 adviceMode 将特定的配置类导入进去，分别为 AutoProxyRegistrar 后置处理器和 ProxyTransactionManagementConfiguration Advisor。
AutoProxyRegistrar 实现了 ImportBeanDefinitionRegistrar 重写了 registerBeanDefinitions 方法
public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) { boolean candidateFound = false; Set<String> annTypes = importingClassMetadata.getAnnotationTypes(); for (String annType : annTypes) { // ...
AopConfigUtils.registerAutoProxyCreatorIfNecessary(registry); } // ...}
@Nullablepublic static BeanDefinition registerAutoProxyCreatorIfNecessary( BeanDefinitionRegistry registry, @Nullable Object source) {
return registerOrEscalateApcAsRequired(InfrastructureAdvisorAutoProxyCreator.class, registry, source);}复制代码
该方法最终注入了 InfrastructureAdvisorAutoProxyCreator。InfrastructureAdvisorAutoProxyCreator 这个类就是一个 bean 的后置处理器，最终的作用就是处理需要的代理对象。
public Object postProcessAfterInitialization(@Nullable Object bean, String beanName) {if (bean != null) { Object cacheKey = getCacheKey(bean.getClass(), beanName); if (this.earlyProxyReferences.remove(cacheKey) != bean) { return wrapIfNecessary(bean, beanName, cacheKey); }}return bean;}
protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) { // ... // 拿当前bean去匹配容器中的 Advisors，如果找到符合的就生成代理对象 // Create proxy if we have advice. Object[] specificInterceptors = getAdvicesAndAdvisorsForBean(bean.getClass(), beanName, null); if (specificInterceptors != DO_NOT_PROXY) { this.advisedBeans.put(cacheKey, Boolean.TRUE); Object proxy = createProxy( bean.getClass(), beanName, specificInterceptors, new SingletonTargetSource(bean)); this.proxyTypes.put(cacheKey, proxy.getClass()); return proxy; }
this.advisedBeans.put(cacheKey, Boolean.FALSE); return bean;}复制代码
ProxyTransactionManagementConfiguration 的作用就是来生成具体的 Advisor，他注册了三个 bean，
该类主要完成以下几个任务：
创建 TransactionAttributeSource 对象：用于解析 @Transactional 注解并生成事务属性。
创建 TransactionInterceptor 对象：用于创建事务通知，将事务属性应用到目标方法，这其实就是一个事务模板，如下所示
protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass, final InvocationCallback invocation) throws Throwable { //TransactionAttributeSource内部保存着当前类某个方法对应的TransactionAttribute---事务属性源 //可以看做是一个存放TransactionAttribute与method方法映射的池子 TransactionAttributeSource tas = getTransactionAttributeSource(); //获取当前事务方法对应的TransactionAttribute final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null); //定位TransactionManager final TransactionManager tm = determineTransactionManager(txAttr); ..... //类型转换为局部事务管理器 PlatformTransactionManager ptm = asPlatformTransactionManager(tm); final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);
if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) { //TransactionManager根据TransactionAttribute创建事务后返回 //TransactionInfo封装了当前事务的信息--包括TransactionStatus TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);
Object retVal; try { //继续执行过滤器链---过滤链最终会调用目标方法 //因此可以理解为这里是调用目标方法 retVal = invocation.proceedWithInvocation(); } catch (Throwable ex) { //目标方法抛出异常则进行判断是否需要回滚 completeTransactionAfterThrowing(txInfo, ex); throw ex; } finally { //清除当前事务信息 cleanupTransactionInfo(txInfo); } ... //正常返回,那么就正常提交事务呗(当然还是需要判断TransactionStatus状态先) commitTransactionAfterReturning(txInfo); return retVal; } ...复制代码
创建 TransactionAdvisor 对象：将事务通知和切点（Pointcut）组合成 Advisor。
创建 TransactionAttributeSourceAdvisor 对象：将事务属性和切点组合成 Advisor
@Configuration(proxyBeanMethods = false)@Role(BeanDefinition.ROLE_INFRASTRUCTURE)public class ProxyTransactionManagementConfiguration extends AbstractTransactionManagementConfiguration {
@Bean(name = TransactionManagementConfigUtils.TRANSACTION_ADVISOR_BEAN_NAME) @Role(BeanDefinition.ROLE_INFRASTRUCTURE) public BeanFactoryTransactionAttributeSourceAdvisor transactionAdvisor( TransactionAttributeSource transactionAttributeSource, TransactionInterceptor transactionInterceptor) {
BeanFactoryTransactionAttributeSourceAdvisor advisor = new BeanFactoryTransactionAttributeSourceAdvisor(); advisor.setTransactionAttributeSource(transactionAttributeSource); advisor.setAdvice(transactionInterceptor); if (this.enableTx != null) { advisor.setOrder(this.enableTx.<Integer>getNumber("order")); } return advisor; }
@Bean @Role(BeanDefinition.ROLE_INFRASTRUCTURE) public TransactionAttributeSource transactionAttributeSource() { // TransactionAttributeSource 是一个接口，具体注入的是 Annotationxxxx return new AnnotationTransactionAttributeSource(); }
@Bean @Role(BeanDefinition.ROLE_INFRASTRUCTURE) public TransactionInterceptor transactionInterceptor(TransactionAttributeSource transactionAttributeSource) { TransactionInterceptor interceptor = new TransactionInterceptor(); interceptor.setTransactionAttributeSource(transactionAttributeSource); if (this.txManager != null) { interceptor.setTransactionManager(this.txManager); } return interceptor; }}复制代码
@Nullableprivate TransactionAttributeSource transactionAttributeSource;
private final TransactionAttributeSourcePointcut pointcut = new TransactionAttributeSourcePointcut() { @Override @Nullable protected TransactionAttributeSource getTransactionAttributeSource() { return transactionAttributeSource; }};
/** * Set the transaction attribute source which is used to find transaction * attributes. This should usually be identical to the source reference * set on the transaction interceptor itself. * @see TransactionInterceptor#setTransactionAttributeSource */public void setTransactionAttributeSource(TransactionAttributeSource transactionAttributeSource) {this.transactionAttributeSource = transactionAttributeSource;}
/** * Set the {@link ClassFilter} to use for this pointcut. * Default is {@link ClassFilter#TRUE}. */public void setClassFilter(ClassFilter classFilter) { this.pointcut.setClassFilter(classFilter);}
@Overridepublic Pointcut getPointcut() { return this.pointcut;}复制代码
可以见到里面已经包含了 pointcut，这就能将我们需要被增加的事务方法找出。
ProxyTransactionManagementConfiguration 负责将需要包装的 bean 和方法找出并包装成 advisor，InfrastructureAdvisorAutoProxyCreator 根据 advisor 生成相应的代理对象。
小结：InfrastructureAdvisorAutoProxyCreator 遍历容器中的 bean，尝试去自动代理，匹配的工作就交由 advisor 中的 point，如果匹配成功就为其创建代理对象，这个代理对象中放入了 TransactionInterceptor 拦截器
，等到相关方法调用时，调用的是代理对象的方法，然后通过责任链模式通过 TransactionInterceptor 处理，以此来进行事务的操作。
声明式事务的介绍先到这里，接下来我们来介绍下编程式事务。
1.2 编程式事务
编程式事务的核心就是将 spring 为我们做好的那些步骤拆出来，交由开发者去控制事务何时开启、提交、回滚，他的运行本质和声明式事务并没有两样。
模板如下
public class TransactionMain { public static void main(String[] args) throws ClassNotFoundException, SQLException { test(); } private static void test() { DataSource dataSource = getDS(); JdbcTransactionManager jtm = new JdbcTransactionManager(dataSource); //JdbcTransactionManager根据TransactionDefinition信息来进行一些连接属性的设置 //包括隔离级别和传播行为等 DefaultTransactionDefinition transactionDef = new DefaultTransactionDefinition(); //开启一个新事务---此时autocommit已经被设置为了false,并且当前没有事务,这里创建的是一个新事务 TransactionStatus ts = jtm.getTransaction(transactionDef); //进行业务逻辑操作 try { update(dataSource); jtm.commit(ts); }catch (Exception e){ jtm.rollback(ts); System.out.println("发生异常,我已回滚"); } } private static void update(DataSource dataSource) throws Exception { JdbcTemplate jt = new JdbcTemplate(); jt.setDataSource(dataSource); jt.update("UPDATE Department SET Dname=\"大忽悠\" WHERE id=6"); throw new Exception("我是来捣乱的"); } }复制代码
三、方案探索
1.直接使用多线程
我们在开启代码中事务，并在业务逻辑中直接使用多线程，是否能保证事务？
@Transactional public void testDirect() { new Thread(()->{ Per per = new Per(); per.setName("t1"); perService.save(per); }).start(); new Thread(()->{ Per per1 = new Per(); per1.setName("t2"); perService.save(per1); throw new RuntimeException("Exception test"); }).start(); }复制代码
显然，这种方式并不能保证事务，哪怕加上了事务注解，因为子线程抛出的异常并不能在主线程中捕获，也不能被其他线程感知到。
2.事务模板中使用多线程
package com.user.util; import lombok.RequiredArgsConstructor;import org.springframework.jdbc.datasource.DataSourceTransactionManager;import org.springframework.stereotype.Component;import org.springframework.transaction.TransactionStatus;import org.springframework.transaction.support.DefaultTransactionDefinition; import javax.sql.DataSource;import java.util.ArrayList;import java.util.List;import java.util.concurrent.CompletableFuture;import java.util.concurrent.ExecutionException;import java.util.concurrent.Executor;import java.util.concurrent.atomic.AtomicBoolean;
@Component @RequiredArgsConstructor public class MultiplyThreadTransactionManager { /** * 如果是多数据源的情况下,需要指定具体是哪一个数据源 */ private final DataSource dataSource; public void runAsyncButWaitUntilAllDown(List<Runnable> tasks, Executor executor) { if(executor==null){ throw new IllegalArgumentException("线程池不能为空"); } DataSourceTransactionManager transactionManager = getTransactionManager(); //是否发生了异常 AtomicBoolean ex=new AtomicBoolean(); List<CompletableFuture> taskFutureList=new ArrayList<>(tasks.size()); List<TransactionStatus> transactionStatusList=new ArrayList<>(tasks.size()); tasks.forEach(task->{ taskFutureList.add(CompletableFuture.runAsync( () -> { try{ //1.开启新事务 transactionStatusList.add(openNewTransaction(transactionManager)); //2.异步任务执行 task.run(); }catch (Throwable throwable){ //打印异常 throwable.printStackTrace(); //其中某个异步任务执行出现了异常,进行标记 ex.set(Boolean.TRUE); //其他任务还没执行的不需要执行了 taskFutureList.forEach(completableFuture -> completableFuture.cancel(true)); } } , executor) ); }); try { //阻塞直到所有任务全部执行结束---如果有任务被取消,这里会抛出异常滴,需要捕获 CompletableFuture.allOf(taskFutureList.toArray(new CompletableFuture[]{})).get(); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } //发生了异常则进行回滚操作,否则提交 if(ex.get()){ System.out.println("发生异常,全部事务回滚"); transactionStatusList.forEach(transactionManager::rollback); }else { System.out.println("全部事务正常提交"); transactionStatusList.forEach(transactionManager::commit); } } private TransactionStatus openNewTransaction(DataSourceTransactionManager transactionManager) { //JdbcTransactionManager根据TransactionDefinition信息来进行一些连接属性的设置 //包括隔离级别和传播行为等 DefaultTransactionDefinition transactionDef = new DefaultTransactionDefinition(); //开启一个新事务---此时autocommit已经被设置为了false,并且当前没有事务,这里创建的是一个新事务 return transactionManager.getTransaction(transactionDef); } private DataSourceTransactionManager getTransactionManager() { return new DataSourceTransactionManager(dataSource); } }复制代码
测试
public void test(){ List<Runnable> tasks=new ArrayList<>(); tasks.add(()->{ Per per = new Per(); per.setName("t1"); perService.save(per); }); tasks.add(()->{ Per per = new Per(); per.setName("t2"); perService.save(per); }); multiplyThreadTransactionManager.runAsyncButWaitUntilAllDown(tasks, Executors.newCachedThreadPool()); }复制代码
执行结果
java.lang.IllegalStateException: No value for key [HikariDataSource (HikariPool-1)] bound to thread at org.springframework.transaction.support.TransactionSynchronizationManager.unbindResource(TransactionSynchronizationManager.java:198) ~[spring-tx-5.3.10.jar:5.3.10] at org.springframework.jdbc.datasource.DataSourceTransactionManager.doCleanupAfterCompletion(DataSourceTransactionManager.java:371) ~[spring-jdbc-5.3.10.jar:5.3.10] at org.springframework.transaction.support.AbstractPlatformTransactionManager.cleanupAfterCompletion(AbstractPlatformTransactionManager.java:992) ~[spring-tx-5.3.10.jar:5.3.10] at org.springframework.transaction.suppoAbstractPlatformTransactionrt.AbstractPlatformTransactionManager.processCommit(AbstractPlatformTransactionManager复制代码
结果报了这个错，这个错误信息室找不到绑定在线程上的 key 为 HikariDataSource 的资源，因为事务资源都是绑定在线程上的，当事务提交或者回滚时，他需要寻找绑定在当前线程上的资源，如果找不到，就会报错。
原理剖析：
首先我们找到绑定线程资源的关键方法 org.springframework.transaction.support.TransactionSynchronizationManager#bindResource
/** * Bind the given resource for the given key to the current thread. * @param key the key to bind the value to (usually the resource factory) * @param value the value to bind (usually the active resource object) * @throws IllegalStateException if there is already a value bound to the thread * @see ResourceTransactionManager#getResourceFactory() */ 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"); } }复制代码
根据 debug 会发现，spring 在开启事务时会自动为我们调用这个方法，绑定 key 为 HikariDataSource，value 为 ConnectionHolder 到 threadlocal 中。第二次 sql 执行时会绑定 key 为 DefaultSqlSessionFactory，value 为 DefaultSqlSessionFactory。
既然讲到了事务资源的绑定时机，下面就顺便讲一下这两种资源在何时释放。我们再回顾一下事务的执行流程及机制。spring 处理事务的原理就是基于 aop，每个需要实现事务的方法都要通过 TransactionInterceptor 这个拦截器，通过这个拦截器去实现事务增强。
@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, new CoroutinesInvocationCallback() { @Override @Nullable public Object proceedWithInvocation() throws Throwable { //原始方法 return invocation.proceed(); } @Override public Object getTarget() { return invocation.getThis(); } @Override public Object[] getArguments() { return invocation.getArguments(); } }); }复制代码
他最终会交给他的父类模板类 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);
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; }
... }复制代码
继续往下，观察 commitTransactionAfterReturning(txInfo)的实现，发现这一步是由事务管理器完成的。
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()); } }复制代码
进而可以推断，rollback 操作也是同样的道理，有兴趣的小伙伴可以自己 debug 一下，继续走下去，观察事务管理器为我们做了什么。
@Override public final void commit(TransactionStatus status) throws TransactionException { if (status.isCompleted()) { throw new IllegalTransactionStateException( "Transaction is already completed - do not call commit or rollback more than once per transaction"); }
DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status; if (defStatus.isLocalRollbackOnly()) { if (defStatus.isDebug()) { logger.debug("Transactional code has requested rollback"); } processRollback(defStatus, false); return; }
if (!shouldCommitOnGlobalRollbackOnly() && defStatus.isGlobalRollbackOnly()) { if (defStatus.isDebug()) { logger.debug("Global transaction is marked as rollback-only but transactional code requested commit"); } processRollback(defStatus, true); return; }
processCommit(defStatus); }复制代码
到此，spring 的整个事务流程就已经非常清晰了，我们想要实现多事务管理的方法也找到了，难就是去控制事务的资源。只要我们拿到了相应的事务资源，然后在创建自己的事务管理器控制事务何时提交或者回滚，这样我们就可以实现一个多线程同时提交回滚，类似于二阶段提交的操作，来达到多线程事务的统一。
3.多线程事务管理器
不多说，直接上代码看最终版本
package com.controller;
import lombok.Builder;import lombok.RequiredArgsConstructor;import lombok.extern.slf4j.Slf4j;import org.slf4j.MDC;import org.springframework.jdbc.datasource.DataSourceTransactionManager;import org.springframework.stereotype.Component;import org.springframework.transaction.TransactionStatus;import org.springframework.transaction.support.DefaultTransactionDefinition;import org.springframework.transaction.support.TransactionSynchronization;import org.springframework.transaction.support.TransactionSynchronizationManager;import org.springframework.util.CollectionUtils;
import javax.sql.DataSource;import java.util.*;import java.util.concurrent.*;import java.util.concurrent.atomic.AtomicBoolean;import java.util.concurrent.atomic.AtomicInteger;
/** * 多线程事务管理 */@Component@Slf4j@RequiredArgsConstructorpublic class MultiplyThreadTransactionManager { /** * 如果是多数据源的情况下,需要指定具体是哪一个数据源 */ private final DataSource dataSource;
private final static ThreadLocal<Boolean> immediatelyCommitFlag = new ThreadLocal<>();
private final static ThreadLocal<List<TransactionStatus>> transactionStatusListThreadLocal = new ThreadLocal<>();
private final static ThreadLocal<List<TransactionResource>> transactionResourcesthreadLocal = new ThreadLocal<>();
private final static ThreadLocal<Map<Object, Object>> mainNativeResourceThreadLocal = new ThreadLocal<>();
/** * 多线程下事务执行 * * @param tasks 任务列表 * @param immediatelyCommit 是否需要立即提交 */ public List<CompletableFuture> runAsyncButWaitUntilAllDown(List<Runnable> tasks, Boolean immediatelyCommit) {
Executor executor = Executors.newCachedThreadPool();
DataSourceTransactionManager transactionManager = getTransactionManager(); //是否发生了异常 AtomicBoolean ex = new AtomicBoolean();
List<CompletableFuture> taskFutureList = new CopyOnWriteArrayList<>(); List<TransactionStatus> transactionStatusList = new CopyOnWriteArrayList<>(); List<TransactionResource> transactionResources = new CopyOnWriteArrayList<>();
//记录原生主事务资源 //这一步可能在原生sql执行前，也可能在原生sql执行后，所以这个资源可能不够充分，需要在下面继续处理 //如果返回的是原资源集合的引用，下面一步可以不用 Map<Object, Object> resourceMap = TransactionSynchronizationManager.getResourceMap(); if (!CollectionUtils.isEmpty(resourceMap)) { mainNativeResourceThreadLocal.set(new HashMap<>(resourceMap)); } Map<String, String> copyOfContextMap = MDC.getCopyOfContextMap(); Executor finalExecutor = executor; AtomicInteger atomicInteger = new AtomicInteger(0); tasks.forEach(task -> { taskFutureList.add(CompletableFuture.runAsync( () -> { log.info("任务开始"); try { //1.开启新事务 TransactionStatus transactionStatus = openNewTransaction(transactionManager); log.info("开启新事务 successfully"); transactionStatusList.add(transactionStatus); atomicInteger.incrementAndGet(); System.out.println("atomicInteger.get()"+atomicInteger.incrementAndGet()); System.out.println(transactionStatus); //2.异步任务执行 task.run(); log.info("异步任务执行 successfully"); //3.继续事务资源复制,因为在sql执行是会产生新的资源对象 transactionResources.add(TransactionResource.copyTransactionResource());
} catch (Throwable throwable) { log.info("任务执行异常"+throwable.getMessage()); log.error("任务执行异常",throwable); //其中某个异步任务执行出现了异常,进行标记 ex.set(Boolean.TRUE); //其他任务还没执行的不需要执行了 taskFutureList.forEach(completableFuture -> completableFuture.cancel(true)); } } , finalExecutor) ); });
try { //阻塞直到所有任务全部执行结束---如果有任务被取消,这里会抛出异常滴,需要捕获 CompletableFuture.allOf(taskFutureList.toArray(new CompletableFuture[]{})).get(); } catch (InterruptedException | ExecutionException e) { log.info("任务被取消"); log.error("任务被取消",e);
}
//发生了异常则进行回滚操作,否则提交 if (ex.get()) { log.info("发生异常,全部事务回滚"); for (int i = 0; i < transactionStatusList.size(); i++) { transactionResources.get(i).autoWiredTransactionResource(); Map<Object, Object> rollBackResourceMap = TransactionSynchronizationManager.getResourceMap(); log.info("回滚前事务资源size{}，本身{}",rollBackResourceMap.size(),rollBackResourceMap); transactionManager.rollback(transactionStatusList.get(i)); transactionResources.get(i).removeTransactionResource(); } } else { if (immediatelyCommit) { log.info("全部事务正常提交"); for (int i = 0; i < transactionStatusList.size(); i++) { //transactionResources.get(i).autoWiredTransactionResource(); Map<Object, Object> commitResourceMap = TransactionSynchronizationManager.getResourceMap(); log.info("提交前事务资源size{}，本身{}",commitResourceMap.size(),commitResourceMap); transactionManager.commit(transactionStatusList.get(i)); transactionResources.get(i).removeTransactionResource(); } } else { //缓存全部待提交数据 immediatelyCommitFlag.set(immediatelyCommit); transactionResourcesthreadLocal.set(transactionResources); transactionStatusListThreadLocal.set(transactionStatusList); } }
//交还给主事务 if (immediatelyCommit) { mainTransactionResourceBack(!ex.get()); }
return taskFutureList; }
public void multiplyThreadTransactionCommit() { try { Boolean immediatelyCommit = immediatelyCommitFlag.get(); if (immediatelyCommit) { throw new IllegalStateException("immediatelyCommit cant call multiplyThreadTransactionCommit"); } //提交 //获取存储的事务资源和状态 List<TransactionResource> transactionResources = transactionResourcesthreadLocal.get(); List<TransactionStatus> transactionStatusList = transactionStatusListThreadLocal.get(); if (CollectionUtils.isEmpty(transactionResources) || CollectionUtils.isEmpty(transactionStatusList)) { throw new IllegalStateException("transactionResources or transactionStatusList is empty"); } //重新提交 DataSourceTransactionManager transactionManager = getTransactionManager(); log.info("全部事务正常提交"); for (int i = 0; i < transactionStatusList.size(); i++) { transactionResources.get(i).autoWiredTransactionResource(); Map<Object, Object> commitResourceMap = TransactionSynchronizationManager.getResourceMap(); log.info("提交前事务资源size{}，本身{}",commitResourceMap.size(),commitResourceMap); transactionManager.commit(transactionStatusList.get(i)); transactionResources.get(i).removeTransactionResource(); } } catch (Exception e) { mainTransactionResourceBack(false); log.error("multiplyThreadTransactionCommit fail", e); } finally { transactionResourcesthreadLocal.remove(); transactionStatusListThreadLocal.remove(); immediatelyCommitFlag.remove(); }
//交还给主事务 mainTransactionResourceBack(true); }
//主线程事务资源返还 public void mainTransactionResourceBack(Boolean subTransactionSuccess) {
if (CollectionUtils.isEmpty(mainNativeResourceThreadLocal.get())) { //清除数据 mainNativeResourceThreadLocal.remove(); return; } Map<Object, Object> nativeResource = new HashMap<>(mainNativeResourceThreadLocal.get()); Map<Object, Object> resourceMap = TransactionSynchronizationManager.getResourceMap(); log.info("当前线程资事务源size{}--------------------------------{}",resourceMap.size(), resourceMap); log.info("原生线程事务资源size{}--------------------------------{}",nativeResource.size(), nativeResource); //已经被绑定的资源不能重复绑定 if (!CollectionUtils.isEmpty(resourceMap)) { for (Object o : resourceMap.keySet()) { if (nativeResource.containsKey(o)) { nativeResource.remove(o); } } }
nativeResource.forEach((k,v)->{ if (!(k instanceof DataSource)){ log.info("nativeResource 没有 DataSource"); } }); //交还不能绑定factory nativeResource.forEach((k,v)->{ if (k instanceof DataSource){ TransactionSynchronizationManager.bindResource(k,v); } }); Map<Object, Object> finResource = TransactionSynchronizationManager.getResourceMap(); log.info("主线程最终事务源size{}--------------------------------{}",finResource.size(), finResource); //防止未激活事务 if (!TransactionSynchronizationManager.isSynchronizationActive()) { TransactionSynchronizationManager.initSynchronization(); } //清除数据 mainNativeResourceThreadLocal.remove(); if (!subTransactionSuccess) { throw new RuntimeException("子事务失败，需要回滚"); } }
private TransactionStatus openNewTransaction(DataSourceTransactionManager transactionManager) { //JdbcTransactionManager根据TransactionDefinition信息来进行一些连接属性的设置 //包括隔离级别和传播行为等 DefaultTransactionDefinition transactionDef = new DefaultTransactionDefinition(); //开启一个新事务---此时autocommit已经被设置为了false,并且当前没有事务,这里创建的是一个新事务 return transactionManager.getTransaction(transactionDef); }
private DataSourceTransactionManager getTransactionManager() { return new DataSourceTransactionManager(dataSource); }
/** * 保存当前事务资源,用于线程间的事务资源COPY操作 */ @Builder private static class TransactionResource { //事务结束后默认会移除集合中的DataSource作为key关联的资源记录 private Map<Object, Object> resources = new HashMap<>();
//下面五个属性会在事务结束后被自动清理,无需我们手动清理 private Set<TransactionSynchronization> synchronizations = new HashSet<>();
private String currentTransactionName;
private Boolean currentTransactionReadOnly;
private Integer currentTransactionIsolationLevel;
private Boolean actualTransactionActive;
public static TransactionResource copyTransactionResource() { return TransactionResource.builder() //返回的是不可变集合，这里为了更加灵活，copy出一个集合过来 .resources(new HashMap<>(TransactionSynchronizationManager.getResourceMap())) //如果需要注册事务监听者,这里记得修改--我们这里不需要,就采用默认负责--spring事务内部默认也是这个值 .synchronizations(new LinkedHashSet<>()) .currentTransactionName(TransactionSynchronizationManager.getCurrentTransactionName()) .currentTransactionReadOnly(TransactionSynchronizationManager.isCurrentTransactionReadOnly()) .currentTransactionIsolationLevel(TransactionSynchronizationManager.getCurrentTransactionIsolationLevel()) .actualTransactionActive(TransactionSynchronizationManager.isActualTransactionActive()) .build(); }
//装配事务资源，为提交/回滚做储备 public void autoWiredTransactionResource() { //获取当前线程事务资源 Map<Object, Object> resourceMap = TransactionSynchronizationManager.getResourceMap(); for (Object o : resourceMap.keySet()) { if (resourceMap.containsKey(o)) { //移除重复事务资源key，避免绑定报错 resources.remove(o); }
} boolean synchronizationActive = TransactionSynchronizationManager.isSynchronizationActive(); //绑定事务资源，注意 绑定是绑定到当前主线程上，记得最后释放交换主线程，再由主线程收回原有事务自选 resources.forEach(TransactionSynchronizationManager::bindResource); //如果需要注册事务监听者,这里记得修改--我们这里不需要,就采用默认负责--spring事务内部默认也是这个值 //避免重复激活或者事务未激活 if (!synchronizationActive) { TransactionSynchronizationManager.initSynchronization(); }
TransactionSynchronizationManager.setActualTransactionActive(actualTransactionActive); TransactionSynchronizationManager.setCurrentTransactionName(currentTransactionName); TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(currentTransactionIsolationLevel); TransactionSynchronizationManager.setCurrentTransactionReadOnly(currentTransactionReadOnly); }
public void removeTransactionResource() { Map<Object, Object> resourceMap = new HashMap<>(TransactionSynchronizationManager.getResourceMap());
//事务结束后默认会移除集合中的DataSource作为key关联的资源记录 //DataSource如果重复移除,unbindResource时会因为不存在此key关联的事务资源而报错 resources.keySet().forEach(key -> { if (resourceMap.containsKey(key)) { TransactionSynchronizationManager.unbindResource(key); } }); } }}复制代码
验证
@Transactional public String test(Integer par) { log.info("get(" + par + ")"); if (par == 3 || par == 5 || par == 6) { Per per2 = new Per(); per2.setName("t3"); per2.setGrou(Thread.currentThread().getName()); perService.save(per2); }
List<Runnable> list = new ArrayList<>(); list.add(() -> { Per per = new Per(); per.setName("t1"); per.setGrou(Thread.currentThread().getName()); log.info("任务开始save"); perService.save(per); log.info("任务完成save"); if (par == 1) { throw new RuntimeException(); } }); list.add(() -> { Per per1 = new Per(); per1.setName("t2"); per1.setGrou(Thread.currentThread().getName()); log.info("任务开始save"); perService.save(per1); log.info("任务完成save"); if (par == 2) { throw new RuntimeException(); }
});
log.info("runAsyncButWaitUntilAllDown start"); multiplyThreadTransactionManager.runAsyncButWaitUntilAllDown(list, false);
if (par == 4 || par == 5 || par == 6) { Per per3 = new Per(); per3.setName("t4"); per3.setGrou(Thread.currentThread().getName()); perService.save(per3); if (par == 6) { throw new RuntimeException(); } } log.info("multiplyThreadTransactionCommit start"); multiplyThreadTransactionManager.multiplyThreadTransactionCommit();
return "ss"; }复制代码
有兴趣的小伙伴可以做进一步测试。
在本公司项目组中利用这个机制优化现有的业务，性能提升了约 70%。
比起网上常见的多线程事务管理器，主要做了如下增强
1.支持在已存在事务下运行。
在很多场景下，我们可能会遇到多线程事务外还存在其他事物的场景下，我们需要支持兼容多种事务环境。
2.支持自定义提交时机。
有时候我们不希望事务立马提交，希望他能够和外围事务保持一致，这时候可以将 runAsyncButWaitUntilAllDown 的 immediatelyCommit 参数写成 false，并手动调用 multiplyThreadTransactionCommit 方法去主动提交。
我们需要注意的地方，任何事都是有舍有得的，耗时的显著降低是因为利用了更多的资源，比如线程资源和数据库连接资源，尤其是数据库连接资源，更是额外宝贵，我们一定要合理评估我们的每一项决策是否有意义，风险和回报是否成正比。
还有一点需要注意，在极高并发的情况下，多线程事务容易造成死锁，因为当主事务开启的情况下，他要为他下面的子线程事务开启连接，当连接不够时就容易造成循环等待。一个比较好的做法是提前获得所有连接，并设置一个合理的超时时间
如果有小伙伴遇到了其他相关疑问，或者使用此代码发现了问题，欢迎留言讨论，共同进步。
文章转载自：doFix
划线
评论
复制
发布于: 刚刚阅读数: 3
快乐非自愿限量之名
关注
还未添加个人签名 2023-06-19 加入
还未添加个人简介
评论