手动事务及多线程事务

阳光穿透心脏的1/2处 2022-10-20 13:47 314阅读 0赞

### 文章目录 ###

*   *  1. 什么是事务？
     *  2.事务的特性（ACID）
     *  3.Spring 支持两种方式的事务管理
     *   *  1).编程式事务管理
         *  2)声明式事务管理
     *  多线程事务解决
     *   *  1）定义线程外集合，将线程处理结果放入集合，由外部线程处理
         *  2）定义线程外手动事务集合，将线程事务放入集合，由外部线程处理事务
         *  3）多线程处理事务管理

## 1. 什么是事务？ ##

事务是逻辑上的一组操作，要么都执行，要么都不执行。

另外，需要格外注意的是：事务能否生效数据库引擎是否支持事务是关键。比如常用的 MySQL 数据库默认使用支持事务的innodb引擎。但是，如果把数据库引擎变为 myisam，那么程序也就不再支持事务了！

## 2.事务的特性（ACID） ##

*  **原子性**（Atomicity）： 一个事务（transaction）中的所有操作，或者全部完成，或者全部不完成，不会结束在中间某个环节。事务在执行过程中发生错误，会被回滚（Rollback）到事务开始前的状态，就像这个事务从来没有执行过一样。即，事务不可分割、不可约简。
 *  **一致性**（Consistency）： 在事务开始之前和事务结束以后，数据库的完整性没有被破坏。这表示写入的资料必须完全符合所有的预设约束、触发器、级联回滚等。
 *  **隔离性**（Isolation）： 数据库允许多个并发事务同时对其数据进行读写和修改的能力，隔离性可以防止多个事务并发执行时由于交叉执行而导致数据的不一致。事务隔离分为不同级别，包括未提交读（Read uncommitted）、提交读（read committed）、可重复读（repeatable read）和串行化（Serializable）。
 *  **持久性**（Durability）: 事务处理结束后，对数据的修改就是永久的，即便系统故障也不会丢失。

## 3.Spring 支持两种方式的事务管理 ##

### 1).编程式事务管理 ###

通过 TransactionTemplate或者TransactionManager手动管理事务，实际应用中很少使用，但是对于你理解 Spring 事务管理原理有帮助。

使用TransactionTemplate 进行编程式事务管理的示例代码如下：

@Autowired
    private TransactionTemplate transactionTemplate;
    
    public void testTransaction() {
        
    
            transactionTemplate.execute(new TransactionCallbackWithoutResult() {
        
                @Override
                protected void doInTransactionWithoutResult(TransactionStatus transactionStatus) {
        
    
                    try {
        
    
                        // ....  业务代码
                    } catch (Exception e){
        
                        //回滚 放在catch 里面,防止程序异常而事务一直卡在哪里未提交
                        transactionStatus.setRollbackOnly();
                    }
    
                }
            });
    }

使用 TransactionManager 进行编程式事务管理的示例代码如下：

@Autowired
    private PlatformTransactionManager transactionManager;
    
    public void testTransaction() {
        
    
      TransactionStatus status = transactionManager.getTransaction(new DefaultTransactionDefinition());
              try {
        
                   // ....  业务代码
                  transactionManager.commit(status);
              } catch (Exception e) {
        
              // 放在catch 里面,防止程序异常而事务一直卡在哪里未提交
                  transactionManager.rollback(status);
              }
    }

### 2)声明式事务管理 ###

推荐使用（代码侵入性最小），实际是通过 AOP 实现（基于@Transactional 的全注解方式使用最多）。

使用 @Transactional注解进行事务管理的示例代码如下：

@Transactional(propagation=propagation.PROPAGATION_REQUIRED)
    public void aMethod {
        
      //do something
      B b = new B();
      C c = new C();
      b.bMethod();
      c.cMethod();
    }

## 多线程事务解决 ##

### 1）定义线程外集合，将线程处理结果放入集合，由外部线程处理 ###

@Autowired
    private ThreadPoolTaskExecutor executor;
    
    @Autowired
    private SaveServicesaveService;
    
    @Transactional(rollbackFor = Exception.class)
    public void handle(TaskList taskList) {
        
          
            List<Object> objList= new ArrayList<>(taskList.size());
            for (OutlineTask task: taskList) {
        
                try {
        
                    executor.submit(() -> {
        
     					// 业务逻辑 处理完成后将结果存入集合
                        objList.add(obj);
                    }).get();
                } catch (Exception e) {
        
                    e.printStackTrace();
                    if (e.getCause() instanceof BizException) {
        
                        throw (BizException) e.getCause();
                    }
                    throw new BizException("xx业务失败!");
                }
            }
    
            if (!CollectionUtils.isEmpty(objList)) {
        
                objList.forEach(item -> saveService.save(item));
            }
        }

### 2）定义线程外手动事务集合，将线程事务放入集合，由外部线程处理事务 ###

@Autowired
    private ThreadPoolTaskExecutor executor;
    
    @Autowired
    private PlatformTransactionManager transactionManager;
    
    @Autowired
    private SaveServicesaveService;
    
    @Transactional(rollbackFor = Exception.class)
    public void handle(TaskList taskList) {
        
            // 是否回滚标志位
            AtomicBoolean success = new AtomicBoolean(true);
            // 事务列表
            List<TransactionStatus> objList = new ArrayList<>(taskList.size());
    
            for (OutlineTask task : taskList) {
        
                try {
        
                    executor.submit(() -> {
        
                        //手动开启事务！
                        TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
                        // 业务逻辑 处理完成后将事务存入集合
                        objList.add(transactionStatus);
                    }).get();
                } catch (Exception e) {
        
                    e.printStackTrace();
                    if (e.getCause() instanceof BizException) {
        
                        throw (BizException) e.getCause();
                    }
                    // 出现异常回滚标志位更新
                    success.set(false);
                    throw new BizException("xx业务失败!");
                } finally {
        
                    if (!CollectionUtils.isEmpty(objList)) {
        
                        //提交
                        if (success.get()) {
        
                            objList.forEach(item -> transactionManager.commit(item));
                        } else {
        
                            objList.forEach(item -> transactionManager.rollback(item));
                        }
                    }
                }
            }
        }

### 3）多线程处理事务管理 ###

*  定义

import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    import org.springframework.transaction.PlatformTransactionManager;
    import org.springframework.transaction.TransactionDefinition;
    import org.springframework.transaction.TransactionStatus;
    import org.springframework.transaction.support.DefaultTransactionDefinition;
    
    import java.util.ArrayList;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.atomic.AtomicReference;
    import java.util.function.Supplier;
    
    /**
     * 多线程异步处理时的事务管理
     * 1.addFunction 添加要异步执行的方法
     * 2.execute方法中，使用全局的计数器和异常标记字段，统计个异步线程执行的结果
     * 当所有异步线程执行完之后，根据异常标记字段判断是回滚还是提交事务。
     *
     * @author yzd
     */
    public class MultiThreadTransactionComponent {
        
    
        private final Logger logger = LoggerFactory.getLogger(this.getClass());
    
        /**
         * 事务管理器
         */
        private PlatformTransactionManager platformTransactionManager;
    
        /**
         * 线程池
         */
        private ThreadPoolTaskExecutor threadPoolExecutor;
    
        /**
         * 有序 任务 集合
         */
        private ArrayList<Supplier> supplierList = new ArrayList<>(16);
    
        /**
         * 创建执行计数器
         */
        private CountDownLatch countDownLatch;
    
        /**
         * 异常标志位
         */
        private AtomicReference<Boolean> isError = new AtomicReference<>(false);
    
        /**
         * 构造函数
         *
         * @param transactionManager 事务管理器
         * @param threadPoolExecutor 线程池
         */
        public MultiThreadTransactionComponent(PlatformTransactionManager transactionManager, ThreadPoolTaskExecutor threadPoolExecutor) {
        
            this.platformTransactionManager = transactionManager;
            this.threadPoolExecutor = threadPoolExecutor;
        }
    
        /**
         * 添加要异步执行的方法程序
         *
         * @param supplier 任务
         */
        public void addFunction(Supplier supplier) {
        
            supplierList.add(supplier);
        }
    
        /**
         * 执行队列中的任务
         */
        public void execute() {
        
    
            // 初始化 线程 计数器
            countDownLatch = new CountDownLatch(supplierList.size());
    
            logger.info("【多线程事务】开始...");
            for (Supplier supplier : supplierList) {
        
                this.threadPoolExecutor.submit(new TransactionRunnable(platformTransactionManager, supplier));
            }
    
            try {
        
                // 主线程 等待计数器为 0 时 进行提交或回滚
                countDownLatch.await();
                if (isError.get()) {
        
                    logger.error("【多线程事务】多线程执行失败,事务已回滚");
                    // 主线程抛出自定义的异常
                    throw new RuntimeException("多线程执行失败");
                }
                logger.info("【多线程事务】多线程执行完成,事务已提交");
            } catch (InterruptedException e) {
        
                logger.error("多线程执行失败");
                // 主线程抛出自定义的异常
                throw new RuntimeException("多线程执行失败" + e.getMessage());
            }
        }
    
    
        /**
         * 实现 Runnable 接口
         */
        class TransactionRunnable implements Runnable {
        
    
            /**
             * 事务管理器
             */
            private PlatformTransactionManager platformTransactionManager;
            /**
             * 当前任务
             */
            private Supplier supplier;
    
            /**
             * 构造函数
             *
             * @param platformTransactionManager 事务管理器
             * @param supplier                   当前任务
             */
            public TransactionRunnable(PlatformTransactionManager platformTransactionManager, Supplier supplier) {
        
                this.platformTransactionManager = platformTransactionManager;
                this.supplier = supplier;
            }
    
            @Override
            public void run() {
        
                DefaultTransactionDefinition def = new DefaultTransactionDefinition();
                // 创建当前线任务的事务
                def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
                TransactionStatus transaction = this.platformTransactionManager.getTransaction(def);
                try {
        
                    // 尝试获取任务值
                    this.supplier.get();
                } catch (Exception e) {
        
                    // 异常时，设置错误标记
                    isError.set(true);
                    logger.error("【多线程事务】执行失败{}", e.getMessage());
                }
                // 线程计数器 -1
                countDownLatch.countDown();
                try {
        
                    // 子线程 等待计数器为 0 时 进行提交或回滚
                    countDownLatch.await();
                    if (isError.get()) {
        
                        logger.info("【多线程事务-子线程】事务回滚");
                        //事务回滚
                        platformTransactionManager.rollback(transaction);
                    } else {
        
                        logger.info("【多线程事务-子线程】事务提交");
                        //事务提交
                        platformTransactionManager.commit(transaction);
                    }
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
        }
    }

*  使用

import com.zbwd.outline.common.config.MultiThreadTransactionComponent;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.lang3.RandomUtils;
    import org.junit.jupiter.api.Test;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    import org.springframework.transaction.PlatformTransactionManager;
    
    /**
     * 多线程异步处理时的事务管理
     */
    @Slf4j
    @SpringBootTest
    class MultiThreadTransactionComponentTest {
        
    
        @Autowired
        PlatformTransactionManager platformTransactionManager;
    
        @Autowired
        private ThreadPoolTaskExecutor threadPoolTaskExecutor;
    
        @Test
        public void testTransaction() {
        
            // 定义此主线程的 异步处理时事务管理
            MultiThreadTransactionComponent mt = new MultiThreadTransactionComponent(platformTransactionManager, threadPoolTaskExecutor);
    
            for (int k = 0; k < 10; k++) {
        
                int i = RandomUtils.nextInt(0, 5);
                int y = RandomUtils.nextInt(0, 5);
                // 添加任务
                mt.addFunction(() -> {
        
                    System.out.println("当前线程：" + Thread.currentThread().getName());
                    System.out.println(i + "--" + y);
                    //除数为0时 执行失败
                    System.out.println(i % y);
                    return 0;
                });
            }
    
            // 执行任务
            mt.execute();
    
        }
    
    }