异步编排多线程任务事务控制

/**
 * <p>
 * <B>Description: 异步编排多线程任务事务控制</B>
 * </P>
 * Revision Trail: (Date/Author/Description)
 * 2022/12/26 Ryan Huang CREATE
 * 多线程异步处理时的事务管理
 *      1. addFunction 添加要异步执行的方法
 *      2. execute 方法中使用全局的计数器和异常标记字段统计异步线程执行的结果，当所有的异步线程执行完后根据异常标记字段判断是否回滚还是提交事务。
 * @author Ryan Huang
 * @version 1.0
 */
public class ThreadTransaction {

    /**
     * 日志
     */
    private final Logger LOG = LoggerFactory.getLogger(ThreadTransaction.class);

    /**
     * 事务管理
     */
    private PlatformTransactionManager platformTransactionManager;

    /**
     * JUC线程池
     */
    private ThreadPoolExecutor threadPoolExecutor;

    /**
     * Spring线程池
     */
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;


    private List<Supplier> supplierList = new ArrayList<>();

    /**
     * 执行计数器
     */
    private volatile CountDownLatch countDownLatch;

    /**
     * 是否存在异常
     */
    AtomicReference<Boolean> isError = new AtomicReference<>(false);

    public ThreadTransaction(PlatformTransactionManager platformTransactionManager, ThreadPoolTaskExecutor threadPoolTaskExecutor) {
        this.platformTransactionManager = platformTransactionManager;
        this.threadPoolTaskExecutor = threadPoolTaskExecutor;
    }

    public ThreadTransaction(PlatformTransactionManager platformTransactionManager, ThreadPoolExecutor threadPoolExecutor) {
        this.platformTransactionManager = platformTransactionManager;
        this.threadPoolExecutor = threadPoolExecutor;
    }

    public ThreadTransaction(PlatformTransactionManager platformTransactionManager, ThreadPoolTaskExecutor threadPoolTaskExecutor, int size) {
        this.platformTransactionManager = platformTransactionManager;
        this.threadPoolTaskExecutor = threadPoolTaskExecutor;
        supplierList = new ArrayList<>(size);
    }

    /**
     * 添加要异步执行的方法程序
     */
    public boolean addFunction(Supplier supplier){
        return supplierList.add(supplier);
    }

    public void execute(){
        LOG.info("多线程事务开始...");
        countDownLatch = new CountDownLatch(supplierList.size());
        for (Supplier supplier : supplierList) {
            this.threadPoolTaskExecutor.submit(new TransactionRunnable(platformTransactionManager, supplier));
        }
        try {
            if (isError.get()) {
                LOG.error("多线程执行失败，事务已回滚！");
                throw new RuntimeException("多线程执行失败！");
            }
            LOG.info("多线程执行成功，事务已提交！");
        }catch (Exception e){
            LOG.error("多线程执行失败：" + e.getMessage());
            e.printStackTrace();
        }
    }

    class TransactionRunnable implements Runnable{

        /**
         * 事务管理
         */
        private PlatformTransactionManager platformTransactionManager;

        private Supplier supplier;

        public TransactionRunnable(PlatformTransactionManager platformTransactionManager, Supplier supplier) {
            this.platformTransactionManager = platformTransactionManager;
            this.supplier = supplier;
        }

        @Override
        public void run() {
            DefaultTransactionDefinition defaultTransactionDefinition = new DefaultTransactionDefinition();
            defaultTransactionDefinition.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
            TransactionStatus transaction = this.platformTransactionManager.getTransaction(defaultTransactionDefinition);
            try {
                this.supplier.get();
            } catch (Exception e){
                isError.set(true);
                LOG.error("多线程事务执行失败{}", e.getMessage());
                e.printStackTrace();
            }
            countDownLatch.countDown();
            try {
                if (isError.get()) {
                    LOG.info("多线程事务（子线程）回滚");
                    platformTransactionManager.rollback(transaction);
                } else {
                    LOG.info("多线程事务（子线程）提交");
                    platformTransactionManager.commit(transaction);
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
}