spring boot使用@Async异步注解的实现原理+源码-FinClip官网

spring boot使用@Async异步注解的实现原理+源码

网友投稿 696 2023-01-07

spring boot使用@Async异步注解的实现原理+源码

1、java的大部分接口的方法都是串行执行的，但是有些业务场景是不需要同步返回结果的，可以把结果直接返回，具体业务异步执行，也有些业务接口是需要并行获取数据，最后把数据聚合在统一返回给前端。

通常我们都是采用多线程的方式来实现上述业务功能，但spring 提供更优雅的方式来实现上述功能，就是@Async 异步注解，在方法上添加@Async，spring就会借助AOP，异步执行方法。

1、如何启用@Async

spring boot通过@EnableAsync 注解启用@Async异步注解

实现AsyncConfigurer接口，getAsyncExecutor是默认自定义的线程池

/**

* 线程池配置（@Async）

@Slf4j

@EnableAsync

@Configuration

public class SimpleExecutorConfig implements AsyncConfigurer {

/** 线程池维护线程的最少数量 */

@Value("${executor.corePoolSize}")

private Integer corePoolSize;

/** 线程池维护线程的最大数量 */

@Value("${executor.maxPoolSize}")

private Integer maxPoolSize;

/** 缓冲队列的大小 */

@Value("${executor.queueCapacity}")

private Integer queueCapacity;

/** 为每个线程名设置一个前缀（1） */

@Value("${executor.threadNamePrefix}")

private String threadNamePrefix;

/** 为每个线程名设置一个前缀（2） */

@Value("${executor.threadNamePrefix_2}")

private String threadNamePrefix_2;

@Bean(ExecutorConstant.simpleExecutor_1)

@Override

public Executor getAsyncExecutor() {

//线程池

ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();

taskExecutor.setCorePoolSize(corePoolSize);

taskExecutor.setMaxPoolSize(maxPoolSize);

taskExecutor.setQueueCapacity(queueCapacity);

taskExecutor.setThreadNamePrefix(threadNamePrefix);

taskExecutor.initialize();

return taskExecutor;

}

@Bean(ExecutorConstant.simpleExecutor_2)

public Executor asyncExecutor2() {

//线程池

ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();

taskExecutor.setCorePoolSize(corePoolSize);

taskExecutor.setMaxPoolSize(maxPoolSize);

taskExecutor.setQueueCapacity(queueCapacity);

taskExecutor.setThreadNamePrefix(threadNamePrefix_2);

taskExecutor.initialize();

return taskExecutor;

}

2、如何使用@Async

下面是代码:

TestAsyncService类：

@Slf4j

@Service

public class TestAsyncService implements ITestAsyncService {

/**

* 异步方法，无返回值

* @return

@Async

@Override

public void asyncFunction_1(){

handleBusinessTime();

log.info("asyncFunction_1 当前线程名称是：{}",Thread.currentThread().getName());

};

/////////////////异步方法，无返回值(指定线程池) start

/**

* 异步方法，无返回值(指定线程池)

* @return

@Async(value = ExecutorConstant.simpleExecutor_2)

@Override

public void asyncFunction_2(){

handleBusinessTime();

log.info("asyncFunction_2 当前线程名称是：{}",Thread.currentThread().getName());

};

@Async(ExecutorConstant.simpleExecutor_2)

@Override

public void asyncFunction_3(){

handleBusinessTime();

log.info("asyncFunction_3 当前线程名称是：{}",Thread.currentThread().getName());

};

/////////////////异步方法，无返回值(指定线程池) end

/**

* 异步方法，有返回值

* @return

@Async

@Override

public Future asyncReturnDta_1(){

handleBusinessTime();

log.info("asyncReturnDta_1 当前线程名称是：{}",Thread.currentThread().getName());

return new AsyncResult(1);

};

/**

* 异步方法，有返回值(指定线程池)

* @return

@Async(ExecutorConstant.simpleExecutor_2)

@Override

public Future asyncReturnDta_2(){

handleBusinessTime();

log.info("asyncReturnDta_2 当前线程名称是：{}",Thread.currentThread().getName());

return new AsyncResult(1);

};

/**

* 异步方法，有返回值-超时

* @return

@Async

@Override

public Future asyncReturnDtaTimeOut(){

handleBusinessTime();

log.info("asyncReturnDta_3 当前线程名称是：{}",Thread.currentThread().getName());

return new AsyncResult(1);

};

/**

* 这方法模拟处理业务或者去操作数据库消耗的时间

public static void handleBusinessTime(){

//去数据库查询数据耗时 start

int[] sleepTime = NumberUtil.generateRandomNumber(2000,5000,1);

try {

//Thread.sleep 休眠的时候相当于业务操作，或者请求数据库的需要消耗的时间

Thread.sleep(sleepTime[0]);

} catch (InterruptedException e) {

e.printStackTrace();

}

//去数据库查询数据耗时 end

}

TestAsyncController

@Slf4j

@RestController

@RequestMapping(value = "/v1/async")

public class TestAsyncController {

@Autowired

ITestAsyncService testAsyncService;

@ApiOperation(value = "调用接口")

@RequestMapping(value = "/test", method = RequestMethod.GET)

public Resp test() throws ExecutionException, InterruptedException {

log.info("asyncFunction_1 start");

testAsyncService.asyncFunction_1();

log.info("asyncFunction_1 start");

log.info("asyncFunction_2 start");

testAsyncService.asyncFunction_2();

log.info("asyncFunction_2 end");

log.info("asyncFunction_3 start");

testAsyncService.asyncFunction_3();

log.info("asyncFunction_3 end");

log.info("asyncReturnDta_1 & asyncReturnDta_2 start");

Future future = testAsyncService.asyncReturnDta_1();

testAsyncService.asyncReturnDta_2();

log.info("asyncReturnDta_1 & asyncReturnDta_2 end");

Integer resp = future.get();

log.info("future.get() resp:{}",resp);

return Resp.buildDataSuccess(resp);

}

@ApiOperation(value = "调用接口-超时")

@RequestMapping(value = "/async_timeOut", method = RequestMethod.GET)

public Resp async_timeOut() throws ExecutionException, InterruptedException {

TimeInterval timeInterval = DateUtil.timer();

log.info("asyncReturnDtaTimeOut start");

Future future = testAsyncService.asyncReturnDtaTimeOut();

log.info("asyncReturnDtaTimeOut end");

Integer resp = null;

try {

//一秒内返回不了数据就报错

resp = future.get(1, TimeUnit.SECONDS);

} catch (TimeoutException e) {

resp = -1;//请求超时了，相当于熔断，服务降级

log.error("asyncReturnDtaTimeOut future.get(1, TimeUnit.SECONDS) timeout:",e);

}

log.info("future.get() resp:{} 耗时：{}毫秒",resp,timeInterval.intervalRestart());

return Resp.buildDataSuccess(resp);

}

/v1/async/test 接口：

2021-06-20 21:09:30.490 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : asyncFunction_1 start

2021-06-20 21:09:30.490 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : asyncFunction_2 start

2021-06-20 21:09:30.491 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : asyncFunction_2 end

2021-06-20 21:09:30.491 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : asyncFunction_3 start

2021-06-20 21:09:30.491 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : asyncFunction_3 end

2021-06-20 21:09:30.491 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : asyncReturnDta_1 & asyncReturnDta_2 start

2021-06-20 21:09:30.492 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : asyncReturnDta_1 & asyncReturnDta_2 end

2021-06-20 21:09:32.679 INFO 14207 --- [le-1-executor-9] c.e.multi.service.impl.TestAsyncService : asyncFunction_1 当前线程名称是：my-simple-1-executor-9

2021-06-20 21:09:33.454 INFO 14207 --- [le-2-executor-8] c.e.multi.service.impl.TestAsyncService : asyncFunction_3 当前线程名称是：my-simple-2-executor-8

2021-06-20 21:09:33.578 INFO 14207 --- [le-2-executor-9] c.e.multi.service.impl.TestAsyncService : asyncReturnDta_2 当前线程名称是：my-simple-2-executor-9

2021-06-20 21:09:34.101 INFO 14207 --- [e-1-executor-10] c.e.multi.service.impl.TestAsyncService : asyncReturnDta_1 当前线程名称是：my-simple-1-executor-10

2021-06-20 21:09:34.102 INFO 14207 --- [nio-8666-exec-7] c.e.m.controller.TestAsyncController : future.get() resp:1

2021-06-20 21:09:34.357 INFO 14207 --- [le-2-executor-7] c.e.multi.service.impl.TestAsyncService : asyncFunction_2 当前线程名称是：my-simple-2-executor-7

从日志上可以看出，都是异步执行的

/v1/async/async_timeOut 接口：

2021-06-20 21:20:58.886 INFO 14427 --- [nio-8666-exec-1] c.e.m.controller.TestAsyncController : asyncReturnDtaTimeOut start

2021-06-20 21:20:58.890 INFO 14427 --- [nio-8666-exec-1] c.e.m.controller.TestAsyncController : asyncReturnDtaTimeOut end

2021-06-20 21:20:59.899 ERROR 14427 --- [nio-8666-exec-1] c.e.m.controller.TestAsyncController : asyncReturnDtaTimeOut future.get(1, TimeUnit.SECONDS) timeout:

java.util.concurrent.TimeoutException: null

at java.util.concurrent.FutureTask.get(FutureTask.java:205) ~[na:1.8.0_231]

at com.example.multi.controller.TestAsyncController.async_timeOut(TestAsyncController.java:69) ~[classes/:na]

at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_231]

at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_231]

at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_231]

at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_231]

at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:197) [spring-web-5.3.8.jar:5.3.8]

at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:141) [spring-web-5.3.8.jar:5.3.8]

at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:106) [spring-webmvc-5.3.8.jar:5.3.8]

at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:894) [spring-webmvc-5.3.8.jar:5.3.8]

at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:808) [spring-webmvc-5.3.8.jar:5.3.8]

at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87) [spring-webmvc-5.3.8.jar:5.3.8]

at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:1063) [spring-webmvc-5.3.8.jar:5.3.8]

at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:963) [spring-webmvc-5.3.8.jar:5.3.8]

at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:1006) [spring-webmvc-5.3.8.jar:5.3.8]

at org.springframework.web.servlet.FrameworkServlet.doGet(FrameworkServlet.java:898) [spring-webmvc-5.3.8.jar:5.3.8]

at javax.servlet.http.HttpServlet.service(HttpServlet.java:626) [tomcat-embed-core-9.0.46.jar:4.0.FR]

at org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:883) [spring-webmvc-5.3.8.jar:5.3.8]

at javax.servlet.http.HttpServlet.service(HttpServlet.java:733) [tomcat-embed-core-9.0.46.jar:4.0.FR]

at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:227) [tomcat-embed-core-9.0.46.jar:9.0.46]