Springboot 多线程实现事务控制

小咪咪 2024-03-22 10:32 75阅读 0赞

### 业务背景 ###

工作中有很多，在一个service实现层，有可能调用其他的service，或者说是使用异步线程，这样就不容易控制当前操作成功和失败  
不多说，上代码

#### 获取数据库事务控制器 ####

@Component
    public class TransactionalUntil {
        
    
        @Resource
        private DataSourceTransactionManager dataSourceTransactionManager;
    
        /**
         * 开启事务
         */
        public TransactionStatus begin() {
        
            TransactionStatus transaction = dataSourceTransactionManager.getTransaction(new DefaultTransactionAttribute());
            return transaction;
        }
    
        /**
         * 提交事务
         */
        public void commit(TransactionStatus transactionStatus) {
        
            dataSourceTransactionManager.commit(transactionStatus);
        }
    
        /**
         * 回滚事务
         */
        public void rollback(TransactionStatus transactionStatus) {
        
            dataSourceTransactionManager.rollback(transactionStatus);
        }
    
    }

#### 测试代码 ####

package com.demo.easypdf;
    
    import cn.hutool.core.thread.ThreadUtil;
    import com.beust.jcommander.internal.Lists;
    import com.demo.easypdf.domain.TestOne;
    import com.demo.easypdf.domain.TestT;
    import com.demo.easypdf.domain.User;
    import com.demo.easypdf.mapper.TestOneMapper;
    import com.demo.easypdf.mapper.TestTMapper;
    import com.demo.easypdf.mapper.UserMapper;
    import com.demo.easypdf.untils.TransactionalUntil;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.transaction.TransactionStatus;
    import org.springframework.transaction.annotation.Transactional;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    import wiki.xsx.core.pdf.component.text.XEasyPdfText;
    import wiki.xsx.core.pdf.doc.XEasyPdfDocument;
    import wiki.xsx.core.pdf.doc.XEasyPdfPage;
    import wiki.xsx.core.pdf.handler.XEasyPdfHandler;
    import wiki.xsx.core.pdf.template.doc.XEasyPdfTemplateDocument;
    import wiki.xsx.core.pdf.template.doc.component.block.XEasyPdfTemplateBlockContainer;
    import wiki.xsx.core.pdf.template.doc.component.image.XEasyPdfTemplateImage;
    import wiki.xsx.core.pdf.template.doc.component.page.XEasyPdfTemplateCurrentPageNumber;
    import wiki.xsx.core.pdf.template.doc.component.page.XEasyPdfTemplateTotalPageNumber;
    import wiki.xsx.core.pdf.template.doc.component.text.XEasyPdfTemplateText;
    import wiki.xsx.core.pdf.template.doc.page.XEasyPdfTemplatePage;
    import wiki.xsx.core.pdf.template.handler.XEasyPdfTemplateHandler;
    
    import javax.annotation.Resource;
    import java.util.Collections;
    import java.util.List;
    import java.util.concurrent.CopyOnWriteArrayList;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.atomic.AtomicBoolean;
    
    /**
     * @Author: lkz
     * @Title: TestController
     * @Description: TODO
     * @Date: 2023/4/27 23:17
     */
    @RestController
    @Slf4j
    public class TestController {
        
    
        @Resource
        private UserMapper userMapper;
        @Resource
        private TestOneMapper testOneMapper;
        @Resource
        private TestTMapper testTMapper;
    
        private volatile  Boolean is_ok=new Boolean(true);
    
        @Resource
        private TransactionalUntil transactionalUntil;
    
        @GetMapping("test2")
        public void testThreadTran(){
        
            // 监控子线程数据
            CountDownLatch childMonitor = new CountDownLatch(2);
            // 主线程收集子线程运行最终结果
            List<Boolean> childResponse= new CopyOnWriteArrayList<>();
            //子线程在该对象上等待主线程的通知
            CountDownLatch mainMonitor = new CountDownLatch(1);
    
            ThreadUtil.execAsync(()->{
        
    
                // 开启事务
                TransactionStatus transactionStatus = transactionalUntil.begin();
                try {
        
                    // 操作数据库逻辑
                    List<User> user = this.getUserList();
                    userMapper.insert(user.get(0));
                    childResponse.add(Boolean.TRUE);
                    childMonitor.countDown();
    
                    mainMonitor.await();
                    log.info("user-线程{}正常执行结束,等待其他线程执行结束",Thread.currentThread().getName());
                    if(is_ok){
        
                        // 事务提交
                        log.info("user-线程正常执行,线程事务提交",Thread.currentThread().getName());
                        transactionalUntil.commit(transactionStatus);
                    }else{
        
                        // 事务回滚
                        log.info("user-线程执行出现异常,线程事务回滚",Thread.currentThread().getName());
                        transactionalUntil.rollback(transactionStatus);
                    }
                }catch (Exception e){
        
                    // 提交失败
                    log.info("user-线程执行出现异常{},",Thread.currentThread().getName());
                    childMonitor.countDown();
                    childResponse.add(Boolean.FALSE);
                    transactionalUntil.rollback(transactionStatus);
                }
    
            });
    
            ThreadUtil.execAsync(()->{
        
                // 开启事务
                TransactionStatus transactionStatus = transactionalUntil.begin();
                try {
        
                    // 操作数据库逻辑
                    List<TestT> testTList = this.getTestTList();
                    testTMapper.insert(testTList.get(0));
                    log.info("test-线程{}正常执行结束,等待其他线程执行结束",Thread.currentThread().getName());
                    // 模拟异常
                  //  int a=10/0;
                    childResponse.add(Boolean.TRUE);
                    childMonitor.countDown();
                    mainMonitor.await();
                    if(is_ok){
        
                        // 事务提交
                        log.info("test-线程正常执行,线程事务提交",Thread.currentThread().getName());
                        transactionalUntil.commit(transactionStatus);
                    }else{
        
                        // 事务回滚
                        log.info("test-线程执行出现异常,线程事务回滚",Thread.currentThread().getName());
                        transactionalUntil.rollback(transactionStatus);
                    }
                }catch (Exception e){
        
                    // 提交失败
                    log.info("test-线程执行出现异常{},",Thread.currentThread().getName());
                    childMonitor.countDown();
                    childResponse.add(Boolean.FALSE);
                    transactionalUntil.rollback(transactionStatus);
                }
            });
    
            try {
        
                childMonitor.await();
                for(Boolean res:childResponse){
        
                    while (!res){
        
                        // 如果有一个子线程只想失败，改变mainResult状态 事务回滚
                        log.info("有线程执行失败,修改标识位,事务回滚");
                        is_ok=false;
                        break;
                    }
                }
                //主线程获取结果，子线程根据主线程的结果 提交或回滚
                mainMonitor.countDown();
                System.out.println("执行结束-------");
            }catch (Exception e){
        
                log.info("事务执行失败");
            }
    
        }
    
    
        public List<User> getUserList(){
        
            List<User> users= Lists.newArrayList();
            for(int i=0;i<20;i++){
        
                User user = new User();
                user.setId(i);
                user.setUsername("小"+i);
                user.setPassword("asc");
                user.setAge(i+0);
                users.add(user);
            }
            return users;
        }
    
        public List<TestT> getTestTList(){
        
            List<TestT> TestT= Lists.newArrayList();
            for(int i=0;i<20;i++){
        
                TestT tt = new TestT();
                tt.setPhone("i"+"-1234537");
                tt.setId(i);
                TestT.add(tt);
            }
            return TestT;
        }
        public List<TestOne> getTestOneList(){
        
            List<TestOne> oneList= Lists.newArrayList();
            for(int i=0;i<20;i++){
        
                TestOne tt = new TestOne();
                tt.setName("i"+"-1234537");
                oneList.add(tt);
            }
            return oneList;
        }
    
    }

#### 测试结果 ####

##### 没有异常 #####

表1  
![在这里插入图片描述][95d423d329944647bba793c0f8d0b125.png]  
表二  
![在这里插入图片描述][6cb0c7e20b9c47bb9f20866a103bd0a8.png]  
控制台打印  
![在这里插入图片描述][afd62c28cefe4b80a8ebec72fe7ae36f.png]

##### 模拟异常 #####

控制台  
![在这里插入图片描述][ab8e94a7aad945f0ac3483ca516382a2.png]  
数据库无新增  
![在这里插入图片描述][238fdcc1d4814e019474cbad88e797da.png]

[95d423d329944647bba793c0f8d0b125.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/db12aa5b0249486bb2f4ac8aac93f412.png
[6cb0c7e20b9c47bb9f20866a103bd0a8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/0799b5c4578e4dcdb3fb98d10ec4a679.png
[afd62c28cefe4b80a8ebec72fe7ae36f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/0f42a5adc53f4626b195feffa968443a.png
[ab8e94a7aad945f0ac3483ca516382a2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/694d676cff834a6b99f85f16a4190a45.png
[238fdcc1d4814e019474cbad88e797da.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/22/c9a5aa76233c480d91c4783c2594bb9b.png