springboot中解决事务中调用异步Async方法

柔光的暖阳◎ 2022-09-10 13:28 743阅读 0赞

**解决思路**  
在数据添加的时候同时上传文件，采用异步进行文件服务器存储实现，问题是不能保证文件一定可以上传成功。  
异步方法其实是多线程执行任务，需要做的是进行阻塞。

1.  创建线程池并注入

@Configuration
    @Slf4j
    public class ThreadPoolConfig {
    
        @Value("${asyncThreadPool.corePoolSize:10}")
        private int corePoolSize;
    
        @Value("${asyncThreadPool.maxPoolSize:20}")
        private int maxPoolSize;
    
        @Value("${asyncThreadPool.queueCapacity:20}")
        private int queueCapacity;
    
        @Value("${asyncThreadPool.keepAliveSeconds:3}")
        private int keepAliveSeconds;
    
        @Value("${asyncThreadPool.awaitTerminationSeconds:5}")
        private int awaitTerminationSeconds;
    
        @Value("${asyncThreadPool.threadNamePrefix:thread}")
        private String threadNamePrefix;
    
        /**
         * 线程池配置
         * @param
         * @return java.util.concurrent.Executor
         * @author wliduo
         * @date 2019/2/15 14:44
         */
        @Bean(name = "threadPoolTaskExecutor")
        public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
            log.info("---------- 线程池开始加载 ----------");
            ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
            // 核心线程池大小
            threadPoolTaskExecutor.setCorePoolSize(corePoolSize);
            // 最大线程数
            threadPoolTaskExecutor.setMaxPoolSize(maxPoolSize);
            // 队列容量----LinkedBlockingQueue 默认使用无边界的阻塞队列   ArrayBlockingQueue 有边界队列
            threadPoolTaskExecutor.setQueueCapacity(keepAliveSeconds);
            // 活跃时间---
            threadPoolTaskExecutor.setKeepAliveSeconds(queueCapacity);
            // 主线程等待子线程执行时间---也有可能是拒绝策略
            threadPoolTaskExecutor.setAwaitTerminationSeconds(awaitTerminationSeconds);
            // 线程名字前缀
            threadPoolTaskExecutor.setThreadNamePrefix(threadNamePrefix);
            // RejectedExecutionHandler:当pool已经达到max-size的时候，如何处理新任务
            // CallerRunsPolicy:不在新线程中执行任务，而是由调用者所在的线程来执行
            threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            // 初始化
            threadPoolTaskExecutor.initialize();
            log.info("---------- 线程池加载完成 ----------");
            return threadPoolTaskExecutor;
        }
    
    }

1.  文件存储功能实现（异步注解）

@Repository
    public class OssMybatisReposity implements OssReposity {
    
        @Autowired(required=false)
        private OssHolder ossHolder;
    
        @Value("${oss.bucketName:zjyh-yygc}")
        private String bucketName;
    
        @Override
        @Async("threadPoolTaskExecutor")
        public Future<String> upload(MultipartFile file, String attachmentKey) {
            int i = 1/0;
            String result = null;
            //自动关闭，防止oom情况出现
            try(InputStream inputStream = file.getInputStream()) {
                result = ossHolder.uploadFile(bucketName, inputStream, attachmentKey);
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println("upload-----------------------");
            return new AsyncResult<>(result);
        }
    }

1.  get阻塞，异常处理。  
    如果不采用get，则不会发生阻塞。只需要建立异常库进行记录即可

@Transactional(rollbackFor = Exception.class,propagation = Propagation.REQUIRED)
        public AttachmentDTO saveAttachment(MultipartFile multipartFiles,AttachmentDTO attachmentDTO) {
            try {
                String attachmentName = multipartFiles.getOriginalFilename();
                String attachmentKey = StringUtils.getFileKey(attachmentName);
                String attachmentType =  attachmentName.substring(attachmentName.lastIndexOf("."));
                InvocationUserInfo userInfo = LocalInvocationContext.getContext();
                String uploadStr = ossReposity.upload(multipartFiles,attachmentKey).get();
                System.out.println("saveAttachment--------------------");
                if (Objects.nonNull(userInfo)) {
                    attachmentDTO.setCreator(userInfo.getEmployeeName());
                }
                attachmentDTO.setAttachmentType(attachmentType);
                attachmentDTO.setAttachmentKey(attachmentKey);
                attachmentDTO.setAttachmentName(attachmentName);
            }catch (Exception e) {
                throw new BusinessException("新增文件失败");
            }
            Attachment attachment = new Attachment();
            BeanUtils.copyProperties(attachmentDTO, attachment);
            attachment = attachmentDomainService.createAttachment(attachment);
            AttachmentDTO newAttachmentDTO = new AttachmentDTO();
            BeanUtils.copyProperties(attachment, newAttachmentDTO);
            return newAttachmentDTO;
        }

1.  异常信息入库

@Configuration
    @EnableAsync
    @Slf4j
    public class AsyncConfig implements AsyncConfigurer {
    
    
    
        /**
         * 只能捕获无返回值的异步方法，有返回值的被主线程处理
         */
        @Override
        public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
            return new CustomAsyncExceptionHandler();
        }
    
        /***
         * 处理异步方法中未捕获的异常
         */
        class CustomAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {
    
            @Override
            public void handleUncaughtException(Throwable throwable, Method method, Object... obj) {
                log.info("Exception message - {}", throwable.getMessage());
                log.info("Method name - {}", method.getName());
                log.info("Parameter values - {}", Arrays.toString(obj));
                if (throwable instanceof Exception) {
                   入库；
                }
                throwable.printStackTrace();
            }
        }
    }