SpringBoot异步调用-蒲公英云

除了异步请求，一般上我们用的比较多的应该是异步调用。通常在开发过程中，会遇到一个方法是和实际业务无关的，没有紧密性的。比如记录日志信息等业务。这个时候正常就是启一个新线程去做一些业务处理，让主线程异步的执行其他业务。

何为异步调用

说异步调用前，我们说说它对应的同步调用。通常开发过程中，一般上我们都是同步调用，即：程序按定义的顺序依次执行的过程，每一行代码执行过程必须等待上一行代码执行完毕后才执行。而异步调用指：程序在执行时，无需等待执行的返回值可继续执行后面的代码。显而易见，同步有依赖相关性，而异步没有，所以异步可并发执行，可提高执行效率，在相同的时间做更多的事情。

题外话：除了异步、同步外，还有一个叫回调。其主要是解决异步方法执行结果的处理方法，比如在希望异步调用结束时返回执行结果，这个时候就可以考虑使用回调机制。

Async异步调用

在SpringBoot中使用异步调用是很简单的，只需要使用@Async注解即可实现方法的异步调用。

注意：需要在启动类加入@EnableAsync使异步调用@Async注解生效。使用@Async很简单，只需要在需要异步执行的方法上加入此注解即可。这里创建一个控制层和一个服务层，进行简单示例下。

@Slf4j
@Service
public class SyncService {
    @Async
    public void asyncEvent() throws InterruptedException {
        //休眠1s
        TimeUnit.SECONDS.sleep(1);
        log.info("异步方法内部线程名称：{}!", Thread.currentThread().getName());
    }
    public void syncEvent() throws InterruptedException {
        TimeUnit.SECONDS.sleep(1);
        log.info("同步方法内部线程名称：{}!", Thread.currentThread().getName());
    }
}
@Slf4j
@RestController
public class AsyncController {
    @Autowired
    private SyncService syncService;
    @GetMapping("/async")
    public String doAsync() throws InterruptedException, ExecutionException {
        long start = System.currentTimeMillis();
        //调用同步方法
        syncService.syncEvent();
        long syncEndTime = System.currentTimeMillis();
        log.info("同步方法用时：{}", syncEndTime - start);
        //调用异步方法
        syncService.asyncEvent();//异步方法无返回值
        long asyncEndTime = System.currentTimeMillis();
        log.info("异步方法用时：{}", asyncEndTime - syncEndTime);
        return "async!!!";
    }
}
@SpringBootApplication
@EnableAsync
public class SpringWebApplication{
    public static void main(String[] args) {
        SpringApplication.run(SpringWebApplication.class, args)；
    }
}

应用启动后，可以看见控制台输出:

可以看出，调用异步方法时，是立即返回的，基本没有耗时。

这里有几点需要注意下：

在默认情况下，未设置TaskExecutor时，默认是使用SimpleAsyncTaskExecutor这个线程池，但此线程不是真正意义上的线程池，因为线程不重用，每次调用都会创建一个新的线程。可通过控制台日志输出可以看出，每次输出线程名都是递增的。
调用的异步方法，不能为同一个类的方法，简单来说，因为Spring在启动扫描时会为其创建一个代理类，而同类调用时，还是调用本身的代理类的，所以和平常调用是一样的。其他的注解如@Cache等也是一样的道理，说白了，就是Spring的代理机制造成的。

自定义线程池

前面有提到，在默认情况下，系统使用的是默认的SimpleAsyncTaskExecutor进行线程创建。所以一般上我们会自定义线程池来进行线程的复用。

@Configuration
public class Config {
    /**
     * 配置线程池
     * @return
     */
    @Bean(name = "asyncPoolTaskExecutor")
    public ThreadPoolTaskExecutor asyncThreadPoolTaskExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(20);
        taskExecutor.setMaxPoolSize(200);
        taskExecutor.setQueueCapacity(25);
        taskExecutor.setKeepAliveSeconds(200);
        taskExecutor.setThreadNamePrefix("MyExecutor-");
        // 线程池对拒绝任务（无线程可用）的处理策略，目前只支持AbortPolicy、CallerRunsPolicy；默认为后者
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        taskExecutor.initialize();
        return taskExecutor;
    }
}

此时，使用的是就只需要在@Async加入线程池名称即可：

@Async("asyncPoolTaskExecutor")
public void asyncEvent() throws InterruptedException {
    //休眠1s
    TimeUnit.SECONDS.sleep(1);
    log.info("异步方法内部线程名称：{}!", Thread.currentThread().getName());
}

这里简单说明下，关于ThreadPoolTaskExecutor参数说明：

corePoolSize：线程池维护线程的最少数量
keepAliveSeconds：允许的空闲时间,当超过了核心线程数之外的线程在空闲时间到达之后会被销毁
maxPoolSize：线程池维护线程的最大数量,只有在缓冲队列满了之后才会申请超过核心线程数的线程
queueCapacity：缓存队列
rejectedExecutionHandler：线程池对拒绝任务（无线程可用）的处理策略。这里采用了CallerRunsPolicy策略，当线程池没有处理能力的时候，该策略会直接在 execute 方法的调用线程中运行被拒绝的任务；如果执行程序已关闭，则会丢弃该任务。还有一个是AbortPolicy策略：处理程序遭到拒绝将抛出运行时RejectedExecutionException。

而在一些场景下，若需要在关闭线程池时等待当前调度任务完成后才开始关闭，可以通过简单的配置，进行优雅的停机策略配置。关键就是通过setWaitForTasksToCompleteOnShutdown(true)和setAwaitTerminationSeconds方法。

setWaitForTasksToCompleteOnShutdown:表明等待所有线程执行完，默认为false。
setAwaitTerminationSeconds:等待的时间，因为不能无限的等待下去。

所以，线程池完整配置为：

@Bean(name = "asyncPoolTaskExecutor")
    public ThreadPoolTaskExecutor asyncThreadPoolTaskExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(20);
        taskExecutor.setMaxPoolSize(200);
        taskExecutor.setQueueCapacity(25);
        taskExecutor.setKeepAliveSeconds(200);
        taskExecutor.setThreadNamePrefix("MyExecutor-");
        // 线程池对拒绝任务（无线程可用）的处理策略，目前只支持AbortPolicy、CallerRunsPolicy；默认为后者
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //调度器shutdown被调用时等待当前被调度的任务完成
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        //等待时长
        taskExecutor.setAwaitTerminationSeconds(60);
        taskExecutor.initialize();
        return taskExecutor;
    }

配置默认的线程池

如果我们想使用默认的线程池，但是只是想修改默认线程池的配置，那怎么做呢，此时我们需要实现AsyncConfigurer类，示例代码如下：

@Configuration
@EnableAsync
public class TaskExecutorConfig implements AsyncConfigurer {
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return null;
    }
    /**
     * 配置线程池
     * @return
     */
    public ThreadPoolTaskExecutor getAsyncExecutor() {
       ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(20);
        taskExecutor.setMaxPoolSize(200);
        taskExecutor.setQueueCapacity(25);
        taskExecutor.setKeepAliveSeconds(200);
        taskExecutor.setThreadNamePrefix("MyExecutor-");//线程名称前缀
        // 线程池对拒绝任务（无线程可用）的处理策略，目前只支持AbortPolicy、CallerRunsPolicy；默认为后者
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //调度器shutdown被调用时等待当前被调度的任务完成
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        //等待时长
        taskExecutor.setAwaitTerminationSeconds(60);
        taskExecutor.initialize();
        return taskExecutor;
    }
}

异步回调及超时处理

对于一些业务场景下，需要异步回调的返回值时，就需要使用异步回调来完成了。主要就是通过Future进行异步回调。

@Async
public Future<String> asyncEventWithReturn() throws InterruptedException {
    //休眠1s
    TimeUnit.SECONDS.sleep(1);
    log.info("异步方法内部线程名称：{}!", Thread.currentThread().getName());
    return new AsyncResult<>("异步方法有返回值");
}

其中AsyncResult是Spring提供的一个Future接口的子类。

然后通过isDone方法，判断是否已经执行完毕。

@Slf4j
@RestController
public class AsyncController {
    @Autowired
    private SyncService syncService;
@GetMapping("/async_return")
public String doAsyncWithReturn() throws InterruptedException, ExecutionException {
    long start = System.currentTimeMillis();
    //调用同步方法
    syncService.syncEvent();
    long syncEndTime = System.currentTimeMillis();
    log.info("同步方法用时：{}", syncEndTime - start);
    //调用异步方法
    // 异步方法有返回值
    Future<String> doFutrue = syncService.asyncEventWithReturn();
    while (true) {
        //判断异步任务是否完成
        if (doFutrue.isDone()) {
            log.info("异步回调结果：" + doFutrue.get());
            break;
        }
        Thread.sleep(100);
    }
    long endTime = System.currentTimeMillis();
    log.info("异步方法用时：{}", endTime - syncEndTime);
    return doFutrue.get();
}
}