Spring集成Servlet3.0 AsyncRequest
背景
为了提高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 才会记录。
