多线程环境spring如何实现事务管理?

我们很多人都知道在springboot中要开启事务，加上@Transactional 就可以了,这个是spring框架提供的声明式事务，用起来确实很方便。但是对事务原理稍微有点深入的话就知道声明式事务是基于AOP实现的，那也就是说事务是在同一个线程内才有效，如果是不同线程的业务使用这个注解事务会失效,比如用@Async注解的方法事务是无效的。至于为什么事务无效，下一节我专门分享spring事务的实现原理。那问题就来了，我想要在多线程环境下批量处理业务，也想让这些业务保证事务特性，那该如何实现呢？

比如有这么一个场景，需要往数据库中的三张表分别插入数据，只要一个表数据插入失败，则所有表数据都需要回滚。正常情况下，我们一般采用下面的方法实现：

1、开始一个事务

2、获取一个连接

3、使用连接分别 写入三张表的数据

4、成功则提交事务

5、失败则回滚事务

伪代码如下：

@Transactional(rollbackFor = Exception.class)
public void doTask(){
  // 写入第一张表
  //写入第二张表
  // 写入第三张表
}

这种方式使用了声明式事务管理，在数据量小的场景是没有问题的，如果这几张表数据量都很大，这样一个线程串行写入的方式就很耗时，那么我们就会想到用多线程并行写入的方式，比如用三个线程分别写入三个表的数据，这样时间就减少三分之二。

伪代码如下：

创建一个线程池
 ExecutorService executorService = Executors.newFixedThreadPool(3)
 使用线程池分别执行三张表的写入业务
 executorService.execute(()=>{
   // 第一张表
 });
 executorService.execute(()=>{
   // 第二张表
 });
 executorService.execute(()=>{
   // 第三张表
 });

这种方式可以并行写入三张表了，但是有一个问题就是他们是三个不同的事务，其次就是各自的事务是自动提交的。也就是说如果都成功了还好，但是如果其中一个表写入时候发生了异常导致数据没写进去，但是另外两张表事务成功了那就出现三张表数据不一致问题。为了实现最终一致性，需要对上面的代码修改一下，保证三张表数据要么都写进去，要么都不写。

改造后代码如下：

    //创建一个事务定义（隔离级别等事务相关的配置)  
    TransactionDefinition transactionDefinition = new DefaultTransactionDefinition();
       // 线程同步器，用来同步多个线程的
        CyclicBarrier cyclicBarrier = new CyclicBarrier(3);
       //线程安全计数器
        AtomicInteger atomicInteger = new AtomicInteger(0);
       //开启三个线程 分别执行业务
        for (long i = 1; i <= 3; i++) {
            final long k = i;
            executorService.execute(() -> {
               //线程内开启各自的事务（transactionManager是spring容器中的事务管理器)
                TransactionStatus status = transactionManager.getTransaction(transactionDefinition);
                try {
                    log.info("线程{}执行业务", Thread.currentThread().getName());
                     // 这里是具体的业务操作，比如向表插入数据
                  
                    // 如果写入成功 则计数器加1
                    atomicInteger.incrementAndGet();
                } catch (Exception e) {
                    log.info("异常:", e);
                }
                try {
                    //这里使用线程屏障 ，等待所有线程执行完毕后 再往下执行
                    cyclicBarrier.await();
                   // 比较计数器，如果是计数器最终结果和任务数量相同，则认为各自业务都没有异常
                    if (atomicInteger.get() == 3) {
                        log.info("线程{}提交事务", Thread.currentThread().getName());
                       // 提交事务
                        transactionManager.commit(status);
                    } else {
                       //只要有一个任务失败，则都需要回滚事务
                        log.info("线程{}回滚事务", Thread.currentThread().getName());
                        transactionManager.rollback(status);
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                } catch (BrokenBarrierException e) {
                    throw new RuntimeException(e);
                }
            });
        }

以上就是通过开启手动编程的事务方式可以灵活的控制事务的提交和回滚，从而实现多线程环境下事务的一致性，保证数据最终强一致性。