您的当前位置：首页正文

Spring WebMVC 源码分析(3)-异步请求 DeferredResult 的原理

2024-11-10 来源：个人技术集锦

1. DeferredResult 简介

在 Spring WebMVC 中有多种方式可以用来进行异步请求处理，比较常见的是控制器方法返回DeferredResult类型的返回值。DeferredResult适用于处理客户端轮询的场景，可以实现延时响应客户端的效果，有效避免轮询请求过于频繁造成服务器压力，其使用流程如下。注意，Spring WebFlux 本就将建立连接和请求处理分离开，天然异步，所以未支持 DeferredResult 特性

2. DeferredResult 实现原理

Spring WebMVC 在调用处理器方法后会对返回值做处理，如果发现返回值为异步请求类型，则不会立即响应客户端，而是直接将请求挂起，中断当前请求的处理流程，直到异步任务超时或者被设置了结果值才响应客户端。这样处理请求的线程将被释放，可以用于处理新的请求，从而提高服务端的吞吐量。在这个过程中，异步请求的核心处理流程大致可以分为两步：

  1. 请求的异步处理
  2. 请求的再次分发

2.1 请求的异步处理

  1. 在 中笔者分析了请求进入 Spring 框架的处理流程，此处为保证逻辑完整，依然从 DispatcherServlet#doDispatch() 开始分析。这个方法是请求处理的主干逻辑，不过本文主要关注以下几点：

    1. WebAsyncUtils.getAsyncManager() 根据 HttpServletRequest 创建或者从缓存中获取 WebAsyncManager 对象
    2. ha.handle() 进入处理器方法被调用的逻辑
    3. asyncManager.isConcurrentHandlingStarted() 判断是否开始了异步处理，是的话直接 return，不进入响应处理的流程，也就是说异步请求第一次处理在此就中断了
    protected void doDispatch(HttpServletRequest request, HttpServletResponse response) throws Exception {
 	HttpServletRequest processedRequest = request;
 	HandlerExecutionChain mappedHandler = null;
 	boolean multipartRequestParsed = false;

 	WebAsyncManager asyncManager = WebAsyncUtils.getAsyncManager(request);

 	try {
 		ModelAndView mv = null;
 		Exception dispatchException = null;

 		try {
 			processedRequest = checkMultipart(request);
 			multipartRequestParsed = (processedRequest != request);

 			// Determine handler for the current request.
 			mappedHandler = getHandler(processedRequest);
 			if (mappedHandler == null) {
 				noHandlerFound(processedRequest, response);
 				return;
 			}

 			// Determine handler adapter for the current request.
 			HandlerAdapter ha = getHandlerAdapter(mappedHandler.getHandler());

 			// Process last-modified header, if supported by the handler.
 			String method = request.getMethod();
 			boolean isGet = "GET".equals(method);
 			if (isGet || "HEAD".equals(method)) {
 				long lastModified = ha.getLastModified(request, mappedHandler.getHandler());
 				if (new ServletWebRequest(request, response).checkNotModified(lastModified) && isGet) {
 					return;
 				}
 			}

 			if (!mappedHandler.applyPreHandle(processedRequest, response)) {
 				return;
 			}

 			// Actually invoke the handler.
 			mv = ha.handle(processedRequest, response, mappedHandler.getHandler());

 			if (asyncManager.isConcurrentHandlingStarted()) {
 				return;
 			}

 			applyDefaultViewName(processedRequest, mv);
 			mappedHandler.applyPostHandle(processedRequest, response, mv);
 		}
 		catch (Exception ex) {
 			dispatchException = ex;
 		}
 		catch (Throwable err) {
 			// As of 4.3, we're processing Errors thrown from handler methods as well,
 			// making them available for @ExceptionHandler methods and other scenarios.
 			dispatchException = new NestedServletException("Handler dispatch failed", err);
 		}
 		processDispatchResult(processedRequest, response, mappedHandler, mv, dispatchException);
 	}
 	catch (Exception ex) {
 		triggerAfterCompletion(processedRequest, response, mappedHandler, ex);
 	}
 	catch (Throwable err) {
 		triggerAfterCompletion(processedRequest, response, mappedHandler,
 				new NestedServletException("Handler processing failed", err));
 	}
 	finally {
 		if (asyncManager.isConcurrentHandlingStarted()) {
 			// Instead of postHandle and afterCompletion
 			if (mappedHandler != null) {
 				mappedHandler.applyAfterConcurrentHandlingStarted(processedRequest, response);
 			}
 		}
 		else {
 			// Clean up any resources used by a multipart request.
 			if (multipartRequestParsed) {
 				cleanupMultipart(processedRequest);
 			}
 		}
 	}
 }

  2. WebAsyncUtils#getAsyncManager()方法如下，可以看到主要逻辑是从 ServletRequest 对象的属性中获取缓存的 WebAsyncManager 对象或者直接创建新的 WebAsyncManager 对象并缓存。此处需要注意，之所以需要缓存WebAsyncManager 对象，是因为异步请求将会被二次分发处理，且后续流程都将用到WebAsyncManager 对象，需要保证请求的相关信息一致

    public static WebAsyncManager getAsyncManager(ServletRequest servletRequest) {
 	WebAsyncManager asyncManager = null;
 	Object asyncManagerAttr = servletRequest.getAttribute(WEB_ASYNC_MANAGER_ATTRIBUTE);
 	if (asyncManagerAttr instanceof WebAsyncManager) {
 		asyncManager = (WebAsyncManager) asyncManagerAttr;
 	}
 	if (asyncManager == null) {
 		asyncManager = new WebAsyncManager();
 		servletRequest.setAttribute(WEB_ASYNC_MANAGER_ATTRIBUTE, asyncManager);
 	}
 	return asyncManager;
 }

  3. 回到步骤1 ha.handle() 的调用，此处首先调用到 AbstractHandlerMethodAdapter#handle()方法，接着调用子类实现 RequestMappingHandlerAdapter#handleInternal() 方法，可以看到核心逻辑其实是执行 RequestMappingHandlerAdapter#invokeHandlerMethod() 方法

    protected ModelAndView handleInternal(HttpServletRequest request,
 		HttpServletResponse response, HandlerMethod handlerMethod) throws Exception {

 	ModelAndView mav;
 	checkRequest(request);

 	// Execute invokeHandlerMethod in synchronized block if required.
 	if (this.synchronizeOnSession) {
 		HttpSession session = request.getSession(false);
 		if (session != null) {
 			Object mutex = WebUtils.getSessionMutex(session);
 			synchronized (mutex) {
 				mav = invokeHandlerMethod(request, response, handlerMethod);
 			}
 		}
 		else {
 			// No HttpSession available -> no mutex necessary
 			mav = invokeHandlerMethod(request, response, handlerMethod);
 		}
 	}
 	else {
 		// No synchronization on session demanded at all...
 		mav = invokeHandlerMethod(request, response, handlerMethod);
 	}

 	if (!response.containsHeader(HEADER_CACHE_CONTROL)) {
 		if (getSessionAttributesHandler(handlerMethod).hasSessionAttributes()) {
 			applyCacheSeconds(response, this.cacheSecondsForSessionAttributeHandlers);
 		}
 		else {
 			prepareResponse(response);
 		}
 	}

 	return mav;
 }

  4. RequestMappingHandlerAdapter#invokeHandlerMethod() 方法非常重要，此处是处理器方法被执行的触发点，不过本文主要关注以下几点：

    1. 调用RequestMappingHandlerAdapter#createInvocableHandlerMethod() 创建处理器的适配对象 ServletInvocableHandlerMethod
    2. WebAsyncUtils.createAsyncWebRequest() 新建异步请求对象 StandardServletAsyncWebRequest
    3. 从 ServletRequest 请求对象的缓存中获取 AsyncManager 对象，并对其进行配置
    4. asyncManager.hasConcurrentResult() 判断异步请求是否已经有结果了，此处逻辑将在异步请求二次分发处理时进入，暂且不表
    5. invocableMethod.invokeAndHandle() 实际调用处理器方法
    protected ModelAndView invokeHandlerMethod(HttpServletRequest request,
 		HttpServletResponse response, HandlerMethod handlerMethod) throws Exception {

 	ServletWebRequest webRequest = new ServletWebRequest(request, response);
 	try {
 		WebDataBinderFactory binderFactory = getDataBinderFactory(handlerMethod);
 		ModelFactory modelFactory = getModelFactory(handlerMethod, binderFactory);

 		ServletInvocableHandlerMethod invocableMethod = createInvocableHandlerMethod(handlerMethod);
 		if (this.argumentResolvers != null) {
 			invocableMethod.setHandlerMethodArgumentResolvers(this.argumentResolvers);
 		}
 		if (this.returnValueHandlers != null) {
 			invocableMethod.setHandlerMethodReturnValueHandlers(this.returnValueHandlers);
 		}
 		invocableMethod.setDataBinderFactory(binderFactory);
 		invocableMethod.setParameterNameDiscoverer(this.parameterNameDiscoverer);

 		ModelAndViewContainer mavContainer = new ModelAndViewContainer();
 		mavContainer.addAllAttributes(RequestContextUtils.getInputFlashMap(request));
 		modelFactory.initModel(webRequest, mavContainer, invocableMethod);
 		mavContainer.setIgnoreDefaultModelOnRedirect(this.ignoreDefaultModelOnRedirect);

 		AsyncWebRequest asyncWebRequest = WebAsyncUtils.createAsyncWebRequest(request, response);
 		asyncWebRequest.setTimeout(this.asyncRequestTimeout);

 		WebAsyncManager asyncManager = WebAsyncUtils.getAsyncManager(request);
 		asyncManager.setTaskExecutor(this.taskExecutor);
 		asyncManager.setAsyncWebRequest(asyncWebRequest);
 		asyncManager.registerCallableInterceptors(this.callableInterceptors);
 		asyncManager.registerDeferredResultInterceptors(this.deferredResultInterceptors);

 		if (asyncManager.hasConcurrentResult()) {
 			Object result = asyncManager.getConcurrentResult();
 			mavContainer = (ModelAndViewContainer) asyncManager.getConcurrentResultContext()[0];
 			asyncManager.clearConcurrentResult();
 			LogFormatUtils.traceDebug(logger, traceOn -> {
 				String formatted = LogFormatUtils.formatValue(result, !traceOn);
 				return "Resume with async result [" + formatted + "]";
 			});
 			invocableMethod = invocableMethod.wrapConcurrentResult(result);
 		}

 		invocableMethod.invokeAndHandle(webRequest, mavContainer);
 		if (asyncManager.isConcurrentHandlingStarted()) {
 			return null;
 		}

 		return getModelAndView(mavContainer, modelFactory, webRequest);
 	}
 	finally {
 		webRequest.requestCompleted();
 	}
 }

  5. ServletInvocableHandlerMethod#invokeAndHandle()的重要处理步骤如下：

    1. ServletInvocableHandlerMethod#invokeForRequest()反射调用处理器方法，获得方法返回值
    2. this.returnValueHandlers.handleReturnValue() 使用返回值处理器处理返回值
    	public void invokeAndHandle(ServletWebRequest webRequest, ModelAndViewContainer mavContainer,
 		Object... providedArgs) throws Exception {

 	Object returnValue = invokeForRequest(webRequest, mavContainer, providedArgs);
 	setResponseStatus(webRequest);

 	if (returnValue == null) {
 		if (isRequestNotModified(webRequest) || getResponseStatus() != null || mavContainer.isRequestHandled()) {
 			disableContentCachingIfNecessary(webRequest);
 			mavContainer.setRequestHandled(true);
 			return;
 		}
 	}
 	else if (StringUtils.hasText(getResponseStatusReason())) {
 		mavContainer.setRequestHandled(true);
 		return;
 	}

 	mavContainer.setRequestHandled(false);
 	Assert.state(this.returnValueHandlers != null, "No return value handlers");
 	try {
 		this.returnValueHandlers.handleReturnValue(
 				returnValue, getReturnValueType(returnValue), mavContainer, webRequest);
 	}
 	catch (Exception ex) {
 		if (logger.isTraceEnabled()) {
 			logger.trace(formatErrorForReturnValue(returnValue), ex);
 		}
 		throw ex;
 	}
 }

  6. this.returnValueHandlers.handleReturnValue() 实际是调用到返回值处理器的聚合类 HandlerMethodReturnValueHandlerComposite#handleReturnValue() 方法，其主要逻辑如下：

    1. 调用 selectHandler() 方法从聚合类对象内部的返回值处理器列表中查找到能够处理当前返回值的处理器，判断依据为 HandlerMethodReturnValueHandler#supportsReturnType() 的返回值
    2. handler.handleReturnValue() 调用返回值处理器进行返回值的处理
    public void handleReturnValue(@Nullable Object returnValue, MethodParameter returnType,
 		ModelAndViewContainer mavContainer, NativeWebRequest webRequest) throws Exception {

 	HandlerMethodReturnValueHandler handler = selectHandler(returnValue, returnType);
 	if (handler == null) {
 		throw new IllegalArgumentException("Unknown return value type: " + returnType.getParameterType().getName());
 	}
 	handler.handleReturnValue(returnValue, returnType, mavContainer, webRequest);
 }

 private HandlerMethodReturnValueHandler selectHandler(@Nullable Object value, MethodParameter returnType) {
 	boolean isAsyncValue = isAsyncReturnValue(value, returnType);
 	for (HandlerMethodReturnValueHandler handler : this.returnValueHandlers) {
 		if (isAsyncValue && !(handler instanceof AsyncHandlerMethodReturnValueHandler)) {
 			continue;
 		}
 		if (handler.supportsReturnType(returnType)) {
 			return handler;
 		}
 	}
 	return null;
 }

  7. 对于类型为 DeferredResult 的返回值，对应的处理器为 DeferredResultMethodReturnValueHandler，则DeferredResultMethodReturnValueHandler#handleReturnValue() 方法将被调用，可以看到这里主要的处理步骤如下：

    1. 对于类型为 ListenableFuture 或者 CompletionStage 的返回值，需要将其包装为 DeferredResult
    2. WebAsyncUtils.getAsyncManager(webRequest).startDeferredResultProcessing() 开始进行异步的 DeferredResult 处理，这一步实际调用到 AsyncManager#startDeferredResultProcessing() 方法
    	public boolean supportsReturnType(MethodParameter returnType) {
 	Class<?> type = returnType.getParameterType();
 	return (DeferredResult.class.isAssignableFrom(type) ||
 			ListenableFuture.class.isAssignableFrom(type) ||
 			CompletionStage.class.isAssignableFrom(type));
 }

public void handleReturnValue(@Nullable Object returnValue, MethodParameter returnType,
 		ModelAndViewContainer mavContainer, NativeWebRequest webRequest) throws Exception {

 	if (returnValue == null) {
 		mavContainer.setRequestHandled(true);
 		return;
 	}

 	DeferredResult<?> result;

 	if (returnValue instanceof DeferredResult) {
 		result = (DeferredResult<?>) returnValue;
 	}
 	else if (returnValue instanceof ListenableFuture) {
 		result = adaptListenableFuture((ListenableFuture<?>) returnValue);
 	}
 	else if (returnValue instanceof CompletionStage) {
 		result = adaptCompletionStage((CompletionStage<?>) returnValue);
 	}
 	else {
 		// Should not happen...
 		throw new IllegalStateException("Unexpected return value type: " + returnValue);
 	}

 	WebAsyncUtils.getAsyncManager(webRequest).startDeferredResultProcessing(result, mavContainer);
 }

  8. WebAsyncManager#startDeferredResultProcessing() 方法是 DeferredResult 处理的重点，此处关键步骤如下：

    1. 首先将 DeferredResult 设置的超时时间取出来，设置到 AsyncWebRequest 对象中
    2. 创建 DeferredResult 的拦截器执行链对象 DeferredResultInterceptorChain，将 DeferredResult 中设置的各个回调方法保存起来
    3. 设置 AsyncWebRequest 对象的各个处理器，包括超时处理器，异常处理器等
    4. 调用 WebAsyncManager#startAsyncProcessing() 开始进行请求异步处理，这个步骤在上层比较简短，就是调用this.asyncWebRequest.startAsync()，实际调用到 StandardServletAsyncWebRequest#startAsync() 方法
    5. deferredResult.setResultHandler() 设置 DeferredResult 的结果处理器，需注意此处传入的是 Lambda 函数
    public void startDeferredResultProcessing(
 		final DeferredResult<?> deferredResult, Object... processingContext) throws Exception {

 	Assert.notNull(deferredResult, "DeferredResult must not be null");
 	Assert.state(this.asyncWebRequest != null, "AsyncWebRequest must not be null");

 	Long timeout = deferredResult.getTimeoutValue();
 	if (timeout != null) {
 		this.asyncWebRequest.setTimeout(timeout);
 	}

 	List<DeferredResultProcessingInterceptor> interceptors = new ArrayList<>();
 	interceptors.add(deferredResult.getInterceptor());
 	interceptors.addAll(this.deferredResultInterceptors.values());
 	interceptors.add(timeoutDeferredResultInterceptor);

 	final DeferredResultInterceptorChain interceptorChain = new DeferredResultInterceptorChain(interceptors);

 	this.asyncWebRequest.addTimeoutHandler(() -> {
 		try {
 			interceptorChain.triggerAfterTimeout(this.asyncWebRequest, deferredResult);
 		}
 		catch (Throwable ex) {
 			setConcurrentResultAndDispatch(ex);
 		}
 	});

 	this.asyncWebRequest.addErrorHandler(ex -> {
 		if (!this.errorHandlingInProgress) {
 			try {
 				if (!interceptorChain.triggerAfterError(this.asyncWebRequest, deferredResult, ex)) {
 					return;
 				}
 				deferredResult.setErrorResult(ex);
 			}
 			catch (Throwable interceptorEx) {
 				setConcurrentResultAndDispatch(interceptorEx);
 			}
 		}
 	});

 	this.asyncWebRequest.addCompletionHandler(()
 			-> interceptorChain.triggerAfterCompletion(this.asyncWebRequest, deferredResult));

 	interceptorChain.applyBeforeConcurrentHandling(this.asyncWebRequest, deferredResult);
 	startAsyncProcessing(processingContext);

 	try {
 		interceptorChain.applyPreProcess(this.asyncWebRequest, deferredResult);
 		deferredResult.setResultHandler(result -> {
 			result = interceptorChain.applyPostProcess(this.asyncWebRequest, deferredResult, result);
 			setConcurrentResultAndDispatch(result);
 		});
 	}
 	catch (Throwable ex) {
 		setConcurrentResultAndDispatch(ex);
 	}
 }

 private void startAsyncProcessing(Object[] processingContext) {
 	synchronized (WebAsyncManager.this) {
 		this.concurrentResult = RESULT_NONE;
 		this.concurrentResultContext = processingContext;
 		this.errorHandlingInProgress = false;
 	}
 	this.asyncWebRequest.startAsync();

 	if (logger.isDebugEnabled()) {
 		logger.debug("Started async request");
 	}
 }

  9. StandardServletAsyncWebRequest#startAsync() 方法主要的处理分为两步：

    1. getRequest().startAsync(getRequest(), getResponse()) 创建 AsyncContext 对象，此处实际进入到了 Tomcat 服务器代码，最终调用到连接器层的 org.apache.catalina.connector.Request#startAsync() 方法
    2. this.asyncContext.addListener(this) 将当前 StandardServletAsyncWebRequest 对象作为监听器添加到 AsyncContext 对象中，并将上层的超时时间设置到 AsyncContext 内部
    public void startAsync() {
 	Assert.state(getRequest().isAsyncSupported(),
 			"Async support must be enabled on a servlet and for all filters involved " +
 			"in async request processing. This is done in Java code using the Servlet API " +
 			"or by adding \"<async-supported>true</async-supported>\" to servlet and " +
 			"filter declarations in web.xml.");
 	Assert.state(!isAsyncComplete(), "Async processing has already completed");

 	if (isAsyncStarted()) {
 		return;
 	}
 	this.asyncContext = getRequest().startAsync(getRequest(), getResponse());
 	this.asyncContext.addListener(this);
 	if (this.timeout != null) {
 		this.asyncContext.setTimeout(this.timeout);
 	}
 }

  10. org.apache.catalina.connector.Request#startAsync() 方法简单明了，主要的处理如下：

    1. 新建 AsyncContextImpl 对象，并使用该对象调用 AsyncContextImpl#setStarted() 通知服务器底层当前请求为开始异步处理状态
    2. 使用AsyncContextImpl 对象调用 AsyncContextImpl#setTimeout() 将连接层默认的超时时间作为异步请求的超时时间
    public AsyncContext startAsync(ServletRequest request,
        ServletResponse response) {
    if (!isAsyncSupported()) {
        IllegalStateException ise =
                new IllegalStateException(sm.getString("request.asyncNotSupported"));
        log.warn(sm.getString("coyoteRequest.noAsync",
                StringUtils.join(getNonAsyncClassNames())), ise);
        throw ise;
    }

    if (asyncContext == null) {
        asyncContext = new AsyncContextImpl(this);
    }

    asyncContext.setStarted(getContext(), request, response,
            request==getRequest() && response==getResponse().getResponse());
    asyncContext.setTimeout(getConnector().getAsyncTimeout());

    return asyncContext;
}

  11. AsyncContextImpl#setStarted()方法将保存异步请求的上下文，并负责与底层服务器进行交互，重要的处理如下：

    1. this.request.getCoyoteRequest().action() 将当前请求开始异步的状态通知到底层，并将自身作为监听器注册到服务器层，此处的请求对象实际为org.apache.coyote.Request
    2. AsyncContextImpl 作为服务器底层与上层交互的中间层，会将底层请求的状态变化通知到上层，体现在代码中也就是遍历内部的监听器列表，一一回调监听器的方法
    public void setStarted(Context context, ServletRequest request,
       ServletResponse response, boolean originalRequestResponse) {

   synchronized (asyncContextLock) {
       this.request.getCoyoteRequest().action(
               ActionCode.ASYNC_START, this);

       this.context = context;
       context.incrementInProgressAsyncCount();
       this.servletRequest = request;
       this.servletResponse = response;
       this.hasOriginalRequestAndResponse = originalRequestResponse;
       this.event = new AsyncEvent(this, request, response);

       List<AsyncListenerWrapper> listenersCopy = new ArrayList<>(listeners);
       listeners.clear();
       if (log.isDebugEnabled()) {
           log.debug(sm.getString("asyncContextImpl.fireOnStartAsync"));
       }
       for (AsyncListenerWrapper listener : listenersCopy) {
           try {
               listener.fireOnStartAsync(event);
           } catch (Throwable t) {
               ExceptionUtils.handleThrowable(t);
               log.warn(sm.getString("asyncContextImpl.onStartAsyncError",
                       listener.getClass().getName()), t);
           }
       }
   }
}

  12. org.apache.coyote.Request#action() 方法其实只是负责分发动作事件，最终的处理还是在 AbstractProcessor#action() 中

     public void action(ActionCode actionCode, Object param) {
    if (hook != null) {
        if (param == null) {
            hook.action(actionCode, this);
        } else {
            hook.action(actionCode, param);
        }
    }
}

  13. AbstractProcessor#action() 对于 ActionCode.ASYNC_START 的处理如下，可以看到实际是调用了异步状态机的 AsyncStateMachine#asyncStart() 方法

    此处需要注意，Tomcat 会根据每一个请求的协议为其创建一个 AbstractProcessor 实现类，这个实现类负责这个请求全部生命周期的处理，这也就是AbstractProcessor内部保存 AsyncStateMachine 状态机的原因

    public final void action(ActionCode actionCode, Object param) {
    switch (actionCode) {
    
    ......

    case COMMIT: {
          asyncStateMachine.asyncStart((AsyncContextCallback) param);
        break;
    }

    ......

    }
}

  14. AsyncStateMachine#asyncStart() 方法实际保存了异步开始时间，并进行状态更新，此处可以看到其将步骤11提到的 AsyncContextImpl 保存了下来，后续将用于回调上层

    AsyncStateMachine这个类实际维护了一个异步请求的状态流转，伴随着这个请求的整个生命周期

     synchronized void asyncStart(AsyncContextCallback asyncCtxt) {
    if (state == AsyncState.DISPATCHED) {
        generation.incrementAndGet();
        state = AsyncState.STARTING;
        // Note: In this instance, caller is responsible for calling
        // asyncCtxt.incrementInProgressAsyncCount() as that allows simpler
        // error handling.
        this.asyncCtxt = asyncCtxt;
        lastAsyncStart = System.currentTimeMillis();
    } else {
        throw new IllegalStateException(
                sm.getString("asyncStateMachine.invalidAsyncState",
                        "asyncStart()", state));
    }
}

  15. 此时回到步骤9，创建完 AsyncContextImpl 对象，还会为其注册监听器，设置超时时间，方法源码如下：

    1. 至此我们可以理出一条异步请求完成时的传播链，自底向上的传播方向为 AsyncStateMatch --> AsyncContextImpl --> StandardServletAsyncWebRequest --> DeferredResult
    2. 可以看到超时时间在 AsyncContextImpl 内部保存了一份，也通过 org.apache.coyote.Request 设置到了底层，此处不再赘述方法堆栈，直接看AbstractProcessor#action() 对于 ActionCode.ASYNC_SETTIMEOUT 的处理
    public void addListener(AsyncListener listener) {
    check();
    AsyncListenerWrapper wrapper = new AsyncListenerWrapper();
    wrapper.setListener(listener);
    listeners.add(wrapper);
}

public void setTimeout(long timeout) {
    check();
    this.timeout = timeout;
    request.getCoyoteRequest().action(ActionCode.ASYNC_SETTIMEOUT,
            Long.valueOf(timeout));
}

  16. AbstractProcessor#action() 对于 ActionCode.ASYNC_SETTIMEOUT 的处理如下，可以看到实际只是作了保存操作而已

    至此超时时间已经设置到底层，那么 Tomcat 是如何做到异步请求超时返回呢？答案如下：

    1. Tomcat 不断轮询 socket，当轮询到可读写的 socket 时会创建 SocketProcessorBase 异步任务扔到线程池中执行 socket 读写任务。上文提到过 AbstractProcessor 实现类负责一个请求的全部生命周期处理，在处理 socket 的过程中，连接处理器 ConnectionHandler 如果发现当前请求是异步请求，则会将负责处理这个请求的 Processor 添加到等待队列中
    2. Tomcat 服务器启动时会开启一个周期线程池，处理等待队列中的 Processor，检查其是否超时。超时则触发处理 Socket 超时事件，最终通过 CoyoteAdapter 适配器将超时回调到上层。读者如有兴趣，可参考 
    public final void action(ActionCode actionCode, Object param) {
    switch (actionCode) {
    
    ......

    case ASYNC_SETTIMEOUT: {
        if (param == null) {
            return;
        }
        long timeout = ((Long) param).longValue();
        setAsyncTimeout(timeout);
        break;
    }

    ......

    }
}

public void setAsyncTimeout(long timeout) {
    asyncTimeout = timeout;
}

  17. 此时回到步骤8第5点，DeferredResult#setResultHandler() 的处理如下，可以看到这个步骤主要目的是保存 DeferredResultHandler，此时 DeferredResult 内部还没有有效的结果，所以不会调用 DeferredResultHandler#handleResult() 处理结果，至此请求的异步处理结束

    public final void setResultHandler(DeferredResultHandler resultHandler) {
	Assert.notNull(resultHandler, "DeferredResultHandler is required");
	// Immediate expiration check outside of the result lock
	if (this.expired) {
		return;
	}
	Object resultToHandle;
	synchronized (this) {
		// Got the lock in the meantime: double-check expiration status
		if (this.expired) {
			return;
		}
		resultToHandle = this.result;
		if (resultToHandle == RESULT_NONE) {
			// No result yet: store handler for processing once it comes in
			this.resultHandler = resultHandler;
			return;
		}
	}
	// If we get here, we need to process an existing result object immediately.
	// The decision is made within the result lock; just the handle call outside
	// of it, avoiding any deadlock potential with Servlet container locks.
	try {
		resultHandler.handleResult(resultToHandle);
	}
	catch (Throwable ex) {
		logger.debug("Failed to process async result", ex);
	}
}

2.2 请求的再次分发

  1. 我们知道当 DeferredResult#setResult() 方法被调用时将会对客户端请求进行响应，请求的再次分发也是由此触发

    DeferredResult#setResult() 方法只是个入口，可以看到当设置了结果之后，将调用 DeferredResultHandler#handleResult() 处理结果

    public boolean setResult(T result) {
 	return setResultInternal(result);
 }

 private boolean setResultInternal(Object result) {
 	// Immediate expiration check outside of the result lock
 	if (isSetOrExpired()) {
 		return false;
 	}
 	DeferredResultHandler resultHandlerToUse;
 	synchronized (this) {
 		// Got the lock in the meantime: double-check expiration status
 		if (isSetOrExpired()) {
 			return false;
 		}
 		// At this point, we got a new result to process
 		this.result = result;
 		resultHandlerToUse = this.resultHandler;
 		if (resultHandlerToUse == null) {
 			// No result handler set yet -> let the setResultHandler implementation
 			// pick up the result object and invoke the result handler for it.
 			return true;
 		}
 		// Result handler available -> let's clear the stored reference since
 		// we don't need it anymore.
 		this.resultHandler = null;
 	}
 	// If we get here, we need to process an existing result object immediately.
 	// The decision is made within the result lock; just the handle call outside
 	// of it, avoiding any deadlock potential with Servlet container locks.
 	resultHandlerToUse.handleResult(result);
 	return true;
 }

  2. DeferredResultHandler#handleResult() 的实现其实在2.1节步骤8 提到过，实际在 WebAsyncManager#startDeferredResultProcessing() 中，如下所示，可以看到核心逻辑是调用 WebAsyncManager#setConcurrentResultAndDispatch() 方法

    public void startDeferredResultProcessing(
 		final DeferredResult<?> deferredResult, Object... processingContext) throws Exception {

 	......

 	try {
 		interceptorChain.applyPreProcess(this.asyncWebRequest, deferredResult);
 		deferredResult.setResultHandler(result -> {
 			result = interceptorChain.applyPostProcess(this.asyncWebRequest, deferredResult, result);
 			setConcurrentResultAndDispatch(result);
 		});
 	}
 	catch (Throwable ex) {
 		setConcurrentResultAndDispatch(ex);
 	}
 }

  3. WebAsyncManager#setConcurrentResultAndDispatch() 方法的核心处理如下：

    1. 将 DeferredResult#setResult() 设置的结果保存在当前 WebAsyncManager 对象中
    2. this.asyncWebRequest.dispatch() 将异步请求再次分发，请求分发的流程与本文相关较小，暂不做分析
    private void setConcurrentResultAndDispatch(Object result) {
 	synchronized (WebAsyncManager.this) {
 		if (this.concurrentResult != RESULT_NONE) {
 			return;
 		}
 		this.concurrentResult = result;
 		this.errorHandlingInProgress = (result instanceof Throwable);
 	}

 	if (this.asyncWebRequest.isAsyncComplete()) {
 		if (logger.isDebugEnabled()) {
 			logger.debug("Async result set but request already complete: " + formatRequestUri());
 		}
 		return;
 	}

 	if (logger.isDebugEnabled()) {
 		boolean isError = result instanceof Throwable;
 		logger.debug("Async " + (isError ? "error" : "result set") + ", dispatch to " + formatRequestUri());
 	}
 	this.asyncWebRequest.dispatch();
 }

  4. 再次分发的请求将重新进入 DispatcherServlet#doDispatch() 处理，其流程与2.1节步骤1-4基本相同，只不过这次请求处理将进入2.1节步骤4第4点逻辑，此时 WebAsyncManager中已经有异步请求的结果了， invocableMethod.wrapConcurrentResult() 将会创建一个新的ConcurrentResultHandlerMethod 对象

    protected ModelAndView invokeHandlerMethod(HttpServletRequest request,
 		HttpServletResponse response, HandlerMethod handlerMethod) throws Exception {

 	......

 		if (asyncManager.hasConcurrentResult()) {
 			Object result = asyncManager.getConcurrentResult();
 			mavContainer = (ModelAndViewContainer) asyncManager.getConcurrentResultContext()[0];
 			asyncManager.clearConcurrentResult();
 			LogFormatUtils.traceDebug(logger, traceOn -> {
 				String formatted = LogFormatUtils.formatValue(result, !traceOn);
 				return "Resume with async result [" + formatted + "]";
 			});
 			invocableMethod = invocableMethod.wrapConcurrentResult(result);
 		}

 		invocableMethod.invokeAndHandle(webRequest, mavContainer);
 		if (asyncManager.isConcurrentHandlingStarted()) {
 			return null;
 		}

 		return getModelAndView(mavContainer, modelFactory, webRequest);
 	}
 	finally {
 		webRequest.requestCompleted();
 	}
 }

  5. ConcurrentResultHandlerMethod 的构造方法如下，可以看到这个对象把异步请求的结果包装到一个 Callable 对象中，并设置了其反射调用的 Method 为 call，这一步其实就是把原本应该反射调用的处理器方法丢弃了，直接返回异步请求已经设置好的返回结果

    ServletInvocableHandlerMethod wrapConcurrentResult(Object result) {
 	return new ConcurrentResultHandlerMethod(result, new ConcurrentResultMethodParameter(result));
 }
 public ConcurrentResultHandlerMethod(final Object result, ConcurrentResultMethodParameter returnType) {
 		super((Callable<Object>) () -> {
 			if (result instanceof Exception) {
 				throw (Exception) result;
 			}
 			else if (result instanceof Throwable) {
 				throw new NestedServletException("Async processing failed", (Throwable) result);
 			}
 			return result;
 		}, CALLABLE_METHOD);

 		if (ServletInvocableHandlerMethod.this.returnValueHandlers != null) {
 			setHandlerMethodReturnValueHandlers(ServletInvocableHandlerMethod.this.returnValueHandlers);
 		}
 		this.returnType = returnType;
 	}

  6. 继续本节步骤4流程，此时 invocableMethod.invokeAndHandle() 依然将调用到 ConcurrentResultHandlerMethod 父类ServletInvocableHandlerMethod#invokeAndHandle() 方法，此处的处理是异步请求响应的最后一步

    需注意 ConcurrentResultHandlerMethod 内部保存的反射所需的 Method 对象已经变成了 Callable
    的 call 方法，则此处invokeForRequest()反射调用实际调用到 Callable#call() 方法，返回值已经是 DeferredResult 实际应该返回的数据了

    public void invokeAndHandle(ServletWebRequest webRequest, ModelAndViewContainer mavContainer,
 		Object... providedArgs) throws Exception {

 	Object returnValue = invokeForRequest(webRequest, mavContainer, providedArgs);
 	setResponseStatus(webRequest);

 	if (returnValue == null) {
 		if (isRequestNotModified(webRequest) || getResponseStatus() != null || mavContainer.isRequestHandled()) {
 			disableContentCachingIfNecessary(webRequest);
 			mavContainer.setRequestHandled(true);
 			return;
 		}
 	}
 	else if (StringUtils.hasText(getResponseStatusReason())) {
 		mavContainer.setRequestHandled(true);
 		return;
 	}

 	mavContainer.setRequestHandled(false);
 	Assert.state(this.returnValueHandlers != null, "No return value handlers");
 	try {
 		this.returnValueHandlers.handleReturnValue(
 				returnValue, getReturnValueType(returnValue), mavContainer, webRequest);
 	}
 	catch (Exception ex) {
 		if (logger.isTraceEnabled()) {
 			logger.trace(formatErrorForReturnValue(returnValue), ex);
 		}
 		throw ex;
 	}
 }

  7. 对于普通类型的返回值，this.returnValueHandlers.handleReturnValue() 处理返回值时会选择对应的返回值处理器，而不会再次由 DeferredResultMethodReturnValueHandler 处理，至此，异步请求的结果将顺利响应给客户端，异步请求结束

Copyright © 2019- zgxue.com 版权所有
京ICP备2021021884号-5

Top