Spring多线程事务解决方案

• • 多线程事务场景以及解决思路
  • 代码实现
  • • 工具类使用演示
    • 效果展示

多线程事务场景以及解决思路

多线程事务场景举例

对批量操作进行性能优化时会用到多线程来并行处理，从而提高运行效率。而时有业务要求保证批量操作事务的一致性，但不同线程所对应的是不同的事务，这就无法通过Spring提供的注解@Transactional来保证线程间的事务一致性。这也是Spring事务失效的一种情况。

线程间不是同一个事务

因为数据库连接Connection以及Spring事务的参数（事务名、事务是否只读、事务传播行为等）是保存在ThreadLocal中，不同的线程中保存的Connection是不同的，所以数据库层面对于不同的连接就无法满足事务。

解决思路（分布式事务思路）

既然Spring提供的声明式事务注解@Transactional无法满足多线程间的事务一致性，那可以试试通过编程式事务的方式来解决。主要思路：每个线程都开启各自的事务，待所有线程的业务执行完成，统一提交或回滚。
我的解决方案参考分布式事务2PC（Two-phase commit protocol），中文叫二阶段提交。需要注意的是2PC是同步阻塞协议，各个线程需要等待所有的线程执行完成后才能进行下一步操作，在使用线程池执行任务时，如果线程池的最大线程数小于任务列表的数量，就会发生“死锁”，即获取到线程的任务阻塞等待没有获取线程的任务执行完成，而没有获取线程的任务会在阻塞队列中等待空闲线程的调用。这种情况需要使用一阶段的超时机制来“解开”，超时机制会发送回滚命令，线程池收到后进行回滚，但这种情况任务始终无法提交，再次提交结果依然是等到超时再回滚。再使用中需要结合具体业务来对线程池参数以及数据库连接池参数进行合理的设置。如果这里听的优点迷，可以先看下面具体代码实现再来结合这段文字思考。

可以参考文章：面试必问：分布式事务六种解决方案

代码实现

我这里封装成了一个工具类，通过方法execute(List runnableList, Executor executor)传入任务列表和线程池，我的代码实现不具备任务编排的能力，有需要可以结合CompletableFuture类自己实现一个。

import lombok.extern.slf4j.Slf4j;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author poxiao
 * @create 2023-01-05 22:22
 * <p>
 * 多线程事务管理器
 * 基于分布式事务思想，采用2PC（Two-phase commit protocol）协议
 * 解决基于线程池的多线程事务一致性问题
 */
@Slf4j
public class MultiThreadingTransactionManager {

    /**
     * 事务管理器
     */
    private final PlatformTransactionManager transactionManager;

    /**
     * 超时时间
     */
    private final long timeout;

    /**
     * 时间单位
     */
    private final TimeUnit unit;

    /**
     * 一阶段门闩，(第一阶段的准备阶段)，当所有子线程准备完成时（除“提交/回滚”操作以外的工作都完成），countDownLatch的值为0
     */
    private CountDownLatch oneStageLatch = null;

    /**
     * 二阶段门闩，(第二阶段的执行执行)，主线程将不再等待子线程执行，直接判定总的任务执行失败，执行第二阶段让等待确认的线程进行回滚
     */
    private final CountDownLatch twoStageLatch = new CountDownLatch(1);

    /**
     * 是否提交事务，默认是true（当任一线程发生异常时，isSubmit会被设置为false，即回滚事务）
     */
    private final AtomicBoolean isSubmit = new AtomicBoolean(true);

    /**
     * 构造方法
     * @param transactionManager 事务管理器
     * @param timeout 超时时间
     * @param unit 时间单位
     */
    public MultiThreadingTransactionManager(PlatformTransactionManager transactionManager, long timeout, TimeUnit unit) {
        this.transactionManager = transactionManager;
        this.timeout = timeout;
        this.unit = unit;
    }

    /**
     * 线程池方式执行任务，可保证线程间的事务一致性
     * @param runnableList 任务列表
     * @param executor 线程池
     * @return
     */
    public boolean execute(List<Runnable> runnableList, ExecutorService executor) {

        // 排除null值
        runnableList.removeAll(Collections.singleton(null));

        // 属性初始化
        innit(runnableList.size());

        // 遍历任务列表并放入线程池
        for (Runnable runnable : runnableList) {
            // 创建线程
            Thread thread = new Thread() {
                @Override
                public void run() {
                    // 如果别的线程执行失败，则该任务就不需要再执行了
                    if (!isSubmit.get()) {
                        log.info("当前子线程执行中止，因为线程事务中有子线程执行失败");
                        oneStageLatch.countDown();
                        return;
                    }
                    // 开启事务
                    TransactionStatus transactionStatus = transactionManager.getTransaction(new DefaultTransactionDefinition());
                    try {
                        // 执行业务逻辑
                        runnable.run();
                    } catch (Exception e) {
                        // 执行体发生异常，设置回滚
                        isSubmit.set(false);
                        log.error("线程{}:业务发生异常,执行体:{}", Thread.currentThread().getName(), runnable);
                    }
                    // 计数器减一
                    oneStageLatch.countDown();
                    try {
                        //等待所有线程任务完成，监控是否有异常，有则统一回滚
                        twoStageLatch.await();
                        // 根据isSubmit值判断事务是否提交，可能是子线程出现异常，也有可能是子线程执行超时
                        if (isSubmit.get()) {
                            // 提交
                            transactionManager.commit(transactionStatus);
                            log.info("线程{}:事务提交成功,执行体:{}", Thread.currentThread().getName(), runnable);
                        } else {
                            // 回滚
                            transactionManager.rollback(transactionStatus);
                            log.info("线程{}:事务回滚成功,执行体:{}", Thread.currentThread().getName(), runnable);
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            executor.execute(thread);
        }

        /**
         * 主线程担任协调者，当第一阶段所有参与者准备完成，oneStageLatch的计数为0
         * 主线程发起第二阶段，执行阶段（提交或回滚）,根据
         */
        try {
            // 主线程等待所有线程执行完成，超时时间设置为五秒
            oneStageLatch.await(timeout, unit);
            long count = oneStageLatch.getCount();
            System.out.println("countDownLatch值：" + count);
            // 主线程等待超时，子线程可能发生长时间阻塞，死锁
            if (count > 0) {
                // 设置为回滚
                isSubmit.set(false);
                log.info("主线线程等待超时,任务即将全部回滚");
            }
            twoStageLatch.countDown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 返回结果，是否执行成功，事务提交即为执行成功，事务回滚即为执行失败
        return isSubmit.get();
    }

    /**
     * 初始化属性
     * @param size 任务数量
     */
    private void innit(int size) {
        oneStageLatch = new CountDownLatch(size);
    }
}

注意1： 2PC是同步阻塞协议，各个任务会等待所有的任务完成准备阶段才能进一步执行，所以在使用中一定要给任务列表提供充足的空闲线程，比如任务列表长度为8，线程池最大线程数不能小于8，否则会使其中的几个任务得不到执行，而其他线程会一直进行等待。即使有一阶段超时处理，事务也始终得不到提交。
注意2： 如果你的任务是对数据库进行操作，需要考虑数据库连接是否充足，线程等待过程中不会释放数据库连接，如果Connection不够，即使任务被线程池调度执行，也会阻塞在获取数据库连接中，同样会发生“死锁”。

工具类使用演示

@Override
    public boolean addBatchByMultithreading2(List<User> users) {
        // 创建线程池
        ThreadPoolExecutor executor = UserServiceThreadPool.getExecutor();
        // 创建多线程事务管理器，传入事务管理器，指定超时时间为3秒
        MultiThreadingTransactionManager multiThreadingTransactionManage = new MultiThreadingTransactionManager(transactionManager, 3, TimeUnit.SECONDS);
        List<Runnable> runnableList = new ArrayList<>();
        // 模拟任务出现异常
        // runnableList.add(() -> {
        //     int a = 10 / 0;
        // });
        users.forEach((x) -> {
            runnableList.add(new Runnable() {
                @Override
                public void run() {
                    System.out.println("当前线程：" + Thread.currentThread().getName() + "插入数据：" + x);
                    userMapper.insert(x);
                }
            });
        });
        // 执行
        return multiThreadingTransactionManage.execute(runnableList, executor);
    }

我这里采用自定义线程池，线程池参数如下：

核心线程数为10、采用无界队列（可能会导致OOM，结合业务使用）

效果展示

任务业务正常，无异常抛出时正常提交事务情况

Controller层代码

Service层代码

数据库无记录

Postman测试

控制台打印日志提示事务提交成功，并且数据库新增8条记录

展示出现异常任务时回滚事务情况

Service层代码
控制台打印日志提示，多个线程事务进行回滚