Java SpringBoot

为什么要用异步框架,它解决什么问题?

在SpringBoot的日常开发中,一般都是同步调用的。但实际中有很多场景非常适合使用异步来处理,如:注册新用户,送100个积分;或下单成功,发送push消息等等。
就拿注册新用户这个用例来说,为什么要异步处理?

  • 第一个原因:容错性、健壮性,如果送积分出现异常,不能因为送积分而导致用户注册失败;因为用户注册是主要功能,送积分是次要功能,即使送积分异常也要提示用户注册成功,然后后面在针对积分异常做补偿处理。
  • 第二个原因:提升性能,例如注册用户花了20毫秒,送积分花费50毫秒,如果用同步的话,总耗时70毫秒,用异步的话,无需等待积分,故耗时20毫秒。

故,异步能解决2个问题,性能和容错性。

SpringBoot如何实现异步调用?

对于异步方法调用,从Spring3开始提供了@Async注解,只需要在方法上标注此注解,此方法即可实现异步调用。
当然,还需要一个配置类,通过Enable模块驱动注解@EnableAsync 来开启异步功能。

实现异步调用

第一步:新建配置类,开启@Async功能支持

使用@EnableAsync来开启异步任务支持,@EnableAsync注解可以直接放在SpringBoot启动类上,也可以单独放在其他配置类上。这里选择使用单独的配置类SyncConfiguration

  1. @Configuration
  2. @EnableAsync
  3. public class AsyncConfiguration {
  4. }

第二步:在方法上标记异步调用

增加一个Component类,用来进行业务处理,同时添加@Async注解,代表该方法为异步处理。

  1. @Component
  2. @Slf4j
  3. public class AsyncTask {
  4. @SneakyThrows
  5. @Async
  6. public void doTask1() {
  7. long t1 = System.currentTimeMillis();
  8. Thread.sleep(2000);
  9. long t2 = System.currentTimeMillis();
  10. log.info("task1 cost {} ms" , t2-t1);
  11. }
  12. @SneakyThrows
  13. @Async
  14. public void doTask2() {
  15. long t1 = System.currentTimeMillis();
  16. Thread.sleep(3000);
  17. long t2 = System.currentTimeMillis();
  18. log.info("task2 cost {} ms" , t2-t1);
  19. }
  20. }

第三步:在Controller中进行异步方法调用

  1. @RestController
  2. @RequestMapping("/async")
  3. @Slf4j
  4. public class AsyncController {
  5. @Autowired
  6. private AsyncTask asyncTask;
  7. @RequestMapping("/task")
  8. public void task() throws InterruptedException {
  9. long t1 = System.currentTimeMillis();
  10. asyncTask.doTask1();
  11. asyncTask.doTask2();
  12. Thread.sleep(1000);
  13. long t2 = System.currentTimeMillis();
  14. log.info("main cost {} ms", t2-t1);
  15. }
  16. }

通过访问http://localhost:8080/async/task查看控制台日志:

  1. 2021-11-25 15:48:37 [http-nio-8080-exec-8] INFO com.jianzh5.blog.async.AsyncController:26 - main cost 1009 ms
  2. 2021-11-25 15:48:38 [task-1] INFO com.jianzh5.blog.async.AsyncTask:22 - task1 cost 2005 ms
  3. 2021-11-25 15:48:39 [task-2] INFO com.jianzh5.blog.async.AsyncTask:31 - task2 cost 3005 ms

通过日志可以看到:主线程不需要等待异步方法执行完成,减少了响应时间,提高了接口性能。
通过上面三步就可以在SpringBoot中欢乐的使用异步方法来提高接口性能了,是不是很简单?
但是,上面的代码忽略了一个最大的问题,就是给@Async异步框架自定义线程池。

为什么要给@Async自定义线程池?

使用@Async注解,在默认情况下用的是SimpleAsyncTaskExecutor线程池,该线程池不是真正意义上的线程池。
使用此线程池无法实现线程重用,每次调用都会新建一条线程。若系统中不断的创建线程,最终会导致系统占用内存过高,引发OutOfMemoryError错误,关键代码如下:

  1. public void execute(Runnable task, long startTimeout) {
  2. Assert.notNull(task, "Runnable must not be null");
  3. Runnable taskToUse = this.taskDecorator != null ? this.taskDecorator.decorate(task) : task;
  4. //判断是否开启限流,默认为否
  5. if (this.isThrottleActive() && startTimeout > 0L) {
  6. //执行前置操作,进行限流
  7. this.concurrencyThrottle.beforeAccess();
  8. this.doExecute(new SimpleAsyncTaskExecutor.ConcurrencyThrottlingRunnable(taskToUse));
  9. } else {
  10. //未限流的情况,执行线程任务
  11. this.doExecute(taskToUse);
  12. }
  13. }
  14. protected void doExecute(Runnable task) {
  15. //不断创建线程
  16. Thread thread = this.threadFactory != null ? this.threadFactory.newThread(task) : this.createThread(task);
  17. thread.start();
  18. }
  19. //创建线程
  20. public Thread createThread(Runnable runnable) {
  21. //指定线程名,task-1,task-2...
  22. Thread thread = new Thread(this.getThreadGroup(), runnable, this.nextThreadName());
  23. thread.setPriority(this.getThreadPriority());
  24. thread.setDaemon(this.isDaemon());
  25. return thread;
  26. }

也可以直接通过上面的控制台日志观察,每次打印的线程名都是[task-1]、[task-2]、[task-3]、[task-4]…..递增的。
正因如此,所以在使用Spring中的@Async异步框架时一定要自定义线程池,替代默认的SimpleAsyncTaskExecutor
Spring提供了多种线程池:

  • SimpleAsyncTaskExecutor:不是真的线程池,这个类不重用线程,每次调用都会创建一个新的线程。
  • SyncTaskExecutor:这个类没有实现异步调用,只是一个同步操作。只适用于不需要多线程的地
  • ConcurrentTaskExecutor:Executor的适配类,不推荐使用。如果ThreadPoolTaskExecutor不满足要求时,才用考虑使用这个类
  • ThreadPoolTaskScheduler:可以使用cron表达式
  • ThreadPoolTaskExecutor :最常使用,推荐。其实质是对java.util.concurrent.ThreadPoolExecutor的包装

    @Async实现一个自定义线程池

    1. @Configuration
    2. @EnableAsync
    3. public class SyncConfiguration {
    4. @Bean(name = "asyncPoolTaskExecutor")
    5. public ThreadPoolTaskExecutor executor() {
    6. ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    7. //核心线程数
    8. taskExecutor.setCorePoolSize(10);
    9. //线程池维护线程的最大数量,只有在缓冲队列满了之后才会申请超过核心线程数的线程
    10. taskExecutor.setMaxPoolSize(100);
    11. //缓存队列
    12. taskExecutor.setQueueCapacity(50);
    13. //许的空闲时间,当超过了核心线程出之外的线程在空闲时间到达之后会被销毁
    14. taskExecutor.setKeepAliveSeconds(200);
    15. //异步方法内部线程名称
    16. taskExecutor.setThreadNamePrefix("async-");
    17. /**
    18. * 当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize,如果还有任务到来就会采取任务拒绝策略
    19. * 通常有以下四种策略:
    20. * ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
    21. * ThreadPoolExecutor.DiscardPolicy:也是丢弃任务,但是不抛出异常。
    22. * ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,然后重新尝试执行任务(重复此过程)
    23. * ThreadPoolExecutor.CallerRunsPolicy:重试添加当前的任务,自动重复调用 execute() 方法,直到成功
    24. */
    25. taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    26. taskExecutor.initialize();
    27. return taskExecutor;
    28. }
    29. }

    配置自定义线程池以后就可以大胆的使用@Async提供的异步处理能力了。

    多个线程池处理

    在现实的互联网项目开发中,针对高并发的请求,一般的做法是高并发接口单独线程池隔离处理。
    假设现在2个高并发接口:一个是修改用户信息接口,刷新用户redis缓存;一个是下订单接口,发送app push信息。往往会根据接口特征定义两个线程池,这时候在使用@Async时就需要通过指定线程池名称进行区分。

    @Async指定线程池名字

    1. @SneakyThrows
    2. @Async("asyncPoolTaskExecutor")
    3. public void doTask1() {
    4. long t1 = System.currentTimeMillis();
    5. Thread.sleep(2000);
    6. long t2 = System.currentTimeMillis();
    7. log.info("task1 cost {} ms" , t2-t1);
    8. }

    当系统存在多个线程池时,也可以配置一个默认线程池,对于非默认的异步任务再通过@Async("otherTaskExecutor")来指定线程池名称。

    配置默认线程池

    可以修改配置类让其实现AsyncConfigurer,并重写getAsyncExecutor()方法,指定默认线程池:

    1. @Configuration
    2. @EnableAsync
    3. @Slf4j
    4. public class AsyncConfiguration implements AsyncConfigurer {
    5. @Bean(name = "asyncPoolTaskExecutor")
    6. public ThreadPoolTaskExecutor executor() {
    7. ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    8. //核心线程数
    9. taskExecutor.setCorePoolSize(2);
    10. //线程池维护线程的最大数量,只有在缓冲队列满了之后才会申请超过核心线程数的线程
    11. taskExecutor.setMaxPoolSize(10);
    12. //缓存队列
    13. taskExecutor.setQueueCapacity(50);
    14. //许的空闲时间,当超过了核心线程出之外的线程在空闲时间到达之后会被销毁
    15. taskExecutor.setKeepAliveSeconds(200);
    16. //异步方法内部线程名称
    17. taskExecutor.setThreadNamePrefix("async-");
    18. /**
    19. * 当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize,如果还有任务到来就会采取任务拒绝策略
    20. * 通常有以下四种策略:
    21. * ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
    22. * ThreadPoolExecutor.DiscardPolicy:也是丢弃任务,但是不抛出异常。
    23. * ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,然后重新尝试执行任务(重复此过程)
    24. * ThreadPoolExecutor.CallerRunsPolicy:重试添加当前的任务,自动重复调用 execute() 方法,直到成功
    25. */
    26. taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    27. taskExecutor.initialize();
    28. return taskExecutor;
    29. }
    30. /**
    31. * 指定默认线程池
    32. */
    33. @Override
    34. public Executor getAsyncExecutor() {
    35. return executor();
    36. }
    37. @Override
    38. public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
    39. return (ex, method, params) ->
    40. log.error("线程池执行任务发送未知错误,执行方法:{}",method.getName(),ex);
    41. }
    42. }

    如下,doTask1()方法使用默认使用线程池asyncPoolTaskExecutordoTask2()使用线程池otherTaskExecutor,非常灵活。 ```java @Async public void doTask1() { long t1 = System.currentTimeMillis(); Thread.sleep(2000); long t2 = System.currentTimeMillis(); log.info(“task1 cost {} ms” , t2-t1); }

@SneakyThrows @Async(“otherTaskExecutor”) public void doTask2() { long t1 = System.currentTimeMillis(); Thread.sleep(3000); long t2 = System.currentTimeMillis(); log.info(“task2 cost {} ms” , t2-t1); } ```