spring 实现异步非阻塞长轮询

spring 实现异步非阻塞长轮询 一. 前言

今天接到一个扫码登录的需求。想一想很简单，服务端提供一个获取二维码接口，在提供一个查询扫码状态的接口，客户端不停轮询"查询扫码状态接口"判断用户是否已扫码登录，并很快实现。本想开发完成后又可以愉快的摸鱼了，但仔细想想又觉得差点意思。客户端如何频繁的去轮询服务端接口势必会大量浪费tomcat的线程，造成服务端的压力。其实大部分的轮询请求都是无意义的，那是否可以考虑服务端将轮询请求挂起，释放tomcat线程，当有结果时在将响应返回给客户端？于是通过查阅文档发现Spring已经提供了实现方法Callable 、DeferredResult。

二. 简单示例 1. 示例

@RestController
@RequestMapping("/test")
public class TestController {

    private Map> deferredResultMap = Maps.newHashMap();

    @RequestMapping("/callable")
    public Callable callable() {
        Callable callable = new Callable() {
            @Override
            public String call() throws Exception {
                System.out.println("execute async task start, " + System.currentTimeMillis() / 1000L);
                try {
                    // 模拟3秒的任务
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("execute async task end, " + System.currentTimeMillis() / 1000L);
                return "success";
            }
        };
        System.out.println("callable return, " + System.currentTimeMillis() / 1000L);
        return callable;
    }

    @RequestMapping("/deferredResult")
    public DeferredResult deferredResult(@RequestParam String key) {
        DeferredResult deferredResult = new DeferredResult<>();
        deferredResult.onCompletion(() -> {
            System.out.println("task success, " + System.currentTimeMillis() / 1000L);
        });
        deferredResultMap.put(key, deferredResult);
        System.out.println("deferredResult return, " + System.currentTimeMillis() / 1000L);
        return deferredResult;
    }

    @RequestMapping("/deferredResult/change")
    public String change(@RequestParam String key) {
        DeferredResult deferredResult = deferredResultMap.get(key);
        if (null != deferredResult) {
            deferredResult.setResult("success, " + System.currentTimeMillis() / 1000L);
            // DeferredResult 如果全局Map保存时，结束或异常时记得remove掉
            deferredResultMap.remove(key);
        }
        return "success";
    }
}

2. 结果分析

Callable : 接口响应时间3秒钟，符合预期。“callable return"比"task end"早打印3秒钟，说明异步任务是在return之后执行完成。


DeferredResult : “/deferredResult” 接口执行后大概3秒执行”/deferredResult/change"接口，"/deferredResult"接口响应时间大约3秒左右，且异步任务是在return之后执行完成。

三. Callable & DeferredResult 原理 1. 概述

Callable & DeferredResult都是加在Controller方法的返回值上，可以猜测Spring MVC是在处理return时做了特殊处理。带着这个猜测从Spring MVC的入口DispatcherServlet开始寻找答案。

2. 原理分析

DispatcherServlet

protected void doDispatch(HttpServletRequest request, HttpServletResponse response) throws Exception {

    // 省略。。。    
    
   WebAsyncManager asyncManager = WebAsyncUtils.getAsyncManager(request);

   try {
        // 省略。。。  
      try {
         // 省略。。。  
         // 实际请求处理，
         mv = ha.handle(processedRequest, response, mappedHandler.getHandler());
         // 判断开启了异步请求，则直接返回，当前请求会被保存到AsyncContext中，tomcat线程会被释放
         if (asyncManager.isConcurrentHandlingStarted()) {
            return;
         }

         // 省略。。。  
      }
      // 省略。。。  
   }
   // 省略。。。  
   finally {
      if (asyncManager.isConcurrentHandlingStarted()) {
         // 处理异步拦截器的回调
         if (mappedHandler != null) {
            mappedHandler.applyAfterConcurrentHandlingStarted(processedRequest, response);
         }
      }
      // 省略。。。  
   }
}

AbstractHandlerMethodAdapter

@Override
@Nullable
public final ModelAndView handle(HttpServletRequest request, HttpServletResponse response, Object handler)
      throws Exception {

   return handleInternal(request, response, (HandlerMethod) handler);
}

RequestMappingHandlerAdapter

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

   // 省略。。。

   // 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);
   }
    // 省略。。。
   return mav;
}


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

   try {
      // 省略。。。
      // 当异步请求处理完成后会重新交给Spring MVC处理
      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);
      }
      // 异步请求第一次进来时，调用Controller处理
      invocableMethod.invokeAndHandle(webRequest, mavContainer);
      if (asyncManager.isConcurrentHandlingStarted()) {
         return null;
      }

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

ServletInvocableHandlerMethod

public void invokeAndHandle(ServletWebRequest webRequest, ModelAndViewContainer mavContainer,
      Object... providedArgs) throws Exception {
   // 处理请求，获得返回值
   Object returnValue = invokeForRequest(webRequest, mavContainer, providedArgs);
   // 省略。。。
   try {
      // 处理返回值
      this.returnValueHandlers.handleReturnValue(
            returnValue, getReturnValueType(returnValue), mavContainer, webRequest);
   }
   // 省略。。。
}

HandlerMethodReturnValueHandlerComposite

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

@Nullable
private HandlerMethodReturnValueHandler selectHandler(@Nullable Object value, MethodParameter returnType) {
   boolean isAsyncValue = isAsyncReturnValue(value, returnType);
   for (HandlerMethodReturnValueHandler handler : this.returnValueHandlers) {
      if (isAsyncValue && !(handler instanceof AsyncHandlerMethodReturnValueHandler)) {
         continue;
      }
      // 这里获得Callable、DeferredResult异步请求的返回处理器
      if (handler.supportsReturnType(returnType)) {
         return handler;
      }
   }
   return null;
}

CallableMethodReturnValueHandler

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

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

   Callable callable = (Callable) returnValue;
   // 异步处理
   WebAsyncUtils.getAsyncManager(webRequest).startCallableProcessing(callable, mavContainer);
}

DeferredResultMethodReturnValueHandler

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

   // 省略。。。

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

3. 小结：

到这里Spring MVC将请求交给Callable、DeferredResult异步返回处理器。Spring MVC默认提供了15种返回处理器，根据返回值的类型决定使用某个返回处理器，比如常用的ModelAndView、ResponseBody等。

四. AsyncContext

目前被广泛使用的配置中心nacos(阿里开源)又是如何实现配置变更的长轮询呢？
LongPollingService

public void addLongPollingClient(HttpServletRequest req, HttpServletResponse rsp, Map clientMd5Map,
        int probeRequestSize) {
    
    // 省略。。。
    
    // 直接通过Request开启异步处理
    final AsyncContext asyncContext = req.startAsync();
    
    // 省略。。。
    
    ConfigExecutor.executeLongPolling(
            new ClientLongPolling(asyncContext, clientMd5Map, ip, probeRequestSize, timeout, appName, tag));
}

class ClientLongPolling implements Runnable {
    
    @Override
    public void run() {
        asyncTimeoutFuture = ConfigExecutor.scheduleLongPolling(new Runnable() {
            @Override
            public void run() {
                try {
                    // 省略。。。
                    if (isFixedPolling()) {
                        // 省略。。。
                        if (changedGroups.size() > 0) {
                            sendResponse(changedGroups);
                        } else {
                            sendResponse(null);
                        }
                    } else {
                        // 省略。。。
                        sendResponse(null);
                    }
                } catch (Throwable t) {
                    LogUtil.DEFAULT_LOG.error("long polling error:" + t.getMessage(), t.getCause());
                }
            }
        }, timeoutTime, TimeUnit.MILLISECONDS);
        allSubs.add(this);
    }
    
    void sendResponse(List changedGroups) {
        // 省略。。。
        generateResponse(changedGroups);
    }
    
    void generateResponse(List changedGroups) {
        if (null == changedGroups) {
            // 发送一个请求处理完成事件，通知Web容器处理响应
            asyncContext.complete();
            return;
        }
        // 省略。。。
        try {
            // 发送一个请求处理完成事件，通知Web容器处理响应
            asyncContext.complete();
        } catch (Exception ex) {
            PULL_LOG.error(ex.toString(), ex);
            // 发送一个请求处理完成事件(这里因为是一个轮询请求，并没有对异常做处理)，通知Web容器处理响应
            asyncContext.complete();
        }
    }
    // 省略。。。
}

小结：Request直接开始异步处理，生成AsyncContext异步上下文对象，并将当前Request&Response保存到AsyncContext中。当异步处理完成后通过AsyncContext.complete()方法发送一个处理完成事件通知Web容器处理最终响应。

五. 总结

Callable & DeferredResult 处理逻辑基本相同，Callable适用于接口处理耗时长的场景，DeferredResult适用于监听某个状态，等待其他线程改变监听状态值。Appollo(携程开源)监听配置变更的长轮询使用DeferredResult实现。Callable & DeferredResult都是Spring MVC封装的、使用简单的非阻塞长轮询实现方案，但它们底层也都是基于AsyncContext实现。