SpringBoot使用@Async实现异步调用

时间:2023-03-09 05:35:53
SpringBoot使用@Async实现异步调用

1、@EnableAsync

首先,我们需要在启动类上添加  @EnableAsync 注解来声明开启异步方法。

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

2、@Async

需要注意的,@Async在使用上有一些限制:

  • 它只能应用于public修饰的方法
  • 自调用–从同一个类中调用async方法,将不起作用

原因很简单:

  • 只有公共方法,才可以被代理。
  • 自调用不起作用,因为它越过了代理直接调用了方法。

2.1、无返回值的异步方法

这是一个异步运行的无返回值方法:

 @Async
public void asyncMethodWithVoidReturnType() {
System.out.println("异步无返回值方法 "
+ Thread.currentThread().getName());
}

实例:

  • AsyncTask:异步式任务类,定义了三个异步式方法。
@Component
public class AsyncTask {
Logger log= LoggerFactory.getLogger(AsyncTask.class); private Random random = new Random(); /**
* 定义三个异步式方法
* @throws InterruptedException
*/
@Async
public void taskOne() throws InterruptedException {
long start = System.currentTimeMillis();
//随机休眠若干毫秒
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
log.info("任务一执行完成耗时{}秒", (end - start)/1000f);
} @Async
public void taskTwo() throws InterruptedException {
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
log.info("任务二执行完成耗时{}秒", (end - start)/1000f);
} @Async
public void taskThree() throws InterruptedException {
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
log.info("任务三执行完成耗时{}秒", (end - start)/1000f);
} }
  • 在测试类中调用三个异步式方法:
@SpringBootTest
@RunWith(SpringRunner.class)
public class AsyncTaskTest { @Autowired
private AsyncTask asyncTask; Logger log= LoggerFactory.getLogger(AsyncTaskTest.class); @Test
public void doAsyncTasks(){
try {
long start = System.currentTimeMillis();
//调用三个异步式方法
asyncTask.taskOne();
asyncTask.taskTwo();
asyncTask.taskThree();
Thread.sleep(5000);
long end = System.currentTimeMillis();
log.info("主程序执行完成耗时{}秒", (end - start)/1000f);
} catch (InterruptedException e) {
e.printStackTrace();
}
} }

运行结果:可以看到三个方法没有顺序执行,这个复执行单元测试,您可能会遇到各种不同的结果,比如:

  • 没有任何任务相关的输出
    • 乱序的任务相关的输出
    • 有部分任务相关的输出

原因是目前doTaskOne、doTaskTwo、doTaskThree三个函数的时候已经是异步执行了。主程序在异步调用之后,主程序并不会理会这三个函数是否执行完成了,由于没有其他需要执行的内容,所以程序就自动结束了,导致了不完整或是没有输出任务相关内容的情况。

2.1、有返回值的异步方法

@Async也可以应用有返回值的方法–通过在Future中包装实际的返回值:

 /**
* 有返回值的异步方法
* @return
*/
@Async
public Future<String> asyncMethodWithReturnType() {
System.out.println("执行有返回值的异步方法 "
+ Thread.currentThread().getName());
try {
Thread.sleep(5000);
return new AsyncResult<String>("hello world !!!!");
} catch (InterruptedException e) {
//
}
return null;
}

Spring还提供了一个实现Future的AsyncResult类。这个类可用于跟踪异步方法执行的结果。

实例:

  • 我们将2.1的实例改造成有返回值的异步方法:
    @Async
public Future<String> taskOne() throws InterruptedException {
long start = System.currentTimeMillis();
//随机休眠若干毫秒
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
log.info("任务一执行完成耗时{}秒", (end - start)/1000f);
return new AsyncResult<>("任务一完事了");
}

taskTwo、taskThree方法做同样的改造。

  • 测试有返回值的异步方法:
 @Test
public void doFutureTask(){
try {
long start=System.currentTimeMillis();
Future<String> future1=asyncTask.taskOne();
Future <String> future2 = asyncTask.taskTwo();
Future <String> future3 = asyncTask.taskThree();
//三个任务执行完再执行主程序
do {
Thread.sleep(100);
} while (future1.isDone() && future2.isDone() && future3.isDone());
log.info("获取异步方法的返回值:{}", future1.get());
Thread.sleep(5000);
long end = System.currentTimeMillis();
log.info("主程序执行完成耗时{}秒", (end - start)/1000f);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}

运行结果:可以看到三个任务完成后才执行主程序,还输出了异步方法的返回值。

3、 Executor

默认情况下,Spring使用SimpleAsyncTaskExecutor异步运行这些方法。

可以在两个级别上重写默认线程池——应用程序级别或方法级别。

3.1、方法级别重写Executor

所需的执行程序需要在配置类中声明 Executor:

@Configuration
@EnableAsync
public class SpringAsyncConfig { @Bean(name = "threadPoolTaskExecutor")
public Executor threadPoolTaskExecutor() {
return new ThreadPoolTaskExecutor();
}
}

然后,在@Async中的属性提供Executor名称:

 @Async("threadPoolTaskExecutor")
public void asyncMethodWithConfiguredExecutor() {
System.out.println("Execute method with configured executor - "
+ Thread.currentThread().getName());
}

3.2、应用级别重写Executor

配置类应实现AsyncConfigurer接口,重写getAsyncExecutor()方法。

在这里,我们将返回整个应用程序的Executor,这样一来,它就成为运行以@Async注释的方法的默认Executor:

@Configuration
@EnableAsync
public class SpringApplicationAsyncConfig implements AsyncConfigurer {
@Override
public Executor getAsyncExecutor() {
return new ThreadPoolTaskExecutor();
}
}

3.3、自定义线程池配置

在上面,自定义线程池只是简单地返回了一个线程池:

return new ThreadPoolTaskExecutor();

实际上,还可以对线程池做一些配置:

@Configuration
@EnableAsync
public class SpringPropertiesAsyncConfig implements AsyncConfigurer { /**
* 对线程池进行配置
* @return
*/
@Override
public Executor getAsyncExecutor() {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(20);
taskExecutor.setMaxPoolSize(200);
taskExecutor.setQueueCapacity(25);
taskExecutor.setKeepAliveSeconds(200);
taskExecutor.setThreadNamePrefix("oKong-");
// 线程池对拒绝任务(无线程可用)的处理策略,目前只支持AbortPolicy、CallerRunsPolicy;默认为后者
taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
taskExecutor.initialize();
return taskExecutor;
}
}

ThreadPoolTaskExecutor配置参数的简单说明:

  • corePoolSize:线程池维护线程的最少数量

  • keepAliveSeconds:允许的空闲时间,当超过了核心线程出之外的线程在空闲时间到达之后会被销毁

  • maxPoolSize:线程池维护线程的最大数量,只有在缓冲队列满了之后才会申请超过核心线程数的线程

  • queueCapacity:缓存队列

  • rejectedExecutionHandler:线程池对拒绝任务(无线程可用)的处理策略。这里采用了CallerRunsPolicy策略,当线程池没有处理能力的时候,该策略会直接在 execute 方法的调用线程中运行被拒绝的任务;如果执行程序已关闭,则会丢弃该任务。还有一个是AbortPolicy策略:处理程序遭到拒绝将抛出运行时RejectedExecutionException。

4、异常处理

当方法返回类型为Future时,异常处理很容易– Future.get()方法将抛出异常。

但是如果是无返回值的异步方法,异常不会传播到调用线程。因此,我们需要添加额外的配置来处理异常。

我们将通过实现AsyncUncaughtExceptionHandler接口来创建自定义异步异常处理程序。

当存在任何未捕获的异步异常时,将调用handleUncaughtException()方法:

public class CustomAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {
@Override
public void handleUncaughtException(Throwable throwable, Method method, Object... objects) {
System.out.println("Exception message - " + throwable.getMessage());
System.out.println("Method name - " + method.getName());
for (Object param : objects) {
System.out.println("Parameter value - " + param);
}
}
}

上面,我们使用配置类实现了AsyncConfigurer接口。

作为其中的一部分,我们还需要重写getAsyncUncaughtExceptionHandler()方法以返回我们的自定义异步异常处理:

 /**
* 返回自定义异常处理
* @return
*/
@Override
public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
return new CustomAsyncExceptionHandler();
}

5、总结

这里异步请求的使用及相关配置,如超时,异常等处理。在剥离一些和业务无关的操作时,就可以考虑使用异步调用进行其他无关业务操作,以此提供业务的处理效率。或者一些业务场景下可拆分出多个方法进行同步执行又互不影响时,也可以考虑使用异步调用方式提供执行效率。

1、@EnableAsync

首先,我们需要在启动类上添加  @EnableAsync 注解来声明开启异步方法。

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

2、@Async

需要注意的,@Async在使用上有一些限制:

  • 它只能应用于public修饰的方法
  • 自调用–从同一个类中调用async方法,将不起作用

原因很简单:

  • 只有公共方法,才可以被代理。
  • 自调用不起作用,因为它越过了代理直接调用了方法。

2.1、无返回值的异步方法

这是一个异步运行的无返回值方法:

  @Async
public void asyncMethodWithVoidReturnType() {
System.out.println("异步无返回值方法 "
+ Thread.currentThread().getName());
}

实例:

  • AsyncTask:异步式任务类,定义了三个异步式方法。
@Component
public class AsyncTask {
Logger log= LoggerFactory.getLogger(AsyncTask.class); private Random random = new Random(); /**
* 定义三个异步式方法
* @throws InterruptedException
*/
@Async
public void taskOne() throws InterruptedException {
long start = System.currentTimeMillis();
//随机休眠若干毫秒
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
log.info("任务一执行完成耗时{}秒", (end - start)/1000f);
} @Async
public void taskTwo() throws InterruptedException {
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
log.info("任务二执行完成耗时{}秒", (end - start)/1000f);
} @Async
public void taskThree() throws InterruptedException {
long start = System.currentTimeMillis();
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
log.info("任务三执行完成耗时{}秒", (end - start)/1000f);
} }
  • 在测试类中调用三个异步式方法:
@SpringBootTest
@RunWith(SpringRunner.class)
public class AsyncTaskTest { @Autowired
private AsyncTask asyncTask; Logger log= LoggerFactory.getLogger(AsyncTaskTest.class); @Test
public void doAsyncTasks(){
try {
long start = System.currentTimeMillis();
//调用三个异步式方法
asyncTask.taskOne();
asyncTask.taskTwo();
asyncTask.taskThree();
Thread.sleep(5000);
long end = System.currentTimeMillis();
log.info("主程序执行完成耗时{}秒", (end - start)/1000f);
} catch (InterruptedException e) {
e.printStackTrace();
}
} }

运行结果:可以看到三个方法没有顺序执行,这个复执行单元测试,您可能会遇到各种不同的结果,比如:

  • 没有任何任务相关的输出
    • 乱序的任务相关的输出
    • 有部分任务相关的输出

原因是目前doTaskOne、doTaskTwo、doTaskThree三个函数的时候已经是异步执行了。主程序在异步调用之后,主程序并不会理会这三个函数是否执行完成了,由于没有其他需要执行的内容,所以程序就自动结束了,导致了不完整或是没有输出任务相关内容的情况。

2.1、有返回值的异步方法

@Async也可以应用有返回值的方法–通过在Future中包装实际的返回值:

   /**
* 有返回值的异步方法
* @return
*/
@Async
public Future<String> asyncMethodWithReturnType() {
System.out.println("执行有返回值的异步方法 "
+ Thread.currentThread().getName());
try {
Thread.sleep(5000);
return new AsyncResult<String>("hello world !!!!");
} catch (InterruptedException e) {
//
}
return null;
}

Spring还提供了一个实现Future的AsyncResult类。这个类可用于跟踪异步方法执行的结果。

实例:

  • 我们将2.1的实例改造成有返回值的异步方法:
 @Async
public Future<String> taskOne() throws InterruptedException {
long start = System.currentTimeMillis();
//随机休眠若干毫秒
Thread.sleep(random.nextInt(10000));
long end = System.currentTimeMillis();
log.info("任务一执行完成耗时{}秒", (end - start)/1000f);
return new AsyncResult<>("任务一完事了");
}

taskTwo、taskThree方法做同样的改造。

  • 测试有返回值的异步方法:
  @Test
public void doFutureTask(){
try {
long start=System.currentTimeMillis();
Future<String> future1=asyncTask.taskOne();
Future <String> future2 = asyncTask.taskTwo();
Future <String> future3 = asyncTask.taskThree();
//三个任务执行完再执行主程序
do {
Thread.sleep(100);
} while (future1.isDone() && future2.isDone() && future3.isDone());
log.info("获取异步方法的返回值:{}", future1.get());
Thread.sleep(5000);
long end = System.currentTimeMillis();
log.info("主程序执行完成耗时{}秒", (end - start)/1000f);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
  @Test
public void doFutureTask(){
try {
long start=System.currentTimeMillis();
Future<String> future1=asyncTask.taskOne();
Future <String> future2 = asyncTask.taskTwo();
Future <String> future3 = asyncTask.taskThree();
//三个任务执行完再执行主程序
do {
Thread.sleep(100);
} while (future1.isDone() && future2.isDone() && future3.isDone());
log.info("获取异步方法的返回值:{}", future1.get());
Thread.sleep(5000);
long end = System.currentTimeMillis();
log.info("主程序执行完成耗时{}秒", (end - start)/1000f);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}

运行结果:可以看到三个任务完成后才执行主程序,还输出了异步方法的返回值。

在这里插入图片描述

3、 Executor

默认情况下,Spring使用SimpleAsyncTaskExecutor异步运行这些方法。

可以在两个级别上重写默认线程池——应用程序级别或方法级别。

3.1、方法级别重写Executor

所需的执行程序需要在配置类中声明 Executor:

@Configuration
@EnableAsync
public class SpringAsyncConfig { @Bean(name = "threadPoolTaskExecutor")
public Executor threadPoolTaskExecutor() {
return new ThreadPoolTaskExecutor();
}
}

然后,在@Async中的属性提供Executor名称:

@Async("threadPoolTaskExecutor")
public void asyncMethodWithConfiguredExecutor() {
System.out.println("Execute method with configured executor - "
+ Thread.currentThread().getName());
}

3.2、应用级别重写Executor

配置类应实现AsyncConfigurer接口,重写getAsyncExecutor()方法。

在这里,我们将返回整个应用程序的Executor,这样一来,它就成为运行以@Async注释的方法的默认Executor:

@Configuration
@EnableAsync
public class SpringApplicationAsyncConfig implements AsyncConfigurer {
@Override
public Executor getAsyncExecutor() {
return new ThreadPoolTaskExecutor();
}
}

3.3、自定义线程池配置

在上面,自定义线程池只是简单地返回了一个线程池:

return new ThreadPoolTaskExecutor();

实际上,还可以对线程池做一些配置:

@Configuration
@EnableAsync
public class SpringPropertiesAsyncConfig implements AsyncConfigurer { /**
* 对线程池进行配置
* @return
*/
@Override
public Executor getAsyncExecutor() {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(20);
taskExecutor.setMaxPoolSize(200);
taskExecutor.setQueueCapacity(25);
taskExecutor.setKeepAliveSeconds(200);
taskExecutor.setThreadNamePrefix("oKong-");
// 线程池对拒绝任务(无线程可用)的处理策略,目前只支持AbortPolicy、CallerRunsPolicy;默认为后者
taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
taskExecutor.initialize();
return taskExecutor;
}
}

ThreadPoolTaskExecutor配置参数的简单说明:

  • corePoolSize:线程池维护线程的最少数量

  • keepAliveSeconds:允许的空闲时间,当超过了核心线程出之外的线程在空闲时间到达之后会被销毁

  • maxPoolSize:线程池维护线程的最大数量,只有在缓冲队列满了之后才会申请超过核心线程数的线程

  • queueCapacity:缓存队列

  • rejectedExecutionHandler:线程池对拒绝任务(无线程可用)的处理策略。这里采用了CallerRunsPolicy策略,当线程池没有处理能力的时候,该策略会直接在 execute 方法的调用线程中运行被拒绝的任务;如果执行程序已关闭,则会丢弃该任务。还有一个是AbortPolicy策略:处理程序遭到拒绝将抛出运行时RejectedExecutionException。

4、异常处理

当方法返回类型为Future时,异常处理很容易– Future.get()方法将抛出异常。

但是如果是无返回值的异步方法,异常不会传播到调用线程。因此,我们需要添加额外的配置来处理异常。

我们将通过实现AsyncUncaughtExceptionHandler接口来创建自定义异步异常处理程序。

当存在任何未捕获的异步异常时,将调用handleUncaughtException()方法:

public class CustomAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {
@Override
public void handleUncaughtException(Throwable throwable, Method method, Object... objects) {
System.out.println("Exception message - " + throwable.getMessage());
System.out.println("Method name - " + method.getName());
for (Object param : objects) {
System.out.println("Parameter value - " + param);
}
}
}

上面,我们使用配置类实现了AsyncConfigurer接口。

作为其中的一部分,我们还需要重写getAsyncUncaughtExceptionHandler()方法以返回我们的自定义异步异常处理:

/**
* 返回自定义异常处理
* @return
*/
@Override
public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
return new CustomAsyncExceptionHandler();
}

5、总结

这里异步请求的使用及相关配置,如超时,异常等处理。在剥离一些和业务无关的操作时,就可以考虑使用异步调用进行其他无关业务操作,以此提供业务的处理效率。或者一些业务场景下可拆分出多个方法进行同步执行又互不影响时,也可以考虑使用异步调用方式提供执行效率。