25.springmvc源码学习（二十五）异步请求管理器WebAsyncManager异步任务初始化

目录

• 前言
• 一、示例
• 二、原理
• 总结

---

前言

以接口的返回值为DeferredResult为例来分析一下异步任务的初始化。

---

一、示例

    @ApiOperation(value = "test", notes = "test")
    @GetMapping(value = "/test", produces = {MediaType.APPLICATION_JSON_VALUE})
    public DeferredResult test() throws ValidDataException {
        Test test = new Test();
        test.setId(1);
        test.setName("test1");

        DeferredResult<Object> result = new DeferredResult<>(2000L);
        result.setResult(test);
        result.onTimeout(() -> {
            result.setResult("请求超时");
        });

        return result;
  	}

二、原理

（1）当Controller方法执行后，会通过returnValueHandlers对返回值进行处理

public void handleReturnValue(@Nullable Object returnValue, MethodParameter returnType,
			ModelAndViewContainer mavContainer, NativeWebRequest webRequest) throws Exception {
		//获取ReturnValueHandler
		HandlerMethodReturnValueHandler handler = selectHandler(returnValue, returnType);
		if (handler == null) {
			throw new IllegalArgumentException("Unknown return value type: " + returnType.getParameterType().getName());
		}
		//通过ReturnValueHandler处理返回值
		handler.handleReturnValue(returnValue, returnType, mavContainer, webRequest);
	}

（2）这里返回值是DeferredResult 类型，则会进入DeferredResultMethodReturnValueHandler的handleReturnValue方法

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

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

		DeferredResult<?> result;
		//DeferredResult类型
		if (returnValue instanceof DeferredResult) {
			result = (DeferredResult<?>) returnValue;
		}
		//ListenableFuture类型
		else if (returnValue instanceof ListenableFuture) {
			result = adaptListenableFuture((ListenableFuture<?>) returnValue);
		}
		//CompletionStage类型
		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);
	}

（3）异步执行任务startDeferredResultProcessing

这里开始异步任务的处理流程

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);
		
		//超时TimeoutHandler
		this.asyncWebRequest.addTimeoutHandler(() -> {
			try {
				interceptorChain.triggerAfterTimeout(this.asyncWebRequest, deferredResult);
			}
			catch (Throwable ex) {
				setConcurrentResultAndDispatch(ex);
			}
		});
		//错误ErrorHandler
		this.asyncWebRequest.addErrorHandler(ex -> {
			if (!this.errorHandlingInProgress) {
				try {
					if (!interceptorChain.triggerAfterError(this.asyncWebRequest, deferredResult, ex)) {
						return;
					}
					deferredResult.setErrorResult(ex);
				}
				catch (Throwable interceptorEx) {
					setConcurrentResultAndDispatch(interceptorEx);
				}
			}
		});
		//任务完成CompletionHandler
		this.asyncWebRequest.addCompletionHandler(()
				-> interceptorChain.triggerAfterCompletion(this.asyncWebRequest, deferredResult));
		//调用拦截器的beforeConcurrentHandling
		interceptorChain.applyBeforeConcurrentHandling(this.asyncWebRequest, deferredResult);
		//执行异步任务
		startAsyncProcessing(processingContext);

		try {
			//调用拦截器的applyPreProcess
			interceptorChain.applyPreProcess(this.asyncWebRequest, deferredResult);
			deferredResult.setResultHandler(result -> {
				//调用拦截器的applyPostProcess
				result = interceptorChain.applyPostProcess(this.asyncWebRequest, deferredResult, result);
				//设置结果，并且分发获取结果的请求
				setConcurrentResultAndDispatch(result);
			});
		}
		catch (Throwable ex) {
			setConcurrentResultAndDispatch(ex);
		}
	}

（4）异步执行任务startAsyncProcessing( )

1、startAsyncProcessing( )

private void startAsyncProcessing(Object[] processingContext) {
		synchronized (WebAsyncManager.this) {
			//设置结果为RESULT_NONE
			this.concurrentResult = RESULT_NONE;
			this.concurrentResultContext = processingContext;
			this.errorHandlingInProgress = false;
		}
		//执行异步任务
		this.asyncWebRequest.startAsync();

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

2、startAsync( )

StandardServletAsyncWebRequest.java

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);
		}
	}

3、startAsync( )

ServletRequestWrapper.java

 public AsyncContext startAsync(ServletRequest servletRequest,
                                   ServletResponse servletResponse)
            throws IllegalStateException {
        return request.startAsync(servletRequest, servletResponse);
    }

4、startAsync( )

HttpServletRequestImpl.java

 @Override
    public AsyncContext startAsync(final ServletRequest servletRequest, final ServletResponse servletResponse) throws IllegalStateException {
        final ServletRequestContext servletRequestContext = exchange.getAttachment(ServletRequestContext.ATTACHMENT_KEY);
        if (!servletContext.getDeployment().getDeploymentInfo().isAllowNonStandardWrappers()) {
            if (servletRequestContext.getOriginalRequest() != servletRequest) {
                if (!(servletRequest instanceof ServletRequestWrapper)) {
                    throw UndertowServletMessages.MESSAGES.requestWasNotOriginalOrWrapper(servletRequest);
                }
            }
            if (servletRequestContext.getOriginalResponse() != servletResponse) {
                if (!(servletResponse instanceof ServletResponseWrapper)) {
                    throw UndertowServletMessages.MESSAGES.responseWasNotOriginalOrWrapper(servletResponse);
                }
            }
        }
        //是否支持异步，默认支持 private boolean asyncSupported = true;
        if (!isAsyncSupported()) {
            throw UndertowServletMessages.MESSAGES.startAsyncNotAllowed();
        } else if (asyncStarted) {
        	//是否已经执行过
            throw UndertowServletMessages.MESSAGES.asyncAlreadyStarted();
        }
        //标记异步任务开始执行
        asyncStarted = true;
        servletRequestContext.setServletRequest(servletRequest);
        servletRequestContext.setServletResponse(servletResponse);
        //创建AsyncContextImpl
        return asyncContext = new AsyncContextImpl(exchange, servletRequest, servletResponse, servletRequestContext, true, asyncContext);
    }

5、AsyncContextImpl

public AsyncContextImpl(final HttpServerExchange exchange, final ServletRequest servletRequest, final ServletResponse servletResponse, final ServletRequestContext servletRequestContext, boolean requestSupplied, final AsyncContextImpl previousAsyncContext) {
        this.exchange = exchange;
        this.servletRequest = servletRequest;
        this.servletResponse = servletResponse;
        this.servletRequestContext = servletRequestContext;
        this.requestSupplied = requestSupplied;
        this.previousAsyncContext = previousAsyncContext;
        initiatingThread = Thread.currentThread();
        //调用dispatch(),参数为SameThreadExecutor和Runnable
        exchange.dispatch(SameThreadExecutor.INSTANCE, new Runnable() {
            @Override
            public void run() {
                exchange.setDispatchExecutor(null);
                initialRequestDone();
            }
        });
    }

6、dispatch( )

 public HttpServerExchange dispatch(final Executor executor, final Runnable runnable) {
        if (isInCall()) {
            if (executor != null) {
                this.dispatchExecutor = executor;
            }
            state |= FLAG_DISPATCHED;
            if(anyAreSet(state, FLAG_SHOULD_RESUME_READS | FLAG_SHOULD_RESUME_WRITES)) {
                throw UndertowMessages.MESSAGES.resumedAndDispatched();
            }
            //设置dispatchTask，为传入的runnable，这个任务会被调用
            this.dispatchTask = runnable;
        } else {
            if (executor == null) {
                getConnection().getWorker().execute(runnable);
            } else {
                executor.execute(runnable);
            }
        }
        return this;
    }

到这里初始化好了一个将被分发的任务dispatchTask

（5）设置结果处理器

public final void setResultHandler(DeferredResultHandler resultHandler) {
		Assert.notNull(resultHandler, "DeferredResultHandler is required");
		// Immediate expiration check outside of the result lock
		//校验expired，请求是否过期
		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;
			//检查DeferredResult的结果
			if (resultToHandle == RESULT_NONE) {
				// No result yet: store handler for processing once it comes in
				//结果为初始化的RESULT_NONE，即还没有获得结果
				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);
		}
	}

（6）处理结果resultHandler.handleResult(resultToHandle);

1、进入lambda表达式

result -> {
				//调用拦截器的applyPostProcess
				result = interceptorChain.applyPostProcess(this.asyncWebRequest, deferredResult, result);
				//设置结果，并且分发获取结果的请求
				setConcurrentResultAndDispatch(result);
			}

2、setConcurrentResultAndDispatch( )

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();
	}

3、dispatch( )

StandardServletAsyncWebRequest.java

    @Override
	public void dispatch() {
		Assert.notNull(this.asyncContext, "Cannot dispatch without an AsyncContext");
		this.asyncContext.dispatch();
	}

4、dispatch( )

AsyncContextImpl.java

 public void dispatch() {
        if (this.dispatched) {
            throw UndertowServletMessages.MESSAGES.asyncRequestAlreadyDispatched();
        } else {
            HttpServletRequestImpl requestImpl = this.servletRequestContext.getOriginalRequest();
            Deployment deployment = requestImpl.getServletContext().getDeployment();
            ServletContainer container;
            String qs;
            if (this.requestSupplied && this.servletRequest instanceof HttpServletRequest) {
                container = deployment.getServletContainer();
                String requestURI = ((HttpServletRequest)this.servletRequest).getRequestURI();
                DeploymentManager context = container.getDeploymentByPath(requestURI);
                if (context == null) {
                    throw UndertowServletMessages.MESSAGES.couldNotFindContextToDispatchTo(requestImpl.getOriginalContextPath());
                }

                qs = requestURI.substring(context.getDeployment().getServletContext().getContextPath().length());
                String qs = ((HttpServletRequest)this.servletRequest).getQueryString();
                if (qs != null && !qs.isEmpty()) {
                    qs = qs + "?" + qs;
                }
				//分发请求
                this.dispatch(context.getDeployment().getServletContext(), qs);
            } else {
                container = deployment.getServletContainer();
                DeploymentManager context = container.getDeploymentByPath(requestImpl.getOriginalContextPath());
                if (context == null) {
                    throw UndertowServletMessages.MESSAGES.couldNotFindContextToDispatchTo(requestImpl.getOriginalContextPath());
                }

                String toDispatch = requestImpl.getExchange().getRelativePath();
                qs = requestImpl.getOriginalQueryString();
                if (qs != null && !qs.isEmpty()) {
                    toDispatch = toDispatch + "?" + qs;
                }

                this.dispatch(context.getDeployment().getServletContext(), toDispatch);
            }

        }
    }

5、dispatch( )

 public void dispatch(ServletContext context, String path) {
        if (this.dispatched) {
            throw UndertowServletMessages.MESSAGES.asyncRequestAlreadyDispatched();
        } else {
        	//设置参数
            HttpServletRequestImpl requestImpl = this.servletRequestContext.getOriginalRequest();
            HttpServletResponseImpl responseImpl = this.servletRequestContext.getOriginalResponse();
            HttpServerExchange exchange = requestImpl.getExchange();
            ((ServletRequestContext)exchange.getAttachment(ServletRequestContext.ATTACHMENT_KEY)).setDispatcherType(DispatcherType.ASYNC);
            requestImpl.setAttribute("javax.servlet.async.request_uri", requestImpl.getOriginalRequestURI());
            requestImpl.setAttribute("javax.servlet.async.context_path", requestImpl.getOriginalContextPath());
            requestImpl.setAttribute("javax.servlet.async.servlet_path", requestImpl.getOriginalServletPath());
            requestImpl.setAttribute("javax.servlet.async.query_string", requestImpl.getOriginalQueryString());
            String newQueryString = "";
            int qsPos = path.indexOf("?");
            String newServletPath = path;
            if (qsPos != -1) {
                newQueryString = path.substring(qsPos + 1);
                newServletPath = path.substring(0, qsPos);
            }

            String newRequestUri = context.getContextPath() + newServletPath;
            Map<String, Deque<String>> newQueryParameters = new HashMap();
            String[] var11 = newQueryString.split("&");
            int var12 = var11.length;

            for(int var13 = 0; var13 < var12; ++var13) {
                String part = var11[var13];
                String name = part;
                String value = "";
                int equals = part.indexOf(61);
                if (equals != -1) {
                    name = part.substring(0, equals);
                    value = part.substring(equals + 1);
                }

                Deque<String> queue = (Deque)newQueryParameters.get(name);
                if (queue == null) {
                    newQueryParameters.put(name, queue = new ArrayDeque(1));
                }

                ((Deque)queue).add(value);
            }

            requestImpl.setQueryParameters(newQueryParameters);
            requestImpl.getExchange().setRelativePath(newServletPath);
            requestImpl.getExchange().setQueryString(newQueryString);
            requestImpl.getExchange().setRequestPath(newRequestUri);
            requestImpl.getExchange().setRequestURI(newRequestUri);
            requestImpl.setServletContext((ServletContextImpl)context);
            responseImpl.setServletContext((ServletContextImpl)context);
            Deployment deployment = requestImpl.getServletContext().getDeployment();
            ServletPathMatch info = deployment.getServletPaths().getServletHandlerByPath(newServletPath);
            ((ServletRequestContext)requestImpl.getExchange().getAttachment(ServletRequestContext.ATTACHMENT_KEY)).setServletPathMatch(info);
            //分发异步请求
            this.dispatchAsyncRequest(deployment.getServletDispatcher(), info, exchange);
        }
    }

6、dispatchAsyncRequest( )

private void dispatchAsyncRequest(final ServletDispatcher servletDispatcher, final ServletPathMatch pathInfo, final HttpServerExchange exchange) {
		//参数是Runnable
        this.doDispatch(new Runnable() {
            public void run() {
            	//这里实现了即将调用的新一轮的异步请求
                Connectors.executeRootHandler(new HttpHandler() {
                    public void handleRequest(HttpServerExchange exchangex) throws Exception {
                        ServletRequestContext src = (ServletRequestContext)exchangex.getAttachment(ServletRequestContext.ATTACHMENT_KEY);
                        src.setServletRequest(AsyncContextImpl.this.servletRequest);
                        src.setServletResponse(AsyncContextImpl.this.servletResponse);
                        servletDispatcher.dispatchToPath(exchangex, pathInfo, DispatcherType.ASYNC);
                    }
                }, exchange);
            }
        });
    }

7、doDispatch( )

private synchronized void doDispatch(final Runnable runnable) {
        if (this.dispatched) {
            throw UndertowServletMessages.MESSAGES.asyncRequestAlreadyDispatched();
        } else {
        	//设置请求分发为true
            this.dispatched = true;
            final HttpServletRequestImpl request = this.servletRequestContext.getOriginalRequest();
            //创建异步请求任务
            this.addAsyncTask(new Runnable() {
                public void run() {
                    request.asyncRequestDispatched();
                    runnable.run();
                }
            });
            if (this.timeoutKey != null) {
                this.timeoutKey.remove();
            }

        }
    }

8、addAsyncTask( )

 public synchronized void addAsyncTask(Runnable runnable) {
 		//将异步任务加入队列
        this.asyncTaskQueue.add(runnable);
        if (!this.processingAsyncTask) {
        	//执行异步任务
            this.processAsyncTask();
        }

    }

9、processAsyncTask( )

校验异步任务是否执行过，没有执行过，则直接返回，第一层异步任务开始走到这里时并没有执行过，那么会直接返回。后面再继续分析异步任务在哪里执行...

private synchronized void processAsyncTask() {
		//异步任务是否执行过
        if (this.initialRequestDone) {
            this.updateTimeout();
            Runnable task = (Runnable)this.asyncTaskQueue.poll();
            if (task != null) {
                this.processingAsyncTask = true;
                this.asyncExecutor().execute(new AsyncContextImpl.TaskDispatchRunnable(task));
            } else {
                this.processingAsyncTask = false;
            }

        }
    }

---

总结

创建异步任务后，servlet会释放资源，后续再继续处理异步任务。

来源：https://blog.csdn.net/RenshenLi/article/details/120693626