DeferredResult解决了什么问题

叔牙

发布于 2022-09-27 08:05:00

7790

发布于 2022-09-27 08:05:00

文章被收录于专栏：一个执拗的后端搬砖工一个执拗的后端搬砖工

内容目录

一、概念与背景

二、应用场景

三、使用方式

四、原理与根源分析

五、解决问题

一、概念与背景

Servlet3.0提供了异步处理请求的特性,DeferredResult 是spring基于 Servlet 3.0 对异步请求的支持实现,目的是对于请求提供异步处理方式,释放容器连接,支持更多的并发。或者基于它的超时机制来做一些长轮询相关的事情。

二、应用场景

之前的一篇文章《Apollo客户端通知原理》中有介绍到,服务端的配置变更是通过长轮询通知到客户端,其实现就是DeferredResult,那么它的使用场景就有如下:

1.服务端消息推送

apollo配置变更,客户端通过长轮询请求服务端,服务端通过DeferredResult实现变更通知
消息推送,对于一些服务端发生变更,需要向客户端发送消息通知的场景,不管是C/S还是B/S模式,也可以通过DeferredResult来实现

2.增加系统吞吐量

拿tomcat作为servlet容器来说,无论是计算型请求还是io型请求,都是交给tomcat容器线程来建立连接和负责业务逻辑处理,如果将io型请求或者rt比较高的请求业务逻辑处理,通过DeferredResult来实现,那么可以尽早地释放连接线程,业务逻辑交由业务线程池处理,那么连接线程池可以接收更多的请求,从而提高了系统吞吐量。

三、使用方式

1.编写DeferredResult返回类型api

@GetMapping("/deferredresult/test")
public DeferredResult<String> testDeferredResult(long sleepTime) {
    DeferredResult<String> deferredResult = new DeferredResult<>(5000L,"server side timeout");
    executorService.submit(() -> {
        try {
            Thread.sleep(sleepTime);
            deferredResult.setResult("server response successfully");
        } catch (InterruptedException e) {
            log.error("occur error",e);
        }
    });
    return deferredResult;
}

2.接口调用

这样就完成了DeferredResult异步调用,当然我们也可以在DeferredResult设置超时相关逻辑。

四、原理与源码分析

为了方便理解,找了一张图来看一下DeferredResult做了什么事情。

接收到请求后,将请求暂存并且释放容器线程,用来接收新的请求
容器超时逻辑和业务正常处理逻辑将结果塞到DeferredResult返回调

spring对于DeferredResult请求处理

1.请求预处理

当然DeferredResult处理逻辑也脱离不了spring mvc的支持,也是要走到DispatcherServlet来处理请求:

  protected void doDispatch(HttpServletRequest request, HttpServletResponse response) throws Exception {
      HttpServletRequest processedRequest = request;
      HandlerExecutionChain mappedHandler = null;
      boolean multipartRequestParsed = false;
      //1.生成异步管理器
      WebAsyncManager asyncManager = WebAsyncUtils.getAsyncManager(request);
      try {
          try {
              ModelAndView mv = null;
              Object dispatchException = null;
              try {
                  //省略...
                  //2.异步处理逻辑
                  mv = ha.handle(processedRequest, response, mappedHandler.getHandler());
                  //3.如果异步处理一开始,返回调用
                  if (asyncManager.isConcurrentHandlingStarted()) {
                      return;
                  }
                  this.applyDefaultViewName(processedRequest, mv);
                  mappedHandler.applyPostHandle(processedRequest, response, mv);
              } catch (Exception var20) {
                  dispatchException = var20;
              } catch (Throwable var21) {
                  dispatchException = new NestedServletException("Handler dispatch failed", var21);
              }

              this.processDispatchResult(processedRequest, response, mappedHandler, mv, (Exception)dispatchException);
          } catch (Exception var22) {

          } catch (Throwable var23) {

          }
      } finally {
          //省略...
      }
  }

对于支持DeferredResult异步处理逻辑有三个关键点:

生成异步管理器
执行异步处理逻辑
如果异步处理已经开始,返回调用

HandleAdapter#handle会调用到DeferredResultMethodReturnValueHandler的handleReturnValue方法:

public void handleReturnValue(@Nullable Object returnValue, MethodParameter returnType, ModelAndViewContainer mavContainer, NativeWebRequest webRequest) throws Exception {
    if (returnValue == null) {
        mavContainer.setRequestHandled(true);
    } else {
        DeferredResult result;
        if (returnValue instanceof DeferredResult) {
            result = (DeferredResult)returnValue;
        } else if (returnValue instanceof ListenableFuture) {
            result = this.adaptListenableFuture((ListenableFuture)returnValue);
        } else {
            if (!(returnValue instanceof CompletionStage)) {
                throw new IllegalStateException("Unexpected return value type: " + returnValue);
            }
            result = this.adaptCompletionStage((CompletionStage)returnValue);
        }
        WebAsyncUtils.getAsyncManager(webRequest).startDeferredResultProcessing(result, new Object[]{mavContainer});
    }
}

进入WebAsyncManager的startDeferredResultProcessing方法:

public void startDeferredResultProcessing(final DeferredResult<?> deferredResult, Object... processingContext) throws Exception {
    //省略...
    this.startAsyncProcessing(processingContext);
    try {
        interceptorChain.applyPreProcess(this.asyncWebRequest, deferredResult);
        deferredResult.setResultHandler((result) -> {
            result = interceptorChain.applyPostProcess(this.asyncWebRequest, deferredResult, result);
            this.setConcurrentResultAndDispatch(result);
        });
    } catch (Throwable var7) {
        this.setConcurrentResultAndDispatch(var7);
    }
}

startAsyncProcessing方法开启异步处理(asyncManager.isConcurrentHandlingStarted会用到),然后进入DeferredResult的setResultHandler设置结果处理器:

public final void setResultHandler(DeferredResult.DeferredResultHandler resultHandler) {
    Assert.notNull(resultHandler, "DeferredResultHandler is required");
    if (!this.expired) {
        Object resultToHandle;
        synchronized(this) {
            if (this.expired) {
                return;
            }
            resultToHandle = this.result;
            if (resultToHandle == RESULT_NONE) {
                this.resultHandler = resultHandler;
                return;
            }
        }
        try {
            resultHandler.handleResult(resultToHandle);
        } catch (Throwable var5) {
            logger.debug("Failed to process async result", var5);
        }
    }
}

由于逻辑未处理,结果未设置,所以逻辑会走到设置结果处理器代码块,然后返回,此时返回值解析过程结束了，同时由于异步servlet的特性，tomcat的连接也得到了释放。

预处理流程如下:

2.返回值处理

这个时候容器连接得到了释放,然而问题并没有解决,请求处理只完成了一半,业务处理返回值并没有真正返回。

我们在业务线程池处理调用了DeferredResult的setResult方法,最终会调用内部setResultInternal：

private boolean setResultInternal(Object result) {
    if (this.isSetOrExpired()) {
        return false;
    } else {
        DeferredResult.DeferredResultHandler resultHandlerToUse;
        synchronized(this) {
            if (this.isSetOrExpired()) {
                return false;
            }
            this.result = result;
            resultHandlerToUse = this.resultHandler;
            if (resultHandlerToUse == null) {
                return true;
            }
            this.resultHandler = null;
        }
        resultHandlerToUse.handleResult(result);
        return true;
    }
}

这里会调用之前传入的函数式接口来处理:

deferredResult.setResultHandler((result) -> {
    result = interceptorChain.applyPostProcess(this.asyncWebRequest, deferredResult, result);
    this.setConcurrentResultAndDispatch(result);
});

然后调用setConcurrentResultAndDispatch:

private void setConcurrentResultAndDispatch(Object result) {
    synchronized(this) {
        if (this.concurrentResult != RESULT_NONE) {
            return;
        }
        this.concurrentResult = result;
        this.errorHandlingInProgress = result instanceof Throwable;
    }
    if (this.asyncWebRequest.isAsyncComplete()) {
      //...
    } else {
        this.asyncWebRequest.dispatch();
    }
}

如果异步处理完成则返回调用,否则执行异步请求分发,该段代码执行完成会发起一次新的请求到后台，又被DispatcherServlet类接收到（但是不会再进入controller了），最终将结果响应给调用方。

tomcat容器维度对异步支持

我们再从容器维度对DeferredResult异步请求的处理做一下分析,分别是请求超时和主动setResult返回。

1.请求超时

Connector是tomcat的最核心的组件之一,主要的职责就是负责接收客户端连接和客户端请求的处理加工,初始化和启动会执行Protocal相关初始化和启动操作,看一下AbstractProtocol的启动:

public void start() throws Exception {
    if (this.getLog().isInfoEnabled()) {
        this.getLog().info(sm.getString("abstractProtocolHandler.start", new Object[]{this.getName()}));
        this.logPortOffset();
    }
    this.endpoint.start();
    this.monitorFuture = this.getUtilityExecutor().scheduleWithFixedDelay(() -> {
        this.startAsyncTimeout();
    }, 0L, 60L, TimeUnit.SECONDS);
}

延时60秒执行启动异步超时支持逻辑,调用startAsyncTimeout:

protected void startAsyncTimeout() {
    if (this.timeoutFuture == null || this.timeoutFuture.isDone()) {
      //省略...
        this.timeoutFuture = this.getUtilityExecutor().scheduleAtFixedRate(() -> {
            long now = System.currentTimeMillis();
            Iterator var3 = this.waitingProcessors.iterator();
            while(var3.hasNext()) {
                Processor processor = (Processor)var3.next();
                processor.timeoutAsync(now);
            }
        }, 1L, 1L, TimeUnit.SECONDS);
    }
}

异步请求会被放入waitingProcessors中,并且设置了超时时间，tomcat会有一个线程每隔1秒遍历waitingProcessors 里面的 processor,检查是否过期,如果过期会往tomcat线程池投掷超时事件:

private void doTimeoutAsync() {
    this.setAsyncTimeout(-1L);
    this.asyncTimeoutGeneration = this.asyncStateMachine.getCurrentGeneration();
    this.processSocketEvent(SocketEvent.TIMEOUT, true);
}

线程池跑到这个任务的时候就知道这个已经超时请求任务,此时就会将超时值塞入到请求中,具体是通过之前设置的DeferredResult相关的拦截器中的handleTimeout,比如spring自己提供的拦截器:

public class TimeoutDeferredResultProcessingInterceptor implements DeferredResultProcessingInterceptor {
    public TimeoutDeferredResultProcessingInterceptor() {
    }
    public <T> boolean handleTimeout(NativeWebRequest request, DeferredResult<T> result) throws Exception {
        result.setErrorResult(new AsyncRequestTimeoutException());
        return false;
    }
}

最终会把值放到管理异步请求AsyncManager中并重新下发请求交给DispatcherServlet#doDispatch处理,第二次进来的时候发现AsyncManager已经有值了,把结果进行包装然后直接返回调用了。

超时逻辑处理流程如下:

2.setResult主动返回

业务线程在执行完逻辑,将结果塞回到DeferredResult时也会调用setResultInternal,赋值完成后调用AsyncWebRequest#dispatch方法重新下发请求,DispatcherServlet处理时发现AsyncManager已经有值了,封装后直接返回,后边逻辑和超时逻辑一样。

处理流程如下:

五、解决了什么问题

我们按照图中介绍的场景,如果服务器能够开100个线程,如果不使用DeferredResult异步处理,那么所有的连接和业务处理都有tomcat线程来处理,那么对于一些io型操作可能会长时间占用连接线程,会导致系统吞吐量下降,如果给tomcat 70个线程作为连接处理,30个作为业务处理,那么对于一些rt比较高的操作使用DeferredResult异步操作,对于一些计算型操作直接使用tomcat连接线程处理业务,系统吞吐量能提升吗？我看未必,服务器资源是固定的,线程数量不可能凭空无限制增加和减少,所以tomcat和服务共享资源的前提下,开启DeferredResult异步服务未必有多少性能提升,这个可以自己研究。

另外一点可能是DeferredResult比较亮眼的一个地方,就是可以实现在不引入额外组件和额外开发工作量的情况下解决服务端到客户端的主动消息触达。

本文参与腾讯云自媒体分享计划，分享自微信公众号。

原始发表：2022-08-25，如有侵权请联系 cloudcommunity@tencent.com 删除

tomcat