SpringBoot 实现异步调用 @Async

ゝ一纸荒年。 2023-10-02 22:36 11阅读 0赞

#### SpringBoot 实现异步调用 @Async ####

--------------------

### 为什么要用异步框架，它解决什么问题？ ###

在SpringBoot的日常开发中，一般都是同步调用的。但实际中有很多场景非常适合使用异步来处理，如：注册新用户，送100个积分；或下单成功，发送push消息等等。

**就拿注册新用户这个用例来说，为什么要异步处理？**

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

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

--------------------

### SpringBoot如何实现异步调用？ ###

对于异步方法调用，从Spring3开始提供了`@Async`注解，我们只需要在方法上标注此注解，此方法即可实现异步调用。

当然，我们还需要一个配置类，通过Enable模块驱动注解`@EnableAsync`来开启异步功能。

--------------------

### 实现异步调用 ###

#### 第一步：新建配置类，开启@[Async][]功能支持 ####

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

**至于为什么使用线程池，后面会讲到。**

import lombok.extern.slf4j.Slf4j;
    import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.scheduling.annotation.AsyncConfigurer;
    import org.springframework.scheduling.annotation.EnableAsync;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    import java.util.concurrent.Executor;
    import java.util.concurrent.ThreadPoolExecutor;
    
    /**
     * 描述：异步配置
     */
    @Slf4j
    @Configuration
    @EnableAsync    // 可放在启动类上或单独的配置类
    public class AsyncConfiguration implements AsyncConfigurer {
    
        @Bean(name = "asyncPoolTaskExecutor")
        public ThreadPoolTaskExecutor executor() {
            ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
            //核心线程数
            taskExecutor.setCorePoolSize(10);
            //线程池维护线程的最大数量,只有在缓冲队列满了之后才会申请超过核心线程数的线程
            taskExecutor.setMaxPoolSize(100);
            //缓存队列
            taskExecutor.setQueueCapacity(50);
            //设置线程的空闲时间,当超过了核心线程出之外的线程在空闲时间到达之后会被销毁
            taskExecutor.setKeepAliveSeconds(200);
            //异步方法内部线程名称
            taskExecutor.setThreadNamePrefix("async-");
            /**
             * 当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize，如果还有任务到来就会采取任务拒绝策略
             * 通常有以下四种策略：
             * ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
             * ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
             * ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
             * ThreadPoolExecutor.CallerRunsPolicy：重试添加当前的任务，自动重复调用 execute() 方法，直到成功
             */
            taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            taskExecutor.initialize();
            return taskExecutor;
        }
    
        /**
         * 指定默认线程池
         * The {@link Executor} instance to be used when processing async method invocations.
         */
        @Override
        public Executor getAsyncExecutor() {
            return executor();
        }
    
        /**
         * The {@link AsyncUncaughtExceptionHandler} instance to be used
         * when an exception is thrown during an asynchronous method execution
         * with {@code void} return type.
         */
        @Override
        public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
            return (ex, method, params) -> log.error("线程池执行任务发送未知错误, 执行方法：{}", method.getName(), ex);
        }
    }

--------------------

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

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

import lombok.SneakyThrows;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.scheduling.annotation.Async;
    import org.springframework.stereotype.Component;
    
    /**
     * 描述：异步方法调用
     */
    @Slf4j
    @Component
    public class AsyncTask {
    
        @SneakyThrows
        @Async  // 在异步配置类设置了默认线程池则不需要再指定线程池名称
    //    @Async("asyncPoolTaskExecutor")
        public void doTask1() {
            long t1 = System.currentTimeMillis();
            Thread.sleep(2000);
            long t2 = System.currentTimeMillis();
            log.info("task1方法耗时 {} ms" , t2-t1);
        }
    
        @SneakyThrows
        @Async
    //    @Async("otherPoolTaskExecutor") // 其他线程池的名称
        public void doTask2() {
            long t1 = System.currentTimeMillis();
            Thread.sleep(3000);
            long t2 = System.currentTimeMillis();
            log.info("task2方法耗时 {} ms" , t2-t1);
        }
    }

--------------------

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

import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    @Slf4j
    @RestController
    @RequestMapping("/async")
    public class AsyncController {
        @Autowired
        private AsyncTask asyncTask;
    
        @RequestMapping("/task")
        public void task() throws InterruptedException {
            long t1 = System.currentTimeMillis();
            asyncTask.doTask1();
            asyncTask.doTask2();
            Thread.sleep(1000);
            long t2 = System.currentTimeMillis();
            log.info("main方法耗时{} ms", t2-t1);
        }
    }

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

[2021-12-01 20:21:36.036] [INFO] [http-nio-8080-exec-1] - task(AsyncController.java:27) - main方法耗时 1009 ms
    [2021-12-01 20:21:37.037] [INFO] [async-1] - doTask1(AsyncTask.java:23) - task1方法耗时 2004 ms
    [2021-12-01 20:21:38.038] [INFO] [async-2] - doTask2(AsyncTask.java:32) - task2方法耗时 3003 ms

通过日志可以看到：主线程不需要等待异步方法执行完成，减少了响应时间，提高了接口性能。

**通过上面三步我们就可以在SpringBoot中使用异步方法来提高我们接口性能了。**

--------------------

### 为什么要给`@Async`自定义线程池？ ###

使用`@Async`注解，在默认情况下用的是`SimpleAsyncTaskExecutor`线程池，该线程池不是真正意义上的线程池。

使用此线程池无法实现线程重用，每次调用都会新建一条线程。若系统中不断的创建线程，最终会导致系统占用内存过高，引发`OutOfMemoryErro`r错误，关键代码如下：

public void execute(Runnable task, long startTimeout) {
      Assert.notNull(task, "Runnable must not be null");
      Runnable taskToUse = this.taskDecorator != null ? this.taskDecorator.decorate(task) : task;
      //判断是否开启限流，默认为否
      if (this.isThrottleActive() && startTimeout > 0L) {
        //执行前置操作，进行限流
        this.concurrencyThrottle.beforeAccess();
        this.doExecute(new SimpleAsyncTaskExecutor.ConcurrencyThrottlingRunnable(taskToUse));
      } else {
        //未限流的情况，执行线程任务
        this.doExecute(taskToUse);
      }
    
    }
    
    protected void doExecute(Runnable task) {
      //不断创建线程
      Thread thread = this.threadFactory != null ? this.threadFactory.newThread(task) : this.createThread(task);
      thread.start();
    }
    
    //创建线程
    public Thread createThread(Runnable runnable) {
      //指定线程名，task-1，task-2...
      Thread thread = new Thread(this.getThreadGroup(), runnable, this.nextThreadName());
      thread.setPriority(this.getThreadPriority());
      thread.setDaemon(this.isDaemon());
      return thread;
    }

我们也可以直接通过上面的控制台日志观察，每次打印的线程名都是\[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`实现一个自定义线程池 ####

import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.scheduling.annotation.EnableAsync;
    import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
    
    import java.util.concurrent.ThreadPoolExecutor;
    
    /**
     * 描述：异步配置2
     */
    @Configuration
    @EnableAsync    // 可放在启动类上或单独的配置类
    public class AsyncConfiguration2 {
        @Bean(name = "asyncPoolTaskExecutor")
        public ThreadPoolTaskExecutor executor() {
            ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
            //核心线程数
            taskExecutor.setCorePoolSize(10);
            //线程池维护线程的最大数量,只有在缓冲队列满了之后才会申请超过核心线程数的线程
            taskExecutor.setMaxPoolSize(100);
            //缓存队列
            taskExecutor.setQueueCapacity(50);
            //许的空闲时间,当超过了核心线程出之外的线程在空闲时间到达之后会被销毁
            taskExecutor.setKeepAliveSeconds(200);
            //异步方法内部线程名称
            taskExecutor.setThreadNamePrefix("async-");
            /**
             * 当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize，如果还有任务到来就会采取任务拒绝策略
             * 通常有以下四种策略：
             * ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
             * ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
             * ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
             * ThreadPoolExecutor.CallerRunsPolicy：重试添加当前的任务，自动重复调用 execute() 方法，直到成功
             */
            taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            taskExecutor.initialize();
            return taskExecutor;
        }
    }

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

#### 多个线程池处理 ####

在现实的互联网项目开发中，针对高并发的请求，一般的做法是高并发接口单独线程池隔离处理。

假设现在2个高并发接口：一个是修改用户信息接口，刷新用户redis缓存；一个是下订单接口，发送app push信息。往往会根据接口特征定义两个线程池，这时候我们在使用`@Async`时就需要通过指定线程池名称进行区分。

#### 为`@Async`指定线程池名字 ####

@SneakyThrows
    @Async("asyncPoolTaskExecutor")
    public void doTask1() {
      long t1 = System.currentTimeMillis();
      Thread.sleep(2000);
      long t2 = System.currentTimeMillis();
      log.info("task1方法耗时 {} ms" , t2-t1);
    }

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

#### 配置默认线程池 ####

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

如下，`doTask1()`方法使用默认使用线程池`asyncPoolTaskExecutor`，`doTask2()`使用线程池`otherTaskExecutor`，非常灵活。

@SneakyThrows
    @Async
    public void doTask1() {
      long t1 = System.currentTimeMillis();
      Thread.sleep(2000);
      long t2 = System.currentTimeMillis();
      log.info("task1方法耗时 {} ms" , t2-t1);
    }
    
    @SneakyThrows
    @Async("otherTaskExecutor")
    public void doTask2() {
      long t1 = System.currentTimeMillis();
      Thread.sleep(3000);
      long t2 = System.currentTimeMillis();
      log.info("task2方法耗时 {} ms" , t2-t1);
    }

--------------------

### 使用`@Async`注解可能会导致的问题 ###

如果`serviceA`、`serviceB`对象之间相互依赖，`serviceA`和`serviceB`总一个一个会先实例化，而`serviceA`或`serviceB`里面使用了`@Async`注解，会导致循环依赖异常：`org.springframework.beans.factory.BeanCurrentlyInCreationException`

在springboot中，以上报错被捕捉，抛出的异常是：  
`The dependencies of some of the beans in the application context form a cycle`

#### 原因 ####

我们知道，spring三级缓存一定程度上解决了循环依赖问题。A对象在实例化之后，属性赋值【opulateBean(beanName, mbd, instanceWrapper)】执行之前，将ObjectFactory添加至三级缓存中，从而使得在B对象实例化后的属性赋值过程中，能从三级缓存拿到ObjectFactory，调用getObject()方法拿到A的引用，B由此能顺利完成初始化并加入到IOC容器。此时A对象完成属性赋值之后，将会执行初始化【initializeBean(beanName, exposedObject, mbd)方法】，重点是@Async注解的处理正是在这地方完成的，其对应的后置处理器AsyncAnnotationBeanPostProcessor，在postProcessAfterInitialization方法中将返回代理对象，此代理对象与B中持有的A对象引用不同，导致了以上报错。

#### 解决办法 ####

*  1.在A类上加`@Lazy`，保证A对象实例化晚于B对象
 *  2.不使用`@Async`注解，通过自定义异步工具类发起异步线程（线程池）
 *  3.不要让`@Async`的Bean参与循环依赖

#### `@Async`问题解决参考博客 ####

*  [https://segmentfault.com/a/1190000021217176][https_segmentfault.com_a_1190000021217176]
 *  [https://cloud.tencent.com/developer/article/1497689][https_cloud.tencent.com_developer_article_1497689]

END

[Async]: https://so.csdn.net/so/search?q=Async&spm=1001.2101.3001.7020
[https_segmentfault.com_a_1190000021217176]: https://segmentfault.com/a/1190000021217176
[https_cloud.tencent.com_developer_article_1497689]: https://cloud.tencent.com/developer/article/1497689