背景

为了提高tomcat线程利用率，避免tomcat连接被打满增大吞吐量，准备在项目中集成Servlet3.0异步请求。

相关技术

• 异步请求的基石：Servlet3.0 async servlet技术
• Spring相关封装：@Async，Callable，DeferedResult

Async注解

这个注解容易混淆，其实他跟 async servlet 没什么关系。 被 @Async 注解的方法会由 Spring 使用 TaskExecutor 在新线程中异步执行。

Callable，DeferedResult

这两个才是 Spring 中实现异步请求的关键。
@ResponseBody 注解的方法如果返回 Callable 或 DeferedResult ，Spring 会自动替我们将这个请求转换为异步请求。Callable 和 DeferedResult 比较相似，不过 DeferedResult 更为强大，Callable 是异步执行返回结果，而 DeferedResult 更为灵活，甚至可以在另外一个请求中放置响应结果，类似与闭包。

• 如果是简单的使用场景，Controller 直接返回 DeferedResult 就可以变成一个异步接口。
• Spring 处理异步请求的核心类是 RequestMappingHandlerAdapter， 一个容器级的过滤器，想深入研究去追这个类就行。

问题

  Spring 异步请求默认不会使用线程池，虽然经过配置优化（AsyncSupportConfigurer）可以使用，但这个线程池是整个服务共享的，不能做到针对接口进行线程池隔离，避免量大请求影响量小请求。
  为了达到接口级线程隔离的目的，只能自己封装 Servlet API 来实现。大致处理流程如下：
在 Controller 层进行 AOP 拦截，开启线程池执行切面逻辑

request.startAsync();
threadPool.execute(()->{
    ...
    runController...
    writeResponse...
    request.complete();
    ...
})
return null;

  本来这套逻辑跑着挺好，然而当集成 Micrometer 进行 Http 请求监控的时候，Controller 中的接口只有在 400 状态时才被记录 metrics。查看 Micrometer 拦截 Http 的代码：

io.micrometer.spring.web.servlet.WebMvcMetricsFilter.java

    protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain) throws ServletException, IOException {
        HandlerExecutionChain handler = null;
        try {
            MatchableHandlerMapping matchableHandlerMapping = mappingIntrospector.getMatchableHandlerMapping(request);
            if (matchableHandlerMapping != null) {
                handler = matchableHandlerMapping.getHandler(request);
            }
        } catch (Exception e) {
            logger.debug("Unable to time request", e);
            filterChain.doFilter(request, response);
            return;
        }

        final Object handlerObject = handler == null ? null : handler.getHandler();

        // If this is the second invocation of the filter in an async request, we don't
        // want to start sampling again (effectively bumping the active count on any long task timers).
        // Rather, we'll just use the sampling context we started on the first invocation.
        TimingSampleContext timingContext = (TimingSampleContext) request.getAttribute(TIMING_SAMPLE);
        if (timingContext == null) {
            timingContext = new TimingSampleContext(request, handlerObject);
        }

        try {
            filterChain.doFilter(request, response);

            if (request.isAsyncSupported()) {
                // this won't be "started" until after the first call to doFilter
                if (request.isAsyncStarted()) {
                    request.setAttribute(TIMING_SAMPLE, timingContext);
                }
            }

            if (!request.isAsyncStarted()) {
                record(timingContext, response, request,
                    handlerObject, (Throwable) request.getAttribute(DispatcherServlet.EXCEPTION_ATTRIBUTE));
            }
        } catch (NestedServletException e) {
            response.setStatus(HttpStatus.INTERNAL_SERVER_ERROR.value());
            record(timingContext, response, request, handlerObject, e.getCause());
            throw e;
        }
    }

  从代码中可以看到，如果开启了异步，micrometer 并不会记录此次请求，这是出于什么原因呢？ 注意注释中写了： If this is the second invocation of the filter in an async request，也就是说一个异步请求会被 WebMvcMetricsFilter 拦截两次，第一次是异步的，而第二次不是，所以会在第二次被拦截的时候记录这个异步请求。
  为了验证这个说法，clone 了 micrometer 的 simple 运行。结果一个异步请求还真是会被WebMvcMetricsFilter 拦截两次，第一次异步，第二次异步并未打开。

image.png

顺着 RequestMappingHandlerAdapter 找到异步处理核心类:
org.springframework.web.context.request.async.WebAsyncManager.java

    public void startCallableProcessing(final WebAsyncTask<?> webAsyncTask, Object... processingContext)
            throws Exception {
        [...]
        try {
            Future<?> future = this.taskExecutor.submit(() -> {
                Object result = null;
                try {
                    interceptorChain.applyPreProcess(this.asyncWebRequest, callable);
                    result = callable.call();
                }
                catch (Throwable ex) {
                    result = ex;
                }
                finally {
                    result = interceptorChain.applyPostProcess(this.asyncWebRequest, callable, result);
                }
                setConcurrentResultAndDispatch(result);
            });
            interceptorChain.setTaskFuture(future);
        }
        catch (RejectedExecutionException ex) {
            Object result = interceptorChain.applyPostProcess(this.asyncWebRequest, callable, ex);
            setConcurrentResultAndDispatch(result);
            throw ex;
        }
        [...]
    }

    private void setConcurrentResultAndDispatch(Object result) {
        synchronized (WebAsyncManager.this) {
            if (this.concurrentResult != RESULT_NONE) {
                return;
            }
            this.concurrentResult = result;
        }
        [...]
        this.asyncWebRequest.dispatch();
    }

  原来如此，Spring 并没有使用 asyncContext.complete() 来结束这个异步请求，而是使用 dispatch()，dispatch() 中也会把这个异步请求标记成 Completed 状态，所以 WebMvcMetricsFilter 会拦截两次。那就照着这个吧 asyncContext.complete() 换成 asyncContext.dispatch()。
  然而，事与愿违，虽然代码一样，但表现完全不一样，asyncContext.complete() 换成 asyncContext.dispatch()后并没有拦截两次，而且这个请求会循环调用下去，形成一个死循环。仔细分析后发现问题：

• Spring DeferedResult 处理流程:
  request -> RequestHandlerMappingAdaptor 拦截 -> WebMvcMetricsFilter 拦截(is async) -> dispatch() -> RequestHandlerMappingAdaptor 拦截 -> WebMvcMetricsFilter 拦截(not async)
• AOP 封装 Servlet API的处理流程：
  request -> AOP拦截 -> startAsync() -> RequestHandlerMappingAdaptor 拦截 -> WebMvcMetricsFilter 拦截(is async) -> dispatch() -> AOP拦截 -> startAsync() -> RequestHandlerMappingAdaptor 拦截 -> WebMvcMetricsFilter 拦截(is async) -> dispatch() -> ....
  就是由于多了AOP的拦截，dispatch 之后依然会开启新的异步请求，导致在这里死循环。看来必须要标记请求的执行状态，参考 DeferedResult 的处理，

@Slf4j
public class CustomerAsyncWebRequest extends StandardServletAsyncWebRequest {
    private final List<Runnable> errorHandlers = new ArrayList<Runnable>();
    public static final String ASYNC_COMPLETED = "com.xx.async.ASYNC_COMPLETED";

    /**
     * Create a new instance for the given request/response pair.
     *
     * @param request  current HTTP request
     * @param response current HTTP response
     */
    public CustomerAsyncWebRequest(HttpServletRequest request, HttpServletResponse response) {
        super(request, response);
    }

    @Override
    public void onComplete(AsyncEvent event) throws IOException {
        super.onComplete(event);
        this.getRequest().setAttribute(ASYNC_COMPLETED, true);
    }

    // 这个在Spring5.x中已经支持
    @Override
    public void onError(AsyncEvent event) throws IOException {
        log.error("AsyncListener onError ", event.getThrowable().getMessage());
        this.errorHandlers.forEach(Runnable::run);
    }

    // 这个在Spring5.x中已经支持
    public void addErrorHandler(Runnable errorHandler){
        this.errorHandlers.add(errorHandler);
    }
}

AOP逻辑变为

if(customerRequest.isAsyncComplete()){
    return;
}
request.startAsync();
threadPool.execute(()->{
    ...
    runController...
    writeResponse...
    // 这里看着像多余，但是由于AsyncListener的存在，这个判断是必不可少的
    if(customerRequest.isAsyncComplete()){
        return;
    }
    customerRequest.onComplete(null);
    asyncContext.dispatch();
    ...
})
return null;

至此，改造完毕。所以在使用 micrometer 进行监控的时候，如果自己使用 Servlet API 实现异步请求，不能使用 complete() 来结束，必须使用 dispatch() 结束请求，micrometer 才会记录。