重新认识一下@Async，开启异步之路

绝地灬酷狼 2024-05-23 22:08 36阅读 0赞

###  ###

前言

异步调用几乎是处理高并发，解决性能问题常用的手段，如何开启异步调用？SpringBoot中提供了非常简单的方式，就是一个注解@Async。今天我们重新认识一下@Async,以及注意事项

###  ###

简单使用

新建三个作业任务

@Service public class TaskDemo \{ private static Logger logger = LoggerFactory.getLogger(TaskDemo.class); public void execute1() \{ logger.info("处理耗时任务1......开始"); try \{ TimeUnit.SECONDS.sleep(3); \} catch (InterruptedException e) \{ e.printStackTrace(); \} logger.info("处理耗时任务1......结束"); \} public void execute2() \{ logger.info("处理耗时任务2......开始"); try \{ TimeUnit.SECONDS.sleep(3); \} catch (InterruptedException e) \{ e.printStackTrace(); \} logger.info("处理耗时任务2......结束"); \} public void execute3() \{ logger.info("处理耗时任务3......开始"); try \{ TimeUnit.SECONDS.sleep(3); \} catch (InterruptedException e) \{ e.printStackTrace(); \} logger.info("处理耗时任务3......结束"); \} \}

测试代码

@RestController public class TaskController \{ @Autowired private TaskDemo taskDemo; @GetMapping("/task/test") public String testTask() \{ taskDemo.execute1(); taskDemo.execute2(); taskDemo.execute3(); return "ok"; \} \}

执行后我们可以发现，上面的代码是同一个线程的同步执行，整体耗时9秒才完成。

![d83130eb63769d184ef8d22ff4927cbd.png][]

###  ###

异步处理

springboot的异步，是非常简单的，加2个注解即可

@Service public class TaskDemo \{ private static Logger logger = LoggerFactory.getLogger(TaskDemo.class); @Async public void execute1() \{ logger.info("处理耗时任务1......开始"); try \{ TimeUnit.SECONDS.sleep(3); \} catch (InterruptedException e) \{ e.printStackTrace(); \} logger.info("处理耗时任务1......结束"); \} @Async public void execute2() \{ logger.info("处理耗时任务2......开始"); try \{ TimeUnit.SECONDS.sleep(3); \} catch (InterruptedException e) \{ e.printStackTrace(); \} logger.info("处理耗时任务2......结束"); \} @Async public void execute3() \{ logger.info("处理耗时任务3......开始"); try \{ TimeUnit.SECONDS.sleep(3); \} catch (InterruptedException e) \{ e.printStackTrace(); \} logger.info("处理耗时任务3......结束"); \} \}

@SpringBootApplication @EnableAsync public class DemoApp \{ public static void main(String\[\] args) \{ SpringApplication.run(DemoApp.class,args); \} \}

增加了@Async和@EnableAsync两个注解

![2c34ee852b0acf346f20889a17cfd1f1.png][]

从执行结果发现，整个流程用了3秒，以及用了3个线程执行。完成了异步调用

###  ###

异步回调

有些场景我们需要知道异步处理的任务什么时候完成，需要做额外的业务处理。如：我们需要在3个任务都完成后，提示一下给用户

@Service public class TaskDemo \{ private static Logger logger = LoggerFactory.getLogger(TaskDemo.class); @Async public Future<String> execute1() \{ logger.info("处理耗时任务1......开始"); try \{ TimeUnit.SECONDS.sleep(3); \} catch (InterruptedException e) \{ e.printStackTrace(); \} logger.info("处理耗时任务1......结束"); return new AsyncResult<>("任务1 ok"); \} @Async public Future<String> execute2() \{ logger.info("处理耗时任务2......开始"); try \{ TimeUnit.SECONDS.sleep(3); \} catch (InterruptedException e) \{ e.printStackTrace(); \} logger.info("处理耗时任务2......结束"); return new AsyncResult<>("任务2 ok"); \} @Async public Future<String> execute3() \{ logger.info("处理耗时任务3......开始"); try \{ TimeUnit.SECONDS.sleep(3); \} catch (InterruptedException e) \{ e.printStackTrace(); \} logger.info("处理耗时任务3......结束"); return new AsyncResult<>("任务3 ok"); \} \}

@RestController public class TaskController \{ private static Logger logger = LoggerFactory.getLogger(TaskController.class); @Autowired private TaskDemo taskDemo; @GetMapping("/task/test") public String testTask() throws InterruptedException \{ Future<String> task1 = taskDemo.execute1(); Future<String> task2 = taskDemo.execute2(); Future<String> task3 = taskDemo.execute3(); while (true)\{ if (task1.isDone() && task2.isDone() && task3.isDone())\{ break; \} TimeUnit.SECONDS.sleep(1); \} logger.info(">>>>>>3个任务都处理完成"); return "ok"; \} \}

![306f332408db3f4a9b94c09271fc802e.png][]

执行结果发现，在请求线程里面给用户提示了3个任务都处理完成了。

这段代码主要改变了什么

1、把具体任务返回类型改为了Future类型对象 2、在调用任务时，循环判断任务是否处理完

###  ###

自定义线程池

说到异步处理，一定要考虑到线程池，什么是线程池，小伙伴可自行网补。@Async的线程池定义比较方便，直接上代码

@Configuration public class ThreadPoolConfig \{ @Bean(name = "taskPool01Executor") public ThreadPoolTaskExecutor getTaskPool01Executor() \{ ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor(); //核心线程数 taskExecutor.setCorePoolSize(10); //线程池维护线程的最大数量,只有在缓冲队列满了之后才会申请超过核心线程数的线程 taskExecutor.setMaxPoolSize(100); //缓存队列 taskExecutor.setQueueCapacity(50); //许的空闲时间,当超过了核心线程出之外的线程在空闲时间到达之后会被销毁 taskExecutor.setKeepAliveSeconds(200); //异步方法内部线程名称 taskExecutor.setThreadNamePrefix("TaskPool-01-"); /\*\* \* 当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize，如果还有任务到来就会采取任务拒绝策略 \* 通常有以下四种策略： \* ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。 \* ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。 \* ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程） \* ThreadPoolExecutor.CallerRunsPolicy：重试添加当前的任务，自动重复调用 execute() 方法，直到成功 \*/ taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy()); taskExecutor.setWaitForTasksToCompleteOnShutdown(true); taskExecutor.initialize(); return taskExecutor; \} @Bean(name = "taskPool02Executor") public ThreadPoolTaskExecutor getTaskPool02Executor() \{ ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor(); //核心线程数 taskExecutor.setCorePoolSize(10); //线程池维护线程的最大数量,只有在缓冲队列满了之后才会申请超过核心线程数的线程 taskExecutor.setMaxPoolSize(100); //缓存队列 taskExecutor.setQueueCapacity(50); //许的空闲时间,当超过了核心线程出之外的线程在空闲时间到达之后会被销毁 taskExecutor.setKeepAliveSeconds(200); //异步方法内部线程名称 taskExecutor.setThreadNamePrefix("TaskPool-02-"); /\*\* \* 当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize，如果还有任务到来就会采取任务拒绝策略 \* 通常有以下四种策略： \* ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。 \* ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。 \* ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程） \* ThreadPoolExecutor.CallerRunsPolicy：重试添加当前的任务，自动重复调用 execute() 方法，直到成功 \*/ taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy()); taskExecutor.setWaitForTasksToCompleteOnShutdown(true); taskExecutor.initialize(); return taskExecutor; \} \}

定义了2个线程池Bean

@Service public class TaskDemo \{ private static Logger logger = LoggerFactory.getLogger(TaskDemo.class); @Async("taskPool01Executor") public Future<String> execute1() \{ logger.info("处理耗时任务1......开始"); try \{ TimeUnit.SECONDS.sleep(3); \} catch (InterruptedException e) \{ e.printStackTrace(); \} logger.info("处理耗时任务1......结束"); return new AsyncResult<>("任务1 ok"); \} @Async("taskPool01Executor") public Future<String> execute2() \{ logger.info("处理耗时任务2......开始"); try \{ TimeUnit.SECONDS.sleep(3); \} catch (InterruptedException e) \{ e.printStackTrace(); \} logger.info("处理耗时任务2......结束"); return new AsyncResult<>("任务2 ok"); \} @Async("taskPool02Executor") public Future<String> execute3() \{ logger.info("处理耗时任务3......开始"); try \{ TimeUnit.SECONDS.sleep(3); \} catch (InterruptedException e) \{ e.printStackTrace(); \} logger.info("处理耗时任务3......结束"); return new AsyncResult<>("任务3 ok"); \} \}

@Async(“线程池名称”)，指定value使用自己定义的线程池

![9b41a73489394e0b3954550554bbead5.png][]

执行结果利用了线程池

[d83130eb63769d184ef8d22ff4927cbd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/e720d208183b44d380c287ed651a69e5.png
[2c34ee852b0acf346f20889a17cfd1f1.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/e8d51d8dda984f55809a9f37b9ca08fc.png
[306f332408db3f4a9b94c09271fc802e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/4af10a809d35447a8c6fde48759b15c4.png
[9b41a73489394e0b3954550554bbead5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/23/1055933b549b4f5c927a8739a81f424a.png