likes
comments
collection
share

浅谈spring servlet异步编程

作者站长头像
站长
· 阅读数 32

1.前言

身为一个互联网从业人员有一部分工作主要做的是接口的设计与开发，一般来说接口请求同步处理即可，但是耗时长接口流量激增时，会导致大量的web容器线程被占用，可能会拖慢所有接口响应耗时，这时需要我们考虑改为异步处理是不是一种更好的方案。

总体来说异步处理的流程适合业务处理比较耗时而导致主线程长时间等待的场景。下面我们谈一下servlet3.0的异步处理以及springMvc是如何封装该特性来方便开发人员的使用

2.原生Servlet3.0 异步处理

我们先了解一下同步请求处理流程。如下图[同步请求线程模型]所示:

每个请求过来web容器线程池都需要分配一个线程去处理请求，处理完请求后线程才能被释放，一旦个别接口耗时上升就可能会导致web容器线程被无效占用，容器线程不够用。

浅谈spring servlet异步编程

如果我们使用异步处理，便可以快速释放容器线程，提高处理请求效率。具体简略流程可以看下图[异步请求线程模型]：

在主线程中开启异步处理，主线程将请求交给其他线程去处理，主线程就结束了，被放回主线程池，由其他线程继续处理请求。 浅谈spring servlet异步编程

servlet异步处理，代码示例如下

//1.设置@WebServlet的asyncSupported属性为true，表示支持异步处理
@WebServlet(name = "AsyncServlet1",
        urlPatterns = "/asyncServlet1",
        asyncSupported = true  
)
public class AsyncLongRunningServlet extends HttpServlet {

  @Override
  protected void service(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
    long st = System.currentTimeMillis();
    logger.info("主线程：" + Thread.currentThread() + "-" + System.currentTimeMillis() + "-start");
    //2、启动异步处理：调用req.startAsync(request,response)方法，获取异步处理上下文对象AsyncContext
    AsyncContext asyncContext = request.startAsync(request, response);
    //3、调用start方法异步处理，调用这个方法之后主线程就结束了
    asyncContext.start(() -> {
      long stSon = System.currentTimeMillis();
      System.out.println("子线程：" + Thread.currentThread() + "-" + System.currentTimeMillis() + "-start");
      try {
        //这里休眠10秒，模拟业务耗时
        TimeUnit.SECONDS.sleep(10);
        //这里是子线程，请求在这里被处理了
        asyncContext.getResponse().getWriter().write("ok");
        //4、调用complete()方法，表示请求请求处理完成
        asyncContext.complete();
      } catch (Exception e) {
        e.printStackTrace();
      }
      System.out.println("子线程：" + Thread.currentThread() + "-" + System.currentTimeMillis() + "-end,耗时(ms):" + (System.currentTimeMillis() - stSon));
    });
    System.out.println("主线程：" + Thread.currentThread() + "-" + System.currentTimeMillis() + "-end,耗时(ms):" + (System.currentTimeMillis() - st));

  }

}

使用异步编程主要有四步

  1. 设置@WebServlet的asyncSupported属性为true，表示支持异步处理
  2. 启动异步处理：调用req.startAsync(request,response)方法，获取异步处理上下文对象AsyncContext
  3. 调用start方法异步处理，调用这个方法之后主线程就结束了,Runnable会有AsyncContext内部的Executor处理
  4. 调用complete()方法，表示请求处理完成,工作线程结束

通过日志可以看到主线程很快就被释放了,由工作线程处理业务逻辑。 浅谈spring servlet异步编程 由于日常使用Spring生态较多，下面我们主要说一下如何在SpringMvc中如何使用。

3.SpringMvc异步接口实现方式

我们先说一下异步接口常用的四种实现方式

3.1不关心返回值的接口

首先我们如果不关心返回值，可以直接借助java线程机制实现异步处理。无需开启servlet异步支持，使用起来更方便

代码示例1:

  @GetMapping("testAsync1")
  public void testAsync1(){
      logger.info("异步请求开始");
      new Thread(this::asyncExecute1).start();
      logger.info("异步请求结束");
  }
      @SneakyThrows
    public void asyncExecute1(){
        logger.info("业务处理开始");
        TimeUnit.SECONDS.sleep(10);
        logger.info("业务处理结束");
    }

如果接口不需要返回值，我们就可以上面那样将实际的业务处理逻辑抽成方法，然后委托给线程异步执行。如上文的代码实例1

这种方式在spring中有更简单的实现方案，我们可以借助注解@Async，如代码示例2 代码示例2:

    @GetMapping("testAsync2")
    public void testAsync2(){
        StopWatch stopWatch = new StopWatch(" @Async test 容器线程");
        stopWatch.start("容器线程");
        logger.info("异步请求开始");
        testService.asyncExecute2();
        logger.info("异步请求结束");
        stopWatch.stop();
        logger.info(String.format("%s秒",stopWatch.getTotalTimeSeconds()));
    }
    @Async
    @SneakyThrows
    public void asyncExecute2(){
        StopWatch stopWatch = new StopWatch(" @Async test 工作线程");
        stopWatch.start("工作线程");
        logger.info("业务处理开始");
        TimeUnit.SECONDS.sleep(10);
        logger.info("业务处理结束");
        stopWatch.stop();
        logger.info(String.format("%s秒",stopWatch.getTotalTimeSeconds()));
    }

在使用该注解时，业务代码和实际调用代码不能在一个类中，因为Spring一般使用JDK动态代理处理目标对象。如@Async会建一个代理类，如果这个类被同一个类下面其它方法调用，会直接访问这个类的实体bean，没有访问到对应的代理类，注解无法起作用

上面代码的执行，主要是两部分，首先tomcat从处理请求的线程池中获取线程A，由线程A处理请求执行方法testAsync，当执行到方法asyncExecute便将工作委托给线程B我们自定义线程或者是spring的工作线程执行。通过代码示例2的执行日志可以印证，主线程直接返回，工作线程异步处理

浅谈spring servlet异步编程

3.2返回值为Callable

上面的接口和第一个例子相比虽然只是由空返回变为Callable，但是内部的处理机制却大有不同。如果不关心返回值，我们可以直接利用java的线程机制即可实现异步处理，反之需要返回值无法通过java的线程来实现。代码示例如下:

    @GetMapping("/testCallable")
    @ResponseBody
    public Callable<String> test2() {
        StopWatch stopWatch = new StopWatch(" Callable test 容器线程");
        stopWatch.start("容器线程");
        logger.info("返回值Callable 异步请求开始");
        Callable<String> callable = () -> {
            StopWatch stopWatch1 = new StopWatch(" Callable test 工作线程");
            stopWatch1.start("工作线程");
            logger.info("返回值Callable 业务处理开始");
            //模拟结果延时返回
            Thread.sleep(10000);
            logger.info("返回值Callable 业务处理结束");
            stopWatch1.stop();
            logger.info(String.format("%s秒",stopWatch1.getTotalTimeSeconds()));
            return "OK";
        };
        logger.info("返回值Callable 异步请求结束");
        stopWatch.stop();
        logger.info(String.format("%s秒",stopWatch.getTotalTimeSeconds()));
        return callable;
    }

代码运行日志如下，主线程释放但是响应未结束 浅谈spring servlet异步编程

接口响应结果如下，接口耗时为10s 376ms，符合预期。从下图我们可以看到浏览器接受到结果【OK】，这表明是异步处理完成后，响应才结束。 浅谈spring servlet异步编程

上面的例子更底层的原因是利用了Servlet 3.0对异步处理的支持，spring对该支持做了封装所以使用起来很方便。下面我们说一下spring如何封装了该特性。

3.3异步处理原理简述

在说异步处理之前需要先说一下springMvc的执行流程，简单来说，用户发起请求由DispatcherServlet统一处理，首先会根据请求类型、返回值等等从HandlerMapping获取handler然后通过HandlerAdapter执行handler,通过下面这张图可以更清晰的了解springMvc的执行流程 浅谈spring servlet异步编程 在了解springMvc后我们大概也能猜到，springMvc正是通过返回值的类型来判断是否对该请求进行异步处理。源码调用顺序大致如下 【下面的代码只保留了核心代码，无关代码已删除】

//调度入口
//方法路径 org.springframework.web.servlet.DispatcherServlet#doDispatch
	protected void doDispatch(HttpServletRequest request, HttpServletResponse response) throws Exception {
		// 获取处理当前请求的handler
		mappedHandler = getHandler(processedRequest);
		// 获取执行当前请求的handler的adapter
		HandlerAdapter ha = getHandlerAdapter(mappedHandler.getHandler());
		// 开始调度
		mv = ha.handle(processedRequest, response, mappedHandler.getHandler());
	}

上面三行代码基本对应上图springMvc执行流程

下面的代码主要是展现如何获取实际处理请求handler的过程

//方法路径 org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod#invokeAndHandle
	public void invokeAndHandle(ServletWebRequest webRequest, ModelAndViewContainer mavContainer,
								Object... providedArgs) throws Exception {

		//获取返回值
		Object returnValue = invokeForRequest(webRequest, mavContainer, providedArgs);
		this.returnValueHandlers.handleReturnValue(
				returnValue,
				getReturnValueType(returnValue), //获取返回值类型
				mavContainer, webRequest);
	}

获取到返回值类型后开始根据返回值类型获取实际处理请求的handler

//方法路径 org.springframework.web.method.support.HandlerMethodReturnValueHandlerComposite#handleReturnValue
	public void handleReturnValue(@Nullable Object returnValue, MethodParameter returnType,
								  ModelAndViewContainer mavContainer, NativeWebRequest webRequest) throws Exception {

		//根据返回值类型获取实际处理handler
		HandlerMethodReturnValueHandler handler = selectHandler(returnValue, returnType);
		if (handler == null) {
			throw new IllegalArgumentException("Unknown return value type: " + returnType.getParameterType().getName());
		}
		//开始调用
		handler.handleReturnValue(returnValue, returnType, mavContainer, webRequest);
	}

通过debug我们可以看到handler的类型是CallableMethodReturnValueHandler,然后通过调用CallableMethodReturnValueHandler中的handleReturnValue方法开始处理返回值为Callable的请求

//方法路径 org.springframework.web.servlet.mvc.method.annotation.CallableMethodReturnValueHandler#handleReturnValue
	public void handleReturnValue(@Nullable Object returnValue, MethodParameter returnType, ModelAndViewContainer mavContainer, NativeWebRequest webRequest) throws Exception {
		Callable<?> callable = (Callable<?>) returnValue;
		//开始执行
		WebAsyncUtils.getAsyncManager(webRequest).startCallableProcessing(callable, mavContainer);
	}

下面开始对异步处理的核心类WebAsyncManager的解析 由上面的代码可知WebAsyncManager.tartCallableProcessing方法处理的请求核心代码如下

	public void startCallableProcessing(Callable<?> callable, Object... processingContext) throws Exception {
		Assert.notNull(callable, "Callable must not be null");
		startCallableProcessing(new WebAsyncTask(callable), processingContext);
	}

首先将callable包装为WebAsyncTask然后调用重载方法

在下面的代码中callable将由taskExecutor任务执行器执行，并将获取到的异步结果作为再一次发起调度的入参，不过这次的入参是返回值Callable的泛型。我们注意看setConcurrentResultAndDispatch()这个方法,就是通过这个方法发起的调度，如果我们深入的看方法的内部逻辑会发现该方法使用容器的调度器将请求再次分派到容器执行请求过程，实际上再一次调用了入口方法org.springframework.web.servlet.DispatcherServlet#doDispatch。

所以如果你的泛型还是Callable那么这次调度还是异步的，因为上面这段代码被复用了，直到泛型类型不是Callable、WebAsyncTask、DeferredResult等不需要异步处理的类型为止。假如你的返回值是Callable<Callable<Callable<String>>>，那么他将会执行三次异步调度，第四次调度后结束。

	public void startCallableProcessing(final WebAsyncTask<?> webAsyncTask, Object... processingContext)
			throws Exception {
		try {
			Future<?> future = this.taskExecutor.submit(() -> {
				Object result = null;
				try {
					//获取异步结果
					result = callable.call();
				}
				catch (Throwable ex) {
					result = ex;
				}
				finally {
					result = interceptorChain.applyPostProcess(this.asyncWebRequest, callable, result);
				}
				//再次发起调度
				setConcurrentResultAndDispatch(result);
			});
		}
	}

至此关于异步处理的源码解析基本结束，其他的和上面的流程基本一样。方法调用流程图如下图所示

浅谈spring servlet异步编程

3.4返回值为WebAsyncTask

WebAsyncTask和Callable相比提供了三个回调函数，方便用户使用。分别是超时、异常、完成。

public void onTimeout(Callable<V> callback) {
  	this.timeoutCallback = callback;
  }

  public void onError(Callable<V> callback) {
  	this.errorCallback = callback;
  }

  public void onCompletion(Runnable callback) {
  	this.completionCallback = callback;
  }

有一点需要注意onError并不是处理业务代码抛出的异常，而是处理执行Callable时出现的异常，可以理解为是系统异常，在业务代码抛出的异常会被onCompletion处理，在上文解读startCallableProcessing方法时大家可以看到callable.call()是在try catch中，业务异常都被捕获正常返回了。

3.5返回值为DeferredResult

DeferredResult和上面的使用方式相比最大的特点是可以灵活的控制异步调用的生命周期。什么时候开始执行，结束都可以由开发人员自己控制，方便我们处理一些比较复杂的异步场景。

//方法路径 org.springframework.web.context.request.async.WebAsyncManager#startDeferredResultProcessing
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);
		}
	}

我们仔细看一下deferredResult.setResultHandler这个方法注册了一个函数，在函数体中调用了方法setConcurrentResultAndDispatch在异步结束后进行二次转发。该函数方法是在deferredResult调用setResult()时触发或者在超时后。通过安排setResult的调用时机，我们可以自由的控制异步的结束。下面是一个简单的小例子。

    private static final Map<String,DeferredResult<String>> taskMap = new ConcurrentHashMap<>();
    //创建任务
  @RequestMapping("/createTask")
  public DeferredResult<String> createTask(String uuid) {
    LOGGER.info("ID[{}]任务开始",uuid);
    StopWatch stopWatch = new StopWatch(" DeferredResult test 容器线程");
    stopWatch.start("容器线程");
    LOGGER.info("返回值DeferredResult 异步请求开始");
    //超时时间100s
    DeferredResult<String> deferredResult = new DeferredResult<>(100000L);
    StopWatch t = new StopWatch(" DeferredResult test 工作线程");
    t.start("工作线程");
    deferredResult.onCompletion(()->{
      LOGGER.info("返回值DeferredResult onCompletion 工作线程处理完毕");
      t.stop();
      LOGGER.info(String.format("%s秒",t.getTotalTimeSeconds()));
    });
    taskMap.put(uuid, deferredResult);
    LOGGER.info("返回值DeferredResult 异步请求结束");
    stopWatch.stop();
    LOGGER.info(String.format("%s秒",stopWatch.getTotalTimeSeconds()));
    return deferredResult;
  }
  //查询任务状态
   @RequestMapping("/queryTaskState")
  public String queryTaskState(String uuid) {
    DeferredResult<String> deferredResult = taskMap.get(uuid);
    if (deferredResult == null) {
      return "未查询到任务,uid:" + uuid;
    }
    if (deferredResult.hasResult()) {
      return deferredResult.getResult().toString();
    } else {
      LOGGER.info("ID[{}]任务进行中",uuid);
      return "进行中";
    }
  }
 //模拟第三方调用通知任务结束
  @RequestMapping("/changeTaskState")
  public String changeTaskState(String uuid) {
    DeferredResult<String> deferredResult = taskMap.remove(uuid);
    if (deferredResult == null) {
      return "未查到到任务";
    }
    if (deferredResult.hasResult()) {
      return "已完成，无需再次设置";
    } else {
      //未完成设置为完成
      deferredResult.setResult("已完成");
      LOGGER.info("将任务ID{},设置为处理完成",uuid);
      return "已完成";
    }
  }

我们通过createTask创建任务并且在初始化时设置一个超时时间。将任务保存到map后者队列中，然后在合适的时机我们可以通过另一个请求来设置任务的结束状态，未得到完成通知或者超时之前请求不会结束，如下图 浅谈spring servlet异步编程 方法changeTaskState可以直接通过任务ID查找到对应任务然后结束任务。如下图执行日志标红部分，只有设置任务已完成，createTask接口响应才会结束 浅谈spring servlet异步编程

3.6小结

除了上面这三种springMvc会自动处理为异步的返回值外还有其他的比如ResponseBodyEmitter，具体的大家可以了解一下AsyncHandlerMethodReturnValueHandler接口下的实现类。

了解异步编程的四种实现方式后我们可以总结一下他们的使用场景

  1. 不关心返回值，可以直接使用多线程机制异步处理
  2. 在上面的基础上需要返回值就需要考虑Callable
  3. 如果还需要超时时间配置、执行完成回调、执行异常和超时后的回调等功能那么WebAsyncTask是一个很好的选择
  4. 如果我们需要自己控制异步的执行，如一个调用涉及多个服务，比如客户端请求服务A，服务A需要调用服务B，服务B处理完在通知服务A，然后服务A在响应客户端，这种情况比较适合用DeferredResult

后三种的执行流程也大同小异。

  1. SpringMVC 根据返回值类型查找处理器returnValueHandler
  2. 在对应的处理器中开启异步
  3. 容器线程被释放，但是response还是处于打开状态
  4. 业务出完完成后，再次执行调度流程，如果返回值不是Callable、WebAsyncTask、DeferredResult，结果将返回给客户端结束此次请求。

4.实际应用

DeferredResult 比较灵活 我们经常用它处理依赖第三方服务的耗时业务逻辑。

比如我们在使用转转内部效率平台Beetle时，开发功能需要新建分支，涉及的服务较多如tapd、gitlab、jenkins，假如需要DeferredResult来解耦，可能流程如下：

  1. 用户发起新建分支请求，请求新建分支接口，接口返回值是DeferredResult
  2. 首先创建分支信息
  3. 往新建分支队列里面发送一个新建请求消息，同时保存要返回的 DeferredResult 实例
  4. 资源创建服务监听新建分支请求队列，收到消息后根据分支类型及其配置创建和申请资源
  5. 操作成功之后，资源创建服务将操作结果广播发送到新建分支响应队列，使得每个实集群实例都能收到消息
  6. beetle服务监听新建分支响应队列，收到结果更新数据，取出第三步中保存的 DeferredResult然后setResult
  7. 用户收到响应，新建分支结束，开始coding

代码大致如下，我们可以参考上面的小例子，首先我们需要个map对deferredResult进行存储，然后引入消息队列解耦调用逻辑。【下面的代码只保留了核心代码，无关代码已删除】

  private static final Map<String,DeferredResult<String>> taskMap = new ConcurrentHashMap<>();
  @RequestMapping("/createTask")
  public DeferredResult<String> createBranch(String projectName) {
    Branch branch = branchManageService.createBranch(projectName);
    DeferredResult<String> deferredResult = new DeferredResult<>(5000L);
    deferredResult.onCompletion(() -> taskMap.remove(branch.getBranchName()));
    deferredResult.onTimeout(() -> taskMap.remove(branch.getBranchName()));
    // 发送创建分支请求到队列
    this.createBranchMQSender.send(branch);
    taskMap.putIfAbsent(branchName, deferredResult);
    return deferredResult;
  }

在收到创建分支请求后，将消息发送到创建分支请求队列，然后把对应的deferredResult保存下来，等待后续调用，最后返回deferredResult。此时容器线程已经释放。

   @ZZMQListener(group = "${create-branch-request-queue}")
  public void onMessage(MessageExt messageExt) {
    log.info("=============== 接受创建分支消息 ->{}", messageExt.toString());
    // 创建gitlab分支
    // 创建jenkins build job
    // 创建jenkins sonar job
    // 关联tapd需求
    TimeUnit.SECONDS.sleep(10);
    // 响应结果
    Response response = new Response();
    //发送响应结果
    this.responseCreateBranchSender.send(response);
  }

在创建资源服务收到创建分支mq消息后开始创建相关资源，然后将结果放入创建分支响应队列。交给创建分支服务消费

    @ZZMQListener(group = "${create-branch-response-queue}")
    public void onMessage(MessageExt messageExt) {
        log.info("=============== 接受创建分支结果信息 ->{}", messageExt.toString());
        Response response = JsonUtil.string2Object(messageExt.getBody(),Response.class);
        DeferredResult<String> deferredResult = taskMap.get(response.getBranchName());
        //更新分支信息
        branchManageService.updateBranch(response.getStatus());
        //结束异步
        deferredResult.setResult(response.getResult());
    }

创建分支服务收到消息后根据分支名拿到对应的deferredResult更新分支信息，设置响应结果结束请求。至此创建分支请求完成，用户无延迟的收到了响应。如果是单体架构可以不使用消息中间件，可以直接用java队列即可。多实例也可以使用redis作为消息中间件，redis更为轻量。具体可以参考下面的方案

  • 利用redis的发布订阅功能，实例可以构造一个有相同前缀的队列ID，比如createbranch-request-084Tkjfh和createbranch-response-084Tkjfh,然后消费方根据createbranch-request这个前缀获取到相关队列列表，消费消息后根据队列命名规则将结果塞到对应的响应队列createbranch-response-084Tkjfh中由请求方消费。其中还需要考虑资源的竞争，以及重复消费的问题，需要做好幂等。

假如我们不使用deferredResult可能需要前端设置一个定时任务轮询后端接口查询分支状态来确定操作的完成状态。

假设操作完成时间为20秒，前端每两秒查询一下后端接口，那么用户感知到操作完成需要22秒，这种处理方式随着操作完成时间的延长，或者轮询间隔的缩短，后端需要响应的次数愈多，服务也会有压力，并且响应结果的延迟是不可避免。deferredResult则给我们提供了一种更优雅的实现方式。

5.总结

上文这些异步开发模式，都可以算是一种响应式编程，都是面向流的、异步化的开发方式。这种方式可以在服务高负载时也能快速响应请求。这种模式有以下优点:

  • 每个请求模型都远离线程，并且可以以较少的线程数处理更多请求
  • 防止线程在等待I/O操作完成时阻塞
  • 轻松进行并行调用

在日常工作中我们可以尝试一下这种开发模式

6.后记

大家有想法可以在下方留言，欢迎大家交流互相学习

关于作者

丁赵雷，转转工程效率组，主要负责效率平台建设。

转转研发中心及业界小伙伴们的技术学习交流平台，定期分享一线的实战经验及业界前沿的技术话题。

关注公众号「转转技术」（综合性）、「大转转FE」（专注于FE）、「转转QA」（专注于QA），更多干货实践，欢迎交流分享~

转载自:https://juejin.cn/post/7208357215686131771
评论
请登录