springboot中解决事务中调用异步Async方法-蒲公英云

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

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

创建线程池并注入

@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;
}

}

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

@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);
}

}

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;
}

异常信息入库

@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();
        }
    }
}