https://www.cnblogs.com/a5513633/p/13969222.html

/**
  2  * 带回滚的异步任务回调
  3  * 基类
  4  * @author Administrator
  5  *
  6  */
  7 public abstract class BaseCallBack implements Callable<String> {
  8 
  9     private static Logger logger = LoggerFactory.getLogger(BaseCallBack.class);
 10     /**
 11      * 需要回滚计数器
 12      */
 13     protected CountDownLatch rollBackLatch;
 14     /**
 15      * 主线程等待计数器
 16      */
 17     protected CountDownLatch mainThreadLatch;
 18     /**
 19      * 是否需要回滚
 20      */
 21     protected AtomicBoolean rollbackFlag;
 22     /**
 23      * 事务
 24      */
 25     protected PlatformTransactionManager transactionManager;
 26 
 27     protected abstract void doWork();
 28 
 29     @Override
 30     public String call() throws Exception {
 31         if (rollbackFlag.get()) {
 32             logger.info("需要回滚，直接不用执行了");
 33             mainThreadLatch.countDown();
 34             return Constant.ERROR_STR; // 如果其他线程已经报错 就停止线程
 35         }
 36         // 设置一个事务
 37         DefaultTransactionDefinition def = new DefaultTransactionDefinition();
 38         def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); // 事物隔离级别，开启新事务，这样会比较安全些。
 39         TransactionStatus status = transactionManager.getTransaction(def); // 获得事务状态
 40         try {
 41             logger.info("业务开始处理：{}", this.getClass().getName());
 42             this.doWork();
 43             logger.info("业务处理结束：{}", this.getClass().getName());
 44             // 业务处理结束
 45             mainThreadLatch.countDown();
 46             logger.info("线程内正常 mainThreadLatch.countDown();");
 47             rollBackLatch.await();// 线程等待
 48             if (rollbackFlag.get()) {
 49                 logger.info("回滚事务：{}", this.getClass().getName());
 50                 transactionManager.rollback(status);
 51             } else {
 52                 logger.info("提交事务：{}", this.getClass().getName());
 53                 transactionManager.commit(status);
 54             }
 55             return Constant.SAVE_SUCCESS;
 56         } catch (Exception e) {
 57             e.printStackTrace();
 58             // 如果出错了 就放开锁 让别的线程进入提交/回滚 本线程进行回滚
 59             rollbackFlag.set(true);
 60             transactionManager.rollback(status);
 61             rollBackLatch.countDown();
 62             mainThreadLatch.countDown();
 63             logger.info("线程内异常 mainThreadLatch.countDown();");
 64             return "操作失败：" + e.getMessage();
 65         }
 66     }
 67 
 68     public CountDownLatch getRollBackLatch() {
 69         return rollBackLatch;
 70     }
 71 
 72     public void setRollBackLatch(CountDownLatch rollBackLatch) {
 73         this.rollBackLatch = rollBackLatch;
 74     }
 75 
 76     public CountDownLatch getMainThreadLatch() {
 77         return mainThreadLatch;
 78     }
 79 
 80     public void setMainThreadLatch(CountDownLatch mainThreadLatch) {
 81         this.mainThreadLatch = mainThreadLatch;
 82     }
 83 
 84     public AtomicBoolean getRollbackFlag() {
 85         return rollbackFlag;
 86     }
 87 
 88     public void setRollbackFlag(AtomicBoolean rollbackFlag) {
 89         this.rollbackFlag = rollbackFlag;
 90     }
 91 
 92     public PlatformTransactionManager getTransactionManager() {
 93         return transactionManager;
 94     }
 95 
 96     public void setTransactionManager(PlatformTransactionManager transactionManager) {
 97         this.transactionManager = transactionManager;
 98     }
 99 
100 }
复制代码

Springboot 在多线程下，处理事务；

   目前很多开发的小伙伴们，都开始采用Springboot了，因为给我们带来了许多的开发便利，只需要我们关注编写逻辑代码。在工作中是会有很多小伙伴在项目逐步实用到多线程、线程池等相关技术，但是在使用多线程会出各种各样的问题；事务就是其中一种相对麻烦的事情；

大家都知道Springboot  只需要使用注解标签@Transactional 就可以使用事务了；但是往往在多线程下该处理事务的办法就失效，毫无效果；那又该如何呢？
首先大家都知道控制事务有两种方式；第一种就是刚才所说的注解式事务、第二种就是编程式事务；既然第一种失效那么我们就使用第二种方式来实现嘛；废话不多说，上代码！

package com.slsp.cloud.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.slsp.cloud.bean.User;
import com.slsp.cloud.service.UserService;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.math.RandomUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.Date;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 项目初始化加载
 */
@Component
@Slf4j
public class TradeApplicationRunner implements ApplicationRunner
{


    @Autowired
    private TransactionTemplate transactionTemplate;

    @Autowired
    private UserService userService;

    /**
     * 模拟多线程下事务处理
     * @param args
     * @throws Exception
     */
    @Override
    public void run(ApplicationArguments args) throws Exception
    {
        //线程池
        ExecutorService cachedThreadPool = Executors.newFixedThreadPool(5);
        for(int i=0;i<1;i++)
        {
            cachedThreadPool.execute(new Runnable()
            {
                @SneakyThrows
                public void run()
                {
                    log.info("多线程执行；");
                    //开启编程式事务；
                    transactionTemplate.execute(new TransactionCallback<Boolean>()
                    {
                        @Override
                        public Boolean doInTransaction(TransactionStatus transactionStatus)
                        {
                            try
                            {

                                User usr = new User();
                                //随机名字
                                String name = UUID.randomUUID().toString().replace("-", "");
                                System.out.println(name);
                                usr.setUserName(name);
                                usr.setSex(1);
                                //插入数据库
                                userService.save(usr);

                                //模拟报错；
                                int i=10/0;

                                String name2 = UUID.randomUUID().toString().replace("-", "");
                                System.out.println(name2);
                                User usr2 = new User();
                                usr2.setUserName(name2);
                                usr2.setSex(1);
                                userService.save(usr2);

                                return true;
                            }
                            catch (Exception e)
                            {
                                //事务回滚；
                                transactionStatus.setRollbackOnly();
                                e.printStackTrace();
                                return false;
                            }
                        }

                    });

                }
            });
        }
    }
}

打完收工；简单方便；

import org.springframework.stereotype.Service;
import org.springframework.transaction.support.TransactionTemplate;

import javax.annotation.Resource;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;

@Service
public class ThreadTest {
    @Resource
    private TransactionTemplate transactionTemplate;

    @Resource
    private TestMapper testMapper;


    public void test1() {
        final CountDownLatch al = new CountDownLatch(3);
        final AtomicBoolean ab = new AtomicBoolean(false);

        CompletableFuture.runAsync(() -> {
            transactionTemplate.execute((var e) -> {
                try {
                    TestDto testDto = new TestDto();
                    testDto.setName("测试线程1");
                    testMapper.add(testDto);
                    al.countDown();
                    al.await();
                    if (ab.get()) throw new RuntimeException("手动抛错");
                } catch (Exception exception) {
                    //事务回滚；
                    e.setRollbackOnly();
                    exception.printStackTrace();
                    al.countDown();
                    ab.set(true);
                    return false;
                }

                return true;
            });
        });
        CompletableFuture.runAsync(() -> {
            transactionTemplate.execute((var e) -> {
                try {
                    TestDto testDto = new TestDto();
                    testDto.setName("测试线程2");
                    testMapper.add(testDto);

                    al.countDown();
                    al.await();
                    if (ab.get()) throw new RuntimeException("手动抛错");
                } catch (Exception exception) {
                    //事务回滚；
                    e.setRollbackOnly();
                    exception.printStackTrace();
                    al.countDown();
                    ab.set(true);
                    return false;
                }

                return true;

            });
        });
        transactionTemplate.execute((var e) -> {
            try {
                TestDto testDto = new TestDto();
                testDto.setName("测试主线程");
                testMapper.add(testDto);
                
                al.countDown();
                al.await();
                if (ab.get()) throw new RuntimeException("手动抛错");
            } catch (Exception exception) {
                //事务回滚；
                e.setRollbackOnly();
                exception.printStackTrace();
                al.countDown();
                ab.set(true);
                return false;
            }

            return true;
        });
    }
}

注意该种方式未经过太多测试