# SpringBoot使用@Async异步注解

在项目中,当访问其他人的接口较慢或者做耗时任务时,不想程序一直卡在耗时任务上,想程序能够并行执行,我们可以使用多线程来并行的处理任务,这里介绍下 SpringBoot 下的 @Async 注解,还有 ApplicationEventPublisher 可以了解下

代码地址

# 1. Config

需要一个注解 @EnableAsync 开启 @Async 的功能,SpringBoot 可以放在 Application 上,也可以放其他配置文件上

@EnableAsync
@SpringBootApplication
public class Application { }

@Async 配置有两个,一个是执行的线程池,一个是异常处理

执行的线程池默认情况下找唯一的 org.springframework.core.task.TaskExecutor,或者一个 Bean 的 Name 为 taskExecutor 的 java.util.concurrent.Executor 作为执行任务的线程池。如果都没有的话,会创建 SimpleAsyncTaskExecutor 线程池来处理异步方法调用,当然 @Async 注解支持一个 String 参数,来指定一个 Bean 的 Name,类型是 Executor 或 TaskExecutor,表示使用这个指定的线程池来执行这个异步任务

异常处理,@Async 标记的方法只能是 void 或者 Future 返回值,在无返回值的异步调用中,异步处理抛出异常,默认是SimpleAsyncUncaughtExceptionHandler 的 handleUncaughtException() 会捕获指定异常,只是简单的输出了错误日志(一般需要自定义配置异常处理),原有任务还会继续运行,直到结束(具有 void 返回类型的方法不能将任何异常发送回调用者,默认情况下此类未捕获异常只会输出错误日志),而在有返回值的异步调用中,异步处理抛出了异常,会直接返回主线程处理,异步任务结束执行,主线程也会被异步方法中的异常中断结束执行

@Async有两个使用的限制

  1. 它必须仅适用于 public 方法
  2. 在同一个类中调用异步方法将无法正常工作(self-invocation)
/**
 * AsyncConfig
 *
 * @author wliduo[i@dolyw.com]
 * @date 2020/5/19 17:58
 */
@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {

    /**
     * logger
     */
    private static final Logger logger = LoggerFactory.getLogger(AsyncConfig.class);

    /**
     * 这里不实现了,使用 ThreadPoolConfig 里的线程池即可
     *
     * @param
     * @return java.util.concurrent.Executor
     * @throws
     * @author wliduo[i@dolyw.com]
     * @date 2020/5/19 18:00
     */
    /*@Override
    public Executor getAsyncExecutor() {
        return null;
    }*/

    /**
     * 只能捕获无返回值的异步方法,有返回值的被主线程处理
     *
     * @param
     * @return org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler
     * @throws
     * @author wliduo[i@dolyw.com]
     * @date 2020/5/20 10:16
     */
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return new CustomAsyncExceptionHandler();
    }

    /***
     * 处理异步方法中未捕获的异常
     *
     * @author wliduo[i@dolyw.com]
     * @date 2020/5/19 19:16
     */
    class CustomAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {

        @Override
        public void handleUncaughtException(Throwable throwable, Method method, Object... obj) {
            logger.info("Exception message - {}", throwable.getMessage());
            logger.info("Method name - {}", method.getName());
            logger.info("Parameter values - {}", Arrays.toString(obj));
            if (throwable instanceof Exception) {
                Exception exception = (Exception) throwable;
                logger.info("exception:{}", exception.getMessage());
            }
            throwable.printStackTrace();
        }
    }
}
/**
 * 线程池配置
 *
 * @author wliduo
 * @date 2019/2/15 14:36
 */
@Configuration
public class ThreadPoolConfig {

    /**
     * logger
     */
    private final static Logger logger = LoggerFactory.getLogger(ThreadPoolConfig.class);

    @Value("${asyncThreadPool.corePoolSize}")
    private int corePoolSize;

    @Value("${asyncThreadPool.maxPoolSize}")
    private int maxPoolSize;

    @Value("${asyncThreadPool.queueCapacity}")
    private int queueCapacity;

    @Value("${asyncThreadPool.keepAliveSeconds}")
    private int keepAliveSeconds;

    @Value("${asyncThreadPool.awaitTerminationSeconds}")
    private int awaitTerminationSeconds;

    @Value("${asyncThreadPool.threadNamePrefix}")
    private String threadNamePrefix;

    /**
     * 线程池配置
     * @param
     * @return java.util.concurrent.Executor
     * @author wliduo
     * @date 2019/2/15 14:44
     */
    @Bean(name = "threadPoolTaskExecutor")
    public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
        logger.info("---------- 线程池开始加载 ----------");
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        // 核心线程池大小
        threadPoolTaskExecutor.setCorePoolSize(corePoolSize);
        // 最大线程数
        threadPoolTaskExecutor.setMaxPoolSize(maxPoolSize);
        // 队列容量
        threadPoolTaskExecutor.setQueueCapacity(queueCapacity);
        // 活跃时间
        threadPoolTaskExecutor.setKeepAliveSeconds(keepAliveSeconds);
        // 主线程等待子线程执行时间
        threadPoolTaskExecutor.setAwaitTerminationSeconds(awaitTerminationSeconds);
        // 线程名字前缀
        threadPoolTaskExecutor.setThreadNamePrefix(threadNamePrefix);
        // RejectedExecutionHandler:当pool已经达到max-size的时候,如何处理新任务
        // CallerRunsPolicy:不在新线程中执行任务,而是由调用者所在的线程来执行
        threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 初始化
        threadPoolTaskExecutor.initialize();
        logger.info("---------- 线程池加载完成 ----------");
        return threadPoolTaskExecutor;
    }
}

# 2. Code

简单的使用,异步异常处理,Service 方法异步,多个 Service 方法异步,工具类异步

# 2.1. Controller

/**
 * AsyncController
 *
 * @author wliduo[i@dolyw.com]
 * @date 2020/5/19 14:46
 */
@RestController
@RequestMapping("/async")
public class AsyncController {

    /**
     * logger
     */
    private final static Logger logger = LoggerFactory.getLogger(AsyncController.class);

    @Autowired
    private AsyncService asyncService;

    @Autowired
    private SmsUtil smsUtil;

    /**
     * 可以看到无返回值异步方法出现异常,主线程还是继续执行完成
     *
     * @param
     * @return void
     * @throws
     * @author wliduo[i@dolyw.com]
     * @date 2020/5/20 9:53
     */
    @GetMapping("/run1")
    public String run1() throws Exception {
        asyncService.task1();
        logger.info("run1开始执行");
        Thread.sleep(5000);
        logger.info("run1执行完成");
        return "run1 success";
    }

    /**
     * 可以看到有返回值异步方法出现异常,异常抛给主线程处理,导致主线程也被中断执行
     *
     * @param
     * @return java.lang.String
     * @throws
     * @author wliduo[i@dolyw.com]
     * @date 2020/5/20 10:15
     */
    @GetMapping("/run2")
    public String run2() throws Exception {
        Future<String> future = asyncService.task2();
        // get()方法阻塞主线程,直到执行完成
        String result = future.get();
        logger.info("run2开始执行");
        Thread.sleep(5000);
        logger.info("run2执行完成");
        return result;
    }

    /**
     * 多个异步执行
     *
     * @param
     * @return java.lang.String
     * @throws
     * @author wliduo[i@dolyw.com]
     * @date 2020/5/20 10:26
     */
    @GetMapping("/run3")
    public String run3() throws Exception {
        logger.info("run3开始执行");
        long start = System.currentTimeMillis();
        Future<String> future3 = asyncService.task3();
        Future<String> future4 = asyncService.task4();
        // 这样与下面是一样的
        logger.info(future3.get());
        logger.info(future4.get());
        // 先判断是否执行完成
        boolean run3Done = Boolean.FALSE;
        while (true) {
            if (future3.isDone() && future4.isDone()) {
                // 执行完成
                run3Done = Boolean.TRUE;
                break;
            }
            if (future3.isCancelled() || future4.isCancelled()) {
                // 取消情况
                break;
            }
        }
        if (run3Done) {
            logger.info(future3.get());
            logger.info(future4.get());
        } else {
            // 其他异常情况
        }
        long end = System.currentTimeMillis();
        logger.info("run3执行完成,执行时间: {}", end - start);
        return "run3 success";
    }

    /**
     * 工具类异步
     *
     * @param
     * @return java.lang.String
     * @throws
     * @author wliduo[i@dolyw.com]
     * @date 2020/5/20 10:59
     */
    @GetMapping("/sms")
    public String sms() throws Exception {
        logger.info("run1开始执行");
        smsUtil.sendCode("15912347896", "135333");
        logger.info("run1执行完成");
        return "send sms success";
    }
}

# 2.2. ServiceImpl

/**
 * AsyncServiceImpl
 *
 * @author wliduo[i@dolyw.com]
 * @date 2020/5/19 14:24
 */
@Service
public class AsyncServiceImpl implements AsyncService {

    /**
     * logger
     */
    private final static Logger logger = LoggerFactory.getLogger(AsyncServiceImpl.class);

    @Override
    @Async("threadPoolTaskExecutor")
    public void task1() throws Exception {
        logger.info("task1开始执行");
        Thread.sleep(3000);
        logger.info("task1执行结束");
        throw new RuntimeException("出现异常");
    }

    @Override
    @Async("threadPoolTaskExecutor")
    public Future<String> task2() throws Exception {
        logger.info("task2开始执行");
        Thread.sleep(3000);
        logger.info("task2执行结束");
        throw new RuntimeException("出现异常");
        // return new AsyncResult<String>("task2 success");
    }

    @Override
    @Async("threadPoolTaskExecutor")
    public Future<String> task3() throws Exception {
        logger.info("task3开始执行");
        Thread.sleep(3000);
        logger.info("task3执行结束");
        return new AsyncResult<String>("task3 success");
    }

    @Override
    @Async("threadPoolTaskExecutor")
    public Future<String> task4() throws Exception {
        logger.info("task4开始执行");
        Thread.sleep(3000);
        logger.info("task4执行结束");
        return new AsyncResult<String>("task4 success");
    }
}

# 2.3. SmsUtil

/**
 * SmsUtil
 *
 * @author wliduo[i@dolyw.com]
 * @date 2020/5/20 10:50
 */
@Component
public class SmsUtil {

    private static final Logger logger = LoggerFactory.getLogger(SmsUtil.class);

    /**
     * 异步发送短信
     *
     * @param phone
	 * @param code
     * @return void
     * @throws
     * @author wliduo[i@dolyw.com]
     * @date 2020/5/20 10:53
     */
    @Async
    public void sendCode(String phone, String code) {
        logger.info("开始发送验证码...");
        // 模拟调用接口发验证码的耗时
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        logger.info("发送成功: {}", phone);
    }
    
}

# 3. 示例

Future 和 CompletableFuture 两种示例

# 3.1. Future

@Test
public R<?> getDataTest() throws Exception {
    // 并行异步线程执行
    List<Future<Resp>> futureList = new ArrayList<>();
    for (Cont cont : contList) {
        Future<Resp> future = testUtil.find(req);
        futureList.add(future);
    }
    List<Resp> respList = new ArrayList<>();
    // 等待执行完成
    for (Future<Resp> future : futureList) {
        while (true) {
            // CPU高速轮询,每个future都并发轮循,判断完成状态然后获取结果
            // 即有10个future在高速轮询,完成一个future的获取结果,就关闭一个轮询
            if (future.isDone() && !future.isCancelled()) {
                // 获取future成功完成状态,如果想要限制每个任务的超时时间
                // 取消本行的状态判断 + future.get(1000 * 2, TimeUnit.MILLISECONDS) + catch超时异常使用即可
                respList.add(future.get());
                // 当前future获取结果完毕,跳出while
                break;
            } else {
                // 每次轮询休息1毫秒(CPU纳秒级),避免CPU高速轮循耗空CPU
                Thread.sleep(1);
            }
        }
    }
    // 获取结果处理
    for (Resp resp : respList) {
        // XXX
    }
    return ResponseMessage.success(respList);
}
@Async("threadPoolTaskExecutor")
public Future<Resp> find(Req req) {
    // 开始发起请求
    OkHttpClient okHttpClient = new OkHttpClient.Builder().build();
    // 请求地址,请求头和参数
    String param = JSON.toJSONStringWithDateFormat(req, JSON.DEFFAULT_DATE_FORMAT, SerializerFeature.WriteDateUseDateFormat);
    RequestBody requestBody = RequestBody.create(MediaType.parse("application/json; charset=utf-8"), param);
    Request request = new Request.Builder().url(baseUrl + XXXUrl)
            .addHeader("token", token)
            .post(requestBody).build();
    log.info("请求地址:{},请求参数:{}", baseUrl + XXXUrl, param);
    // 返回结果
    String result = "";
    // 发起请求
    try (Response response = okHttpClient.newCall(request).execute()) {
        // 获取结果
        result = response.body().string();
        log.info("请求结果:{},请求地址:{},请求参数:{}", result, baseUrl + XXXUrl, param);
        return new AsyncResult<Resp>(JSON.parseObject(result, Resp.class));
    } catch (SocketTimeoutException e) {
        log.error("请求超时异常,请求地址:{},请求参数:{}", baseUrl + XXXUrl, param);
        return new AsyncResult<Resp>(null);
    } catch (Exception e) {
        log.error("请求出现异常,请求结果:{},错误信息: {}", result, ExceptionUtil.getExceptionMessage(e));
        return new AsyncResult<Resp>(null);
    }
}

# 3.2. CompletableFuture

  • 异步线程无返回参数
// 数据入库,将数据按一次处理上限拆分为多个List批量处理
List<List<ImportDataDto>> lists = ListUtil.split(importDataList, 1500);
// 新建多线程任务List
List<CompletableFuture> completableFutureList = new ArrayList<>();
for (List<ImportDataDto> list : lists) {
    // 添加到多线程任务
    completableFutureList.add(CompletableFuture.runAsync(() -> {
        long timeBegin = System.currentTimeMillis();
        int count = this.saveImportDate(list);
        long timeEnd = System.currentTimeMillis();
        logger.info("处理条数:{},服务响应时间:{}秒", count, (timeEnd - timeBegin) / 1000.0D);
    }, threadPoolTaskExecutor));
}
// 等待所有线程执行完成
CompletableFuture.allOf(completableFutureList.toArray(new CompletableFuture[]{})).join();
  • 异步线程有返回参数
// 数据入库计时
long sumTimeBegin = System.currentTimeMillis();
// 数据入库,将数据按一次处理上限拆分为多个List批量处理
List<List<ImportDataDto>> lists = ListUtil.split(importDataList, 3000);
// 新建多线程任务List
List<CompletableFuture<Integer>> completableFutureList = new ArrayList<>();
for (List<ImportDataDto> list : lists) {
    // 添加到多线程任务
    completableFutureList.add(CompletableFuture.supplyAsync(() -> {
        return this.saveImportDate(list, importReq);
    }, threadPoolTaskExecutor));
}
// 等待所有线程执行完成
CompletableFuture.allOf(completableFutureList.toArray(new CompletableFuture[]{})).join();
// 获取结果集
Integer sumCount = 0;
for (CompletableFuture<Integer> completableFuture : completableFutureList) {
    sumCount = sumCount + completableFuture.get();
}
long sumTimeEnd = System.currentTimeMillis();
logger.info("批次号:{},总处理条数:{},总数据入库响应时间:{}秒", importReq.getBatchNo(), sumCount, (sumTimeEnd - sumTimeBegin) / 1000.0D);
  • 异步线程处理
public Integer saveImportDate(List<ImportDataDto> list, ImportReq importReq) {
    long timeBegin = System.currentTimeMillis();
    // 当前批次数据段处理条数
    Integer count = list.size();
    SqlSession sqlSession = sqlSessionTemplate.getSqlSessionFactory().openSession(ExecutorType.BATCH,false);
    ImportDataDto importDataDto = list.get(0);
    try {
        ImportDataDao1 importDataDao1 = sqlSession.getMapper(ImportDataDao1.class);
        ImportDataDao2 importDataDao2 = sqlSession.getMapper(ImportDataDao2.class);
        ImportDataDao3 importDataDao3 = sqlSession.getMapper(ImportDataDao3.class);
        // 判断数据不为空就插入数据库
        if (importDataDto.getImportDataDto1() != null) {
            importDataDao1.insertBatch(list);
        }
        if (importDataDto.getImportDataDto2() != null) {
            importDataDao2.insertBatch(list);
        }
        if (importDataDto.getImportDataDto3() != null) {
            importDataDao3.insertBatch(list);
        }
        // 提交事务
        sqlSession.commit();
        sqlSession.clearCache();
    } catch (Exception e) {
        // sqlSession.rollback();
        // 出现异常
        Integer serialNo = importDataDto.getImportDataDto1().getSerialNo();
        logger.info("批次号:{},数据段处理出现异常,序号区间:{}-{}", importReq.getBatchNo(), serialNo, serialNo + count - 1);
        count = 0;
    } finally {
        sqlSession.close();
    }
    long timeEnd = System.currentTimeMillis();
    logger.info("批次号:{},单线程处理条数:{},数据入库响应时间:{}秒", importReq.getBatchNo(), count, (timeEnd - timeBegin) / 1000.0D);
    return count;
}

参考

上次更新时间: 2023-12-15 03:14:55