一、前言

本系列用来记录一些在实际项目中的小东西，并记录在过程中想到一些小东西，因为是随笔记录，所以内容不会过于详细。

---

本篇从开篇到完成至少搁置了半年，因此之前的一些想法记不太清楚了。总之：本篇的方案未经过实际验证，谨慎使用。

本篇的灵感来源自 Spring在多线程环境下如何确保事务一致性，正如其文章内容所属，想实现 多线程下事务的一致性 有多重方案选择，包括最基础的 JDBC 编程、或者 分布式事务思想等，不必局限于此。

二、项目背景

某个项目中需要处理用户上传的 Excel 文件，对每条记录单独处理且处理过程比较耗时，因此使用了线程池来加快处理速度。简单Demo 如下：

java">    @Override
    public void testTransaction() {
        // 模拟构造大量数据
        List<Integer> indexs = initData(100);
        // 子线程插入数据
        indexs.parallelStream()
                .forEach(index ->
                        transactionTemplate.executeWithoutResult(
                                transactionStatus ->
                                        saveIndexDemo(index.toString())));
    }
 	/**
     * 模拟大量数据
     *
     * @return
     */
    private List<Integer> initData(int size) {
        List<Integer> indexs = Lists.newArrayList();
        for (int i = 0; i < size; i++) {
            indexs.add(i);
        }
        return indexs;
    }

    /**
     * 子线程保存记录
     *
     * @param index
     */
    @SneakyThrows
    private void saveIndexDemo(String index) {
        log.info("{} 保存数据 {}", Thread.currentThread().getId(), index);
        demoMapper.insert(getDemo(index));
        // TODO : 模拟业务耗时
        Thread.sleep(10000);
    }


    /**
     * 获取实例
     *
     * @param userId
     * @return
     */
    private Demo getDemo(String userId) {
        final Demo demo = new Demo();
        demo.setUserId(userId);
        return demo;
    }

本来相安无事，但是有一天晚上几个用户上传了一个2w行的Excel，这里就在开启线程处理，而每天记录的比较耗时，因此导致每个线程都占用着一个DB连接不释放。最终导致 DB 连接数不够用，其他业务异常。

---

三、实现方案

实际改造方案：限制线程开启的数量来控制并发度。改造比较简单，这里直接列出，重点在下面的思路延伸。

java">    /**
     * 线程池
     */
    private static final ExecutorService EXECUTOR_SERVICE = Executors.newFixedThreadPool(5);

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void testTransaction() {
        // 模拟构造大量数据
        List<Integer> indexs = initData(100);

        final CompletableFuture[] futures =
                indexs.stream()
                        .map(index -> CompletableFuture.runAsync(() ->
                                transactionTemplate.executeWithoutResult(
                                        transactionStatus ->
                                                saveIndexDemo(index.toString())), EXECUTOR_SERVICE))
                        .toArray(CompletableFuture[]::new);
        CompletableFuture.allOf(futures).join();
    }

---

四、思路延伸

上面的逻辑实际是有问题的：每个 saveIndexDemo 都处于不同的事务中，也就是说其中某个 saveIndexDemo 如果异常了其他 saveIndexDemo 并不会回滚。因此这里的问题就演变成了：多线程下如何保证事务的一致性。

1. 方案一

子线程业务执行结束后统一挂起，等待所有子线程执行结束后判断是否提交事务 （2PC、3PC 的简化版）。如下：这里使用 CountDownLatch 来阻塞所有子线程，当所有子线程执行结束后根据 atomicBoolean 的值判断是否有事务执行异常，如果异常，则将所有事务回滚。

java">      @SneakyThrows
    @Override
    public String testTransaction01() {
        demoMapper.delete(null);

        // 初始化数据
        List<Integer> indexs = initData(10);
        // 等待子线程执行结束
        CountDownLatch countDownLatch = new CountDownLatch(indexs.size());
        // 业务执行标志
        AtomicBoolean atomicBoolean = new AtomicBoolean(true);
        CountDownLatch endFlag = new CountDownLatch(indexs.size());
        // 模拟每个业务单独开一个线程来执行，并且每个业务内部都开启了一个事务
        for (Integer index : indexs) {
            // 如果线程池线程数量 < 任务数量 会导致任务卡死：线程执行后 countDownLatch.await(); 等待剩余任务执行完成，而线程池中线程被占满，导致线程无法释放就无法再执行剩余任务
            EXECUTOR_SERVICE.execute(() ->
                    transactionTemplate.executeWithoutResult(transactionStatus -> {
                        runTranscation(index, countDownLatch, atomicBoolean, endFlag);
                    }));
        }
        // 等待所有线程执行结束后接口再返回
        endFlag.await();

        return "success";
    }

    /**
     * 执行事务
     *
     * @param index
     * @param countDownLatch
     * @param atomicBoolean
     * @param endFlag
     */
    private void runTranscation(Integer index, CountDownLatch countDownLatch, AtomicBoolean atomicBoolean, CountDownLatch endFlag) {
        try {
            // 业务执行结束
            saveIndexDemo(index.toString());
            countDownLatch.countDown();
            // 等待其他线程业务执行结束
            countDownLatch.await();
        } catch (Exception e) {
            // 如果业务执行异常，设置标志位为 false
            countDownLatch.countDown();
            atomicBoolean.set(false);
            // 如果某个线程执行异常，抛出异常回滚
            throw new RuntimeException("测试异常, 事务回滚");
        } finally {
            endFlag.countDown();
        }
    }

上面代码虽然实现了所谓的功能，但是还是存在一些漏洞的，包括但不限于：

1. 线程池线程数量必须大于等于任务数量，否则会导致任务卡死：线程执行后 countDownLatch.await(); 等待剩余任务执行完成，而线程池中线程被占满，导致线程无法释放就无法再执行剩余任务。
2. 主线程 countDownLatch.await() 这一步执行结束时，子线程中的事务可能还没提交，不过可以使用其他标志位判断是否全部执行结束,这里不再演示。
3. 如果子线程其中一个因为网络或锁等问题迟迟不能执行结束，会导致其他子线程的事务一并无法提交，数据库连接池的连接会一直被占用。
4. 因为子线程是同时开启线程并且互相等待的，所以一定要注意死锁情况。如下：
   1. 线程A 执行业务逻辑时执行 SQL select * from demo where id = 1 for update， 此SQL会锁住 id = 1 的记录。
   2. 线程B 执行业务逻辑时执行SQL update demo set user_id = 'demo' where id = '1'，此时会发现 id = 1 的记录被线程A锁住，因此等待线程A执行结束。
   3. 线程A执行完业务逻辑后执行到 countDownLatch.await(); 后等待其他线程业务逻辑执行结束。而此时线程B因为等待线程A持有的记录A 的锁迟迟无法结束业务逻辑，也就无法执行到 countDownLatch.countDown(); 的逻辑，从而形成死锁的局面。

综上，该方案不推荐使用。

2. 方案二

本方案基于Spring在多线程环境下如何确保事务一致性 中的内容，做了部分改造，详细思路在该文中有过介绍，本篇不过多赘述细节。

以下面Demo为例，实现功能为 ：初始化 15 条数据，每条数据使用一个单独的线程，每个线程开辟一个单独的事务来将数据插入到数据库中。当其中一个线程的事务出现异常时，所有事务统一回滚。

java">    @Transactional(rollbackFor = Exception.class)
    @Override
    public void testTransaction02() {
        // 初始化 15 条数据
        List<Integer> indexs = initData(15);
        // 并发执行，每一个线程执行事务，事务的内容是将 Index 插入到 demo 表中
        CompletableFuture<TransactionUtil.TransactionResult>[] futures =
                indexs.stream()
                        .map(index ->
                                CompletableFuture.supplyAsync(() ->
                                        transactionUtil.executeWithoutResult(
                                                transactionStatus ->
                                                        saveIndexDemo(index.toString()))))
                        .toArray(CompletableFuture[]::new);
        // 等待所有 CompletableFuture 执行结束
        CompletableFuture.allOf(futures).join();

        // 获取子线程执行结果
        List<TransactionUtil.TransactionResult> transactionResults =
                Arrays.stream(futures)
                        .map(CompletableFuture::join)
                        .collect(Collectors.toList());
        // 批量处理所有子事物
        transactionUtil.batchTrim(transactionResults);
 }

上面代码的关键实现是下面的工具类，如下：

java">/**
 * @Desc : 从 org.springframework.transaction.interceptor.TransactionAspectSupport#invokeWithinTransaction 中参考下整个流程
 */
@Slf4j
@Component
public class TransactionUtil {

    @Resource
    private PlatformTransactionManager transactionManager;

    @Resource
    private TransactionDefinition transactionDefinition;


    /**
     * 基于 TransactionOperations#executeWithoutResult(java.util.function.Consumer) 改造
     *
     * @param function
     */
    public <T> TransactionResult<T> execute(Function<TransactionStatus, T> function) {
        TransactionStatus status = beginNewTransaction();
        TransactionResult<T> transactionResult =
                new TransactionResult<>(status, TransactionResource.copyTransactionResource());
        try {
            transactionResult.setData(function.apply(status));
        } catch (Throwable ex) {
            log.error("Transaction failed", ex);
            transactionResult.setEx(ex);
        } finally {
            clearTransactionResource();
        }
        return transactionResult;
    }

    /**
     * 基于 TransactionOperations#executeWithoutResult(java.util.function.Consumer) 改造
     * 需要注意：在本篇代码中，执行该方法的是线程池中的某一个线程，而不是主线程
     *
     * @param consumer
     */
    public TransactionResult<Void> executeWithoutResult(Consumer<TransactionStatus> consumer) {
        // 为当前线程开启一个新的事务
        TransactionStatus status = beginNewTransaction();
        // 封装成 TransactionResult 对象
        TransactionResult<Void> transactionResult =
                new TransactionResult<>(status, TransactionResource.copyTransactionResource());
        try {
            //执行具体业务逻辑
            consumer.accept(status);
        } catch (Throwable ex) {
            log.error("Transaction failed", ex);
            transactionResult.setEx(ex);
        } finally {
            // 清除 beginNewTransaction(); 方法创建事务时绑定当前线程的一些事务资源，防止线程池的线程复用时出现问题
            clearTransactionResource();
        }
        return transactionResult;
    }

    /**
     * 开启事务资源
     *
     * @return
     */
    private TransactionStatus beginNewTransaction() {
        // 开启一个新事务---此时autocommit已经被设置为了false,并且当前没有事务,这里创建的是一个新事务
        return transactionManager.getTransaction(transactionDefinition);
    }

    /**
     * 清除事务资源
     * beginNewTransaction 方法在创建新事物的同时还会给当前线程绑定一些事务属性，而由于线程池的线程是复用的，所以当下次再使用这个线程创新的事务时可能会出现问题：
     * 如 : 当第一次使用该线程创建事务时会给该线程绑定一个 DB 连接，而当事务提交后这个连接就会关闭，
     * 当下一次复用该线程执行事务时会判断当前线程已经绑定了 DB 连接就不会重新创建连接，而这个连接实际上已经关闭了，会出现 java.sql.SQLException: Connection is closed 问题
     */
    private void clearTransactionResource() {
        Maps.newHashMap(TransactionSynchronizationManager.getResourceMap()).keySet()
                .forEach(TransactionSynchronizationManager::unbindResource);
        if (TransactionSynchronizationManager.isActualTransactionActive()) {
            TransactionSynchronizationManager.clearSynchronization();
        }
    }

    /**
     * 批量处理事务
     * 需要注意：在本篇代码中，执行该方法的是主线程，线程池中每个线程都会开启一个事务并将事务属性返回交由主线程来统一处理。
     *
     * @param transactionResults
     * @return 是否提交
     */
    public boolean batchTrim(Collection<TransactionResult> transactionResults) {
        // 判断是否出现异常
        boolean rollback = transactionResults.stream()
                .anyMatch(transactionResult -> Objects.nonNull(transactionResult.getEx()));
        for (TransactionResult<?> transactionResult : transactionResults) {
            // 处理事务，提交或回滚
            if (rollback) {
                transactionResult.rollback();
            } else {
                transactionResult.commit();
            }
        }
        return !rollback;
    }


    /**
     * 保存当前事务资源,用于线程间的事务资源COPY操作
     */
    @Builder
    public static class TransactionResource {
        /**
         * 当前线程事务资源
         */
        private static final ThreadLocal<TransactionResource> RESOURCE_THREAD_LOCAL = new ThreadLocal<>();

        /**
         * 事务结束后默认会移除集合中的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;

        /**
         * 当前事务的属性拷贝
         *
         * @return
         */
        public static TransactionResource copyTransactionResource() {
            // 如果当前线程存在活跃事务则拷贝，否则返回一个基础 TransactionResource
            if (TransactionSynchronizationManager.isActualTransactionActive()) {
                return TransactionResource.builder()
                        .resources(Maps.newHashMap(TransactionSynchronizationManager.getResourceMap()))
                        .synchronizations(Sets.newHashSet(TransactionSynchronizationManager.getSynchronizations()))
                        .currentTransactionName(TransactionSynchronizationManager.getCurrentTransactionName())
                        .currentTransactionReadOnly(TransactionSynchronizationManager.isCurrentTransactionReadOnly())
                        .currentTransactionIsolationLevel(TransactionSynchronizationManager.getCurrentTransactionIsolationLevel())
                        .actualTransactionActive(TransactionSynchronizationManager.isActualTransactionActive())
                        .build();
            } else {
                return TransactionResource.builder()
                        .actualTransactionActive(TransactionSynchronizationManager.isActualTransactionActive())
                        .build();
            }
        }

        /**
         * 给当前线程注入事务属性
         */
        public void autowiredTransactionResource() {
            if (actualTransactionActive) {
                TransactionSynchronizationManager.initSynchronization();
                synchronizations.forEach(TransactionSynchronizationManager::registerSynchronization);
                resources.forEach(TransactionSynchronizationManager::bindResource);
                TransactionSynchronizationManager.setActualTransactionActive(actualTransactionActive);
                TransactionSynchronizationManager.setCurrentTransactionName(currentTransactionName);
                TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(currentTransactionIsolationLevel);
                TransactionSynchronizationManager.setCurrentTransactionReadOnly(currentTransactionReadOnly);
            }
        }

        /**
         * 移除事务属性
         *
         * @param all : 是否全部清除，默认只移除非 DataSource资源。这里解释下该参数的作用：
         *            事务提交后会自动清除事务相关的资源信息，所以不需要调用TransactionSynchronizationManager.clearSynchronization();同时事务信息也会自动清除 DataSource 相关的属性，所以也不需要清除。
         *            而对于上面代码来说，如果是子线程事务因为会执行提交或者回滚操作，会自动清除 DataSource 相关的属性，所以传 false; 而对主线程事务来说，事务提交并不是由我们控制，所以需要传 true，清空本身事务信息，为绑定子线程事务信息做准备
         */
        public void removeTransactionResource(boolean all) {
            if (actualTransactionActive) {
                // 事务结束后默认会移除集合中的DataSource作为key关联的资源记录,DataSource如果重复移除,unbindResource时会因为不存在此key关联的事务资源而报错
                resources.keySet().forEach(key -> {
                    if (all || !(key instanceof DataSource)) {
                        TransactionSynchronizationManager.unbindResource(key);
                    }
                });
            }
        }
    }

    /**
     * 执行结果
     *
     * @param <T>
     */
    public class TransactionResult<T> {
        /**
         * 事务状态
         */
        private TransactionStatus status;

        /**
         * 线程资源
         */
        private TransactionResource resource;

        /**
         * 异常信息
         */
        @Getter
        @Setter
        private Throwable ex;

        /**
         * 执行结果
         */
        @Getter
        @Setter
        private T data;

        public TransactionResult(TransactionStatus status, TransactionResource resource) {
            this.status = status;
            this.resource = resource;
        }

        /**
         * 事务提交
         * 需要注意：在本篇代码中，执行该方法的是主线程
         */
        public void commit() {
            // 如果主线程开启了事务，则保存主线程事务信息：因为主线程可能同步存在事务，所以需要先备份主线程的事务信息
            TransactionResource currentResource = TransactionResource.copyTransactionResource();
            try {
                // 移除当前线程(主线程)事务资源, 防止下面绑定其他事务资源时冲突，这里调用入参传 true ，清除主线程事务的全部资源信息
                currentResource.removeTransactionResource(true);
                if (TransactionSynchronizationManager.isSynchronizationActive()) {
                    // 清除活跃状态，否则注入子事务时会判断当前线程已经有活跃事务
                    TransactionSynchronizationManager.clearSynchronization();
                }
                // 注入子线程事务信息 : resource 中保存的事子线程的事务信息，将子线程事务信息注入到这里后再提交子线程事务
                resource.autowiredTransactionResource();
                // 提交子线程 : 事务提交后会自动清除事务相关的资源信息，所以不需要调用TransactionSynchronizationManager.clearSynchronization();,同时事务信息也会自动清除 DataSource 相关的属性，所以也不需要清除
                transactionManager.commit(status);
                // 清除子线程事务信息, 防止下面恢复主线程事务时属性冲突, 这里调用入参传 false ，因为上面调用了事务提交，所以会自动清除 DataSource 相关的属性，也就不需要我们自己清除了
                resource.removeTransactionResource(false);
            } catch (Exception ex) {
                log.info("commit error", ex);
                // 清除子线程事务信息, 防止下面恢复主线程事务时属性冲突
                resource.removeTransactionResource(false);
            } finally {
                log.info("commit");
                // 恢复主线程事务信息
                currentResource.autowiredTransactionResource();
            }
        }

        /**
         * 事务回滚
         */
        public void rollback() {
            // 如果主线程开启了事务，则保存主线程事务信息：因为主线程可能同步存在事务，所以需要先备份主线程的事务信息
            TransactionResource currentResource = TransactionResource.copyTransactionResource();
            try {
                // 移除当前线程(主线程)事务资源, 防止下面绑定其他事务资源时冲突，这里调用入参传 true ，清除主线程事务的全部资源信息
                currentResource.removeTransactionResource(true);
                if (TransactionSynchronizationManager.isSynchronizationActive()) {
                    // 清除活跃状态，否则注入子事务时会判断当前线程已经有活跃事务
                    TransactionSynchronizationManager.clearSynchronization();
                }
                // 注入子线程事务信息 : resource 中保存的事子线程的事务信息，将子线程事务信息注入到这里后再提交子线程事务
                resource.autowiredTransactionResource();
                // 提交子线程 : 事务提交后会自动清除事务相关的资源信息，所以不需要调用TransactionSynchronizationManager.clearSynchronization();,同时事务信息也会自动清除 DataSource 相关的属性，所以也不需要清除
                rollbackOnException(status, ex);
                // 清除子线程事务信息, 防止下面恢复主线程事务时属性冲突, 这里调用入参传 false ，因为上面调用了事务提交，所以会自动清除 DataSource 相关的属性，也就不需要我们自己清除了
                resource.removeTransactionResource(false);
            } catch (Exception ex) {
                log.info("commit error", ex);
                // 清除子线程事务信息, 防止下面恢复主线程事务时属性冲突
                resource.removeTransactionResource(false);
            } finally {
                log.info("commit");
                // 恢复主线程事务信息
                currentResource.autowiredTransactionResource();
            }
        }

        /**
         * 回滚
         *
         * @param status
         * @param ex
         * @throws TransactionException
         */
        private void rollbackOnException(TransactionStatus status, Throwable ex) throws TransactionException {
            Assert.state(transactionManager != null, "No PlatformTransactionManager set");

            log.debug("Initiating transaction rollback on application exception", ex);
            try {
                transactionManager.rollback(status);
            } catch (TransactionSystemException ex2) {
                log.error("Application exception overridden by rollback exception", ex);
                ex2.initApplicationException(ex);
                throw ex2;
            } catch (RuntimeException | Error ex2) {
                log.error("Application exception overridden by rollback exception", ex);
                throw ex2;
            }
        }
    }
}

上面的方案存在问题：

1. 在编写代码时注意多个线程事务在执行过程中的死锁问题。
2. 最终事务的提交方法 TransactionUtil#batchTrim 并不是原子的，当多个事务中某个事务提交失败了，则无法对之前提交成功的事务回滚。

3. 方案三

基于方案二，又发散出了思维三：所有子线程与主线程公用同一个事务信息，最后在主线程执行结束后将这个事务提交。

以下面Demo为例，实现功能为 ：初始化 15 条数据，每条数据使用一个单独的线程，每个线程都会使用主线程的事务来将数据插入到数据库中。当其中一个线程或主线程出现异常时，事务统一回滚。

java">    @Transactional
    @Override
    public void testTransaction03() {
        // 初始化 15 条数据
        List<Integer> indexs = initData(15);

        // 拷贝主线程事务资源
        SingleTransactionUtil.TransactionResource transactionResource =
                SingleTransactionUtil.TransactionResource.copyTransactionResource();

        // 并发执行，每一个线程执行事务，事务的内容是将 Index 插入到 demo 表中
        CompletableFuture<SingleTransactionUtil.TransactionResult>[] futures =
                indexs.stream()
                        .map(index ->
                                CompletableFuture.supplyAsync(() ->
                                        singleTransactionUtil.executeWithoutResult(transactionResource,
                                                () -> saveIndexDemo(index.toString()))))
                        .toArray(CompletableFuture[]::new);
        // 等待所有 CompletableFuture 执行结束
        CompletableFuture.allOf(futures).join();

        final boolean anyMatch = Arrays.stream(futures)
                .anyMatch(f -> f.join().getEx() != null);
        if (anyMatch){
            throw new RuntimeException("测试异常, 事务回滚");
        }

        dataDemoMapper.insert(new DataDemo("1"));
        dataDemoMapper.insert(new DataDemo("2"));
    }

上面代码的关键实现是下面的工具类，如下：

java">/**
 * @Desc : 从 org.springframework.transaction.interceptor.TransactionAspectSupport#invokeWithinTransaction 中参考下整个流程
 */
@Slf4j
@Component
public class SingleTransactionUtil {

    @Resource
    private PlatformTransactionManager transactionManager;

    @Resource
    private TransactionDefinition transactionDefinition;


    /**
     * 基于 TransactionOperations#executeWithoutResult(java.util.function.Consumer) 改造
     * 执行
     *
     * @param supplier
     */
    public <T> TransactionResult<T> execute(TransactionResource transactionResource,  Supplier<T> supplier) {
        transactionResource.autowiredTransactionResource();
        TransactionResult<T> transactionResult = new TransactionResult<>();
        try {
            transactionResult.setData(supplier.get());
        } catch (Throwable ex) {
            log.error("Transaction failed", ex);
            transactionResult.setEx(ex);
        } finally {
            clearTransactionResource();
        }
        return transactionResult;
    }

    /**
     * 基于 TransactionOperations#executeWithoutResult(java.util.function.Consumer) 改造
     * 执行
     *
     * @param runnable
     */
    public TransactionResult<Void> executeWithoutResult(TransactionResource transactionResource, Runnable runnable) {
        transactionResource.autowiredTransactionResource();
        TransactionResult<Void> transactionResult = new TransactionResult<>();
        try {
            runnable.run();
        } catch (Throwable ex) {
            log.error("Transaction failed", ex);
            transactionResult.setEx(ex);
        } finally {
            clearTransactionResource();
        }

        return transactionResult;
    }
    
    /**
     * 清除事务资源
     * beginNewTransaction 方法在创建新事物的同时还会给当前线程绑定一些事务属性，而由于线程池的线程是复用的，所以当下次再使用这个线程创新的事务时可能会出现问题：
     * 如 : 当第一次使用该线程创建事务时会给该线程绑定一个 DB 连接，而当事务提交后这个连接就会关闭，
     * 当下一次复用该线程执行事务时会判断当前线程已经绑定了 DB 连接就不会重新创建连接，而这个连接实际上已经关闭了，会出现 java.sql.SQLException: Connection is closed 问题
     */
    private void clearTransactionResource() {
        Maps.newHashMap(TransactionSynchronizationManager.getResourceMap()).keySet()
                .forEach(TransactionSynchronizationManager::unbindResource);
        if (TransactionSynchronizationManager.isActualTransactionActive()) {
            TransactionSynchronizationManager.clearSynchronization();
        }
    }

    @Slf4j
    public static class TransactionResult<T> {
        /**
         * 异常信息
         */
        @Getter
        @Setter
        private Throwable ex;

        /**
         * 执行结果
         */
        @Getter
        @Setter
        private T data;
    }


    /**
     * 保存当前事务资源,用于线程间的事务资源COPY操作
     */
    @Builder
    public static class TransactionResource {
        /**
         * 当前线程事务资源
         */
        private static final ThreadLocal<TransactionResource> RESOURCE_THREAD_LOCAL = new ThreadLocal<>();

        /**
         * 事务结束后默认会移除集合中的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;

        /**
         * 当前事务的属性拷贝
         *
         * @return
         */
        public static TransactionResource copyTransactionResource() {
            // 如果当前线程存在活跃事务则拷贝，否则返回一个基础 TransactionResource
            if (TransactionSynchronizationManager.isActualTransactionActive()) {
                return TransactionResource.builder()
                        .resources(Maps.newHashMap(TransactionSynchronizationManager.getResourceMap()))
                        .synchronizations(Sets.newHashSet(TransactionSynchronizationManager.getSynchronizations()))
                        .currentTransactionName(TransactionSynchronizationManager.getCurrentTransactionName())
                        .currentTransactionReadOnly(TransactionSynchronizationManager.isCurrentTransactionReadOnly())
                        .currentTransactionIsolationLevel(TransactionSynchronizationManager.getCurrentTransactionIsolationLevel())
                        .actualTransactionActive(TransactionSynchronizationManager.isActualTransactionActive())
                        .build();
            } else {
                return TransactionResource.builder()
                        .actualTransactionActive(TransactionSynchronizationManager.isActualTransactionActive())
                        .build();
            }
        }

        /**
         * 给当前线程注入事务属性
         */
        public void autowiredTransactionResource() {
            if (actualTransactionActive) {
                TransactionSynchronizationManager.initSynchronization();
                synchronizations.forEach(TransactionSynchronizationManager::registerSynchronization);
                resources.forEach(TransactionSynchronizationManager::bindResource);
                TransactionSynchronizationManager.setActualTransactionActive(actualTransactionActive);
                TransactionSynchronizationManager.setCurrentTransactionName(currentTransactionName);
                TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(currentTransactionIsolationLevel);
                TransactionSynchronizationManager.setCurrentTransactionReadOnly(currentTransactionReadOnly);
            }
        }

        /**
         * 移除事务属性
         *
         * @param all : 是否全部清除，默认只移除非 DataSource资源。这里解释下该参数的作用：
         *            事务提交后会自动清除事务相关的资源信息，所以不需要调用TransactionSynchronizationManager.clearSynchronization();同时事务信息也会自动清除 DataSource 相关的属性，所以也不需要清除。
         *            而对于上面代码来说，如果是子线程事务因为会执行提交或者回滚操作，会自动清除 DataSource 相关的属性，所以传 false; 而对主线程事务来说，事务提交并不是由我们控制，所以需要传 true，清空本身事务信息，为绑定子线程事务信息做准备
         */
        public void removeTransactionResource(boolean all) {
            if (actualTransactionActive) {
                // 事务结束后默认会移除集合中的DataSource作为key关联的资源记录,DataSource如果重复移除,unbindResource时会因为不存在此key关联的事务资源而报错
                resources.keySet().forEach(key -> {
                    if (all || !(key instanceof DataSource)) {
                        TransactionSynchronizationManager.unbindResource(key);
                    }
                });
            }
        }
    }
}

五、参考内容

Spring在多线程环境下如何确保事务一致性
https://mp.weixin.qq.com/s/gyEKbnzD3gIqfatL94DXsg