Fork me on GitHub

rookiedev

just do it

0%

接口采用异步的方式执行

Posted on Edited on In

不需要返回结果的接口

1
2
3
4
5
6
@GetMapping("asyncEndpoint")
public void asyncEndpoint(){
    log.info("========== received async endpoint request");
    this.taskService.asyncExecute();
    log.info("========== the servlet thread that received the request has been released");
}
1
2
3
4
5
6
7
8
@SneakyThrows
@Async
@Override
public void asyncExecute() {
    log.info("========== task async execute start");
    TimeUnit.SECONDS.sleep(5);
    log.info("========== task async execute end");
}

上面 asyncEndpoint 是一个不需要返回结果的接口，主要是通过 @Async 注解来实现的，请求到了后台，从 Tomcat 服务器处理请求的线程池中获取一个线程 A 处理该请求，在该线程中执行到调用 asyncExecute 方法的时候，由于 asyncExecute 方法的上面由 @Async 注解所标注，这样的话该方法会以异步的方式来执行，也就是说在 asyncEndpoint 方法中调用 asyncExecute 方法以异步的方式执行，asyncExecute 方法中的代码在另一个线程 B 中执行，线程 A 继续往下执行，打印完最后一行日志之后，整个请求就结束了，线程 A 也就释放了，可以继续处理其他的请求了。 而比较耗时的方法 asyncExecute 继续在线程 B 中执行。

请求 asyncEndpoint 接口，通过以下日志可以看出来：

asyncEndpoint_log

有返回值的接口之 Callable 作为返回值

1
2
3
4
5
6
7
@GetMapping("callableEndpoint")
public Callable<String> callableEndpoint(){
    log.info("========== received callable endpoint request");
    Callable<String> callable = this.taskService ::execute;
    log.info("========== the servlet thread that received the request has been released");
    return callable;
}
1
2
3
4
5
6
7
8
@SneakyThrows
@Override
public String execute() {
    log.info("========== task execute start");
    TimeUnit.SECONDS.sleep(5);
    log.info("========== task execute end");
    return "task execute success";
}

上面 callableEndpoint 接口，它的返回值是 Callable，这里就需要注意了，他在这里同样也会以异步的方式将结果返回给客户端请求，但这里和上面的这种方式又有所不同，这种方式稍微复杂一点。上面的方式更多的是由于我们不关心方法执行的返回值，我们自己在业务逻辑层面来控制方法以异步的方式执行，然后处理请求的线程执行结束，应用程序线程继续异步执行。

而在这里，主要是利用了 Servlet 3.0 中引入了对于异步的支持，至于 Servlet 3.0 中引入的对异步的支持，下次再用一篇文章详细介绍下，我们先大概了解以 Callable 作为返回值的接口为什么会以异步的方式执行。

这里首先后台收到 asyncEndpoint 请求后，同样从 Tomcat 容器处理请求的线程池中获取一个线程处理该请求。这里大概说下，我们是使用的 SpringMVC 的那一套请求执行流程，SpringMVC 中有一个 DispatcherServlet 类里面的 doDispatch 方法是请求的统一入口，通过它来协调其它组件来处理前端请求。

比如说通过处理器映射器 HandlerMapping 查找对应的 Handler 处理器 Controller，然后再通过请求适配器 HandlerAdapter 对处理器进行执行，在处理器执行之后得到一个返回值结果，SpringMVC 里面有一些处理返回值的 Handler 来对结果进行解析，根据返回值类型选取合适的 returnValueHandler 来处理。下面是直接到了处理器执行获取执行结果的代码。

handler_return_value

至于选择处理范湖结果的 handler 的过程通过下面代码可以看出，是根据返回值的类型来从已经注册的 returnValueHandler 选择能处理当前返回值类型的 handler，

handler_select_handler

select_handler

support_return_value

从上面第三段代码中可以看到由于返回值类型 Callable，最终会匹配到 CallableMethodReturnValueHandler 这个返回值处理器，在这个类中还有一个方法 handleReturnValue，也就是在选择好匹配的 returnValueHandler 之后调用的那个方法，异步的处理也就是在这个方法的最后一行代码中，

1
WebAsyncUtils.getAsyncManager(webRequest).startCallableProcessing(callable, mavContainer);

这行代码先是调用 getAsyncManager 获取到 WebAsyncManager，然后调用 startCallableProcessing 开始 Callable 结果的处理流程，而在 startCallableProcessing 方法里面又是调用了一个内部重载的方法，通过 WebAsyncTask 包装了以下 Callable 作为参数

startCallableProcessing

set_concurrent_result_dispatch

在 startCallableProcessing 方法中可以清楚的看到将 callable 交给 taskExecutor 来执行，然后设置并发执行得到的结果并且调用 dispatch 方法将请求再次分派到容器，以便应用程序线程中并发执行之后的恢复处理，其实也就是将并发执行的结果再次交给容器线程来恢复刚才的请求，将刚才的流程再执行一次。

只不过在处理器执行的时候调用的不再是 Contrller 里面的方法，而是直接调用 Callable 的 call 方法拿刚才执行的结果，等到再执行到 handleReturnValue 方法的时候，这时由于 return value 是一个字符串，所以匹配到的 returnValueHandler 是 RequestResponseBodyMethodProcessor,这时会将 callable 的执行结果返回给客户端，本次请求结束。

需要注意的是，应用程序的异步任务的执行时间过长会导致超时的问题，还记得上面的 startCallableProcessing 方法里面有 addTimeoutHandler，addErrorHandler，addCompletionHandler 三个回调，分别是超时，发生错误以及异步执行完成的回调处理。里面其实会去读 WebAsyncTask 中设置的超时时间，但里面没有设置，但我测试是有一个默认的超时时间 30s(我通过测试，发现 30s 没执行完就会 timeout)。

还有一个就是需要注意的是在整个的处理过程中，服务端和客户端的连接是一直保持着的，所以需要注意连接超时的问题，虽然默认没有设置的话是无限的，但如果设置了就需要注意了，这个的超时时间必然要比异步执行里面的超时时间要大一些。

上面的 taskExecutor 是 SpringBoot 中 TaskExecutionAutoConfiguration 自动配置的 ThreadPoolTaskExecutor，对于这个类的配置也可以在配置文件中进行配置，线程数，queue capacity 等。

最终的执行打印日志如下：
callableEndpoint_log

有返回值的接口之 WebAsyncTask 作为返回值

1
2
3
4
5
6
7
8
9
10
11
12
13
@GetMapping("webAsyncTaskEndpoint")
public WebAsyncTask<String> webAsyncTaskEndpoint(){
    log.info("========== received web async task endpoint request");
    Callable<String> callable = () -> this.taskService.execute();
    WebAsyncTask<String> webAsyncTask = new WebAsyncTask<>(3000, callable);
    webAsyncTask.onCompletion(() -> log.info("========== execute finish"));

    webAsyncTask.onTimeout(() -> "execute timeout");

    webAsyncTask.onError(() -> "execute error");
    log.info("========== the servlet thread that received the request has been released");
    return webAsyncTask;
}

在理解了 Callable 作为接口返回值之后，就能够很好地理解这个流程了，大体的流程是一样的，只不过就是返回值的 Handler 处理器不一样，这里用到的处理器是 AsyncTaskMethodReturnValueHandler，在这个类中 handleReturnValue 方法调用的 startCallableProcessing 传的参数就是 Controller 里面返回的这个 webAsyncTask，这个 webAsyncTask 定义了三个回调函数，方便做特殊的处理。

这里需要注意的是 onError 的回调并不是说异步任务里面执行出错了的回调，其实异步任务里面的出错的话已经 try catch 了，并且将异常信息作为结果返回，所以说如果异步任务里面抛异常了，并不会回调到这里，而且这里回调的是 onCompletion，只不过得到的结果是抛出的异常。至于这里的 onError 是哪里出现异常的回调呢，看 WebAsyncTask 类的 onError 方法上面的 doc 描述有一段这样的话：

1
This method is called from a container thread when an error occurred while processing an async request before the {@code Callable} has completed

看意思是在异步任务执行完成之前，处理异步请求的过程中发生的异常会回调，具体有待测试。

所以说 WebAsyncTask 作为接口返回值的话，扩展性更强一些，提供了更完善的超时，异常，异步执行完成的回调处理。

最终的执行打印日志如下:

webAsyncTaskEndpoint_log

有返回值的接口之 DeferredResult 作为返回值

1
2
3
4
5
6
7
8
9
@GetMapping("deferredResultEndpoint")
public DeferredResult<String> deferredResultEndpoint(){
    log.info("========== received deferred result endpoint request");
    DeferredResult<String> deferredResult = new DeferredResult<>();
    CompletableFuture.supplyAsync(this.taskService ::execute)
            .whenCompleteAsync((result, throwable) -> deferredResult.setResult(result));
    log.info("========== the servlet thread that received the request has been released");
    return deferredResult;
}

上面这个接口的返回值是 DeferredResult，那么请求的处理流程和 Callable 是一样的，不同的地方也是在于
返回值的 Handler 处理器不一样，这里用到的处理器是 DeferredResultMethodReturnValueHandler，而且这个类里面的 handleReturnValue 方法和上面的 CallableMethodReturnValueHandler 以及 AsyncTaskMethodReturnValueHandler 方法又不一样，这里调用的是 WebAsyncManager 类中 startDeferredResultProcessing 方法

startDeferredResultProcessing

startDeferredResultProcessing 方法和 startCallableProcessing 差不多，前面都是设置超时时间，设置超时，错误，完成的回调，最后有点不同，在 startCallableProcessing 方法里面是开始通过 taskExecutor 来执行异步任务。

而在这里是为 deferredResult 设置了一个 resultHandler，然后在 resultHandler 里面调用 setConcurrentResultAndDispatch，来设置结果并且将请求再次分派到容器恢复请求过程返回结果给客户端。至于这里的 resultHandler 的用处是 Controller 里面异步任务执行完成后为 deferredResult 设置结果的时候调用 setResult 方法里面会用到。

setResult

从上面可以看出在设置结果之后，会调用在 startDeferredResultProcessing 设置好的 resultHandler ，也就是执行 setConcurrentResultAndDispatch，到这里就又和上面的流程一样了，调用 dispatch 方法将请求再次分派到容器，执行之后的恢复过程返回结果给客户端。

对于 DeferredResult 作为接口返回值还有个不同的地方就是异步任务的执行完全是由开发人员自己控制了，而不是像之前的那样交给应用程序默认的 ThreadPoolTaskExecutor 来执行，这其实就给开发人员很大的灵活性了。

采用这种异步方式的好处：
一般来说，我们写的更多的是阻塞同步式的接口，但有时候如果我们碰到比较耗时任务的接口，这时候我们如果还采用阻塞同步式的写法的话，那么前端请求到后台，首先 Tomcat 服务器从处理请求的线程池中获取一个线程 A 来处理该请求，

然后由于这个接口要执行的任务耗时比较长，耗时任务一直在线程 A 中执行，直到任务执行完成才得到释放，对于并发量比较小的程序来说那还好，线程占着也就占着了，但如果是并发量比较大的程序，容器中大量的处理请求的线程将一直被占用，如果突然请求的高峰到来，那么容器中处理请求的线程有可能被耗尽，就不能再处理更多请求了，也就是说这严重影响了系统的并发吞吐量。

而如果我们采取的是异步的方式来处理的话，这时就像我们前面所介绍的一样，请求到了后台，由于是异步的方式，处理请求的线程在处理完请求之后就被释放了，可以继续处理其他请求，等到异步任务执行完之后再移交给容器线程返回结果给客户端，这样的话就很大的提高了系统的并发量。

最终的执行打印日志如下：
deferredResultEndpoint_log

各自的适用场景，

首先一个前提就是要执行的任务是耗时的任务，或者需要等第三方的通知的任务

  1. 通过 @Async 注解的方式
  • 不需要关注任务执行的结果
  1. 以 Callable 作为接口的返回值
  • 需要知道任务执行的结果，并且将该结果返回给前端
  1. 以 WebAsyncTask 作为接口的返回值
  • 需要知道任务执行的结果，并且将该结果返回给前端，同时希望控制异步任务的超时时间，任务执行结束时要做特殊处理，以及捕获处理请求的过程中发生了异常的回调。
  1. 以 DeferredResult 作为接口的返回值

  • 需要知道任务执行的结果，并且将该结果返回给前端，同时希望控制异步任务的超时时间，任务执行结束时要做特殊处理，以及捕获处理请求的过程中发生了异常的回调。

  • 需要控制异步任务的执行，自己维护处理返回结果的线程。

比如说，接口需要等第三方服务的处理结果才能设置返回给前端的结果。Client 请求服务 A，服务 A 通过消息队列发送任务给服务 B，同时监听服务 B 的处理结果队列，服务 B 处理完成之后将结果塞回给服务 A 的结果队列，服务 A 收到结果后再返回给 Client，这个流程就很适合用采用 DeferredResult 作为接口的返回值。

wechat
扫描上面图中二维码关注微信公众号