Spring MVC 集成Servlet3.0 AsyncContext异步请求实现

Servlet 和基于 Servlet 的容器（例如Tomcat）默认处理请求和处理业务是同一条线程，即有请求进来后，分配一条线程接收并处理请求，并继续执行业务处理，直到所有处理结束才完成响应，线程才会释放。这使得 Servlet 对业务方法的调用变成一种阻塞调用，效率较低。

Servlet3.0 提供了AsyncContex 来异步处理请求，将请求线程与业务处理线程分离。

异步处理请求

异步处理请求

Servlet 容器基本上都有线程池来维护请求线程，会有最大线程数限制。例如，Spring Boot默认嵌入的 Tomcat 默认最大线程数是 200。

业务处理会有复杂业务和简单业务，当并发量高时可能会因为复杂业务长时间占用线程而耗完容器线程池中的所有线程，继而导致有简单业务的请求进来也无法快速处理，影响系统性能，产生性能瓶颈。

需要将请求线程与业务处理线程分离。步骤大概如下：

1. 请求线程接收请求完成后，交给异步线程处理业务。
2. 异步线程处理业务完成后，回调容器的线程完成响应。

请求线程与业务处理线程分离，请求线程接收完请求后就交给业务线程处理业务，请求线程不会阻塞可以继续接收其它请求，业务线程处理即使耗时长些也不影响请求的接收。

假如正在处理业务的线程数量超过请求线程池的数量，这时进来一个简单的请求仍可以快速响应。通常处理业务的线程也会由线程池来维护，最大业务线程数会远远大于接收请求的线程池，并使用一个较大的任务队列维护排队的任务。

下面演示请求同步阻塞 和 异步处理示例。以 SpringBoot 默认嵌入的 Tomcat 为例。

异步请求示例

SpringBoot 默认嵌入的 Tomcat 的最大线程数默认是 200。为了便于测试，使用下面配置将最大线程数改为 1 。

server.tomcat.threads.max=1

模拟复杂业务耗时较长，让线程睡眠 8 秒。使用浏览器或 Postman 同时发两个或多个请求，观察结果。

请求线程同步阻塞示例代码：

@RestController
@RequestMapping("/asyncServlet")
public class AsyncServletController {

    @GetMapping("/asyncContext2")
    public void asyncContext2(HttpServletRequest servletRequest, HttpServletResponse servletResponse) {
        System.out.println("接收请求线程开始:" + Thread.currentThread().getName() + ";" + DateUtil.format(new Date(), DateUtil.PATTERN_YMDHMS_SLASH));
        try {
            servletResponse.getWriter().println("Hello");
            System.out.println(Thread.currentThread().getName() + ": 睡眠8秒");
            Thread.sleep(8000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        System.out.println("接收请求线程结束," + DateUtil.format(new Date(), DateUtil.PATTERN_YMDHMS_SLASH));
        System.out.println("--------------------------------");
    }
}

执行结果：

接收请求线程开始:http-nio-80-exec-1;2022/07/28 20:26:02
http-nio-80-exec-1: 睡眠8秒
接收请求线程结束,2022/07/28 20:26:10
--------------------------------
接收请求线程开始:http-nio-80-exec-1;2022/07/28 20:26:10
http-nio-80-exec-1: 睡眠8秒
接收请求线程结束,2022/07/28 20:26:18
--------------------------------

同时发两个请求，从结果可以看到，第二个请求是在第一个请求完成后开始的，在第一个请求未完成前，第二个请求被一直阻塞。

请求线程异步处理示例：

@RestController
@RequestMapping("/asyncServlet")
public class AsyncServletController {

    @GetMapping("/asyncContext2")
    public void asyncContext2(HttpServletRequest servletRequest, HttpServletResponse servletResponse) {
        String s = RandomStringUtils.randomNumeric(4);
        System.out.println(s + ":接收请求线程开始:" + Thread.currentThread().getName() + ";" + DateUtil.format(new Date(), DateUtil.PATTERN_YMDHMS_SLASH));
        AsyncContext asyncContext = servletRequest.startAsync();
        new Thread(() -> {
            System.out.println(s + ":异步线程启动:" + Thread.currentThread().getName());
            PrintWriter writer = null;
            try {
                Thread.sleep(8000);
                ServletResponse response = asyncContext.getResponse();
                response.setContentType("application/json;charset=UTF-8");
                writer = response.getWriter();
                writer.println("Hello");
                asyncContext.complete();
                System.out.println(s + ":异步线程结束;" + DateUtil.format(new Date(), DateUtil.PATTERN_YMDHMS_SLASH));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
        System.out.println(s + ":接收请求线程结束," + DateUtil.format(new Date(), DateUtil.PATTERN_YMDHMS_SLASH));
    }
}

执行结果：

2698:接收请求线程开始:http-nio-80-exec-1;2022/07/28 22:30:18
2698:接收请求线程结束,2022/07/28 22:30:18
2698:异步线程启动:Thread-17
2359:接收请求线程开始:http-nio-80-exec-1;2022/07/28 22:30:18
2359:接收请求线程结束,2022/07/28 22:30:18
2359:异步线程启动:Thread-18
2359:异步线程结束;2022/07/28 22:30:26
2698:异步线程结束;2022/07/28 22:30:26

从结果可以看到，虽然只有一个线程处理请求，但两个同时到达的请求几乎是同时处理完的，不会因为上一个请求的业务耗时较长而阻塞后续的请求。这大大提高了系统性能。

Servlet3.0 AsyncContext

Servlet 3.0 引入了异步处理，在Servlet 3.1中又引入了非阻塞 IO 来进一步增强异步处理的性能。

HttpServletRequest 对象提供了一个AsyncContext对象，该对象构成了异步处理的上下文。可以从 AsyncContext对象中获取 Request 和 Response 对象。

可以将 AsyncContext 从当前线程传给异步线程，并在新的线程中完成对请求的处理并返回结果给客户端，初始线程便可以还回给容器线程池以处理更多的请求。如此，通过将请求从一个线程传给另一个线程处理的过程便构成了Servlet 3.0中的异步处理。

AsyncContext应用

@RestController
@RequestMapping("/asyncServlet")
public class AsyncServletController {
    /**
     * @param servletRequest 请求
     * @desc 异步测试,  asyncContext.start()开启线程
     */
    @GetMapping("/asyncContext3")
    public void asyncContext3(HttpServletRequest servletRequest, HttpServletResponse servletResponse) {
        String s = RandomStringUtils.randomNumeric(4);
        System.out.println(s + ":接收请求线程开始:" + Thread.currentThread().getName() + ";" + DateUtil.format(new Date(), DateUtil.PATTERN_YMDHMS_SLASH));
        AsyncContext asyncContext = servletRequest.startAsync();
        asyncContext.start(new Thread(new Work(asyncContext)));
        System.out.println(s + ":接收请求线程结束," + DateUtil.format(new Date(), DateUtil.PATTERN_YMDHMS_SLASH));
    }
    /**
     * @param servletRequest 请求
     * @desc 异步测试
     */
    @GetMapping("/asyncContext2")
    public void asyncContext2(HttpServletRequest servletRequest) {
        System.out.println("接收请求线程开始:" + Thread.currentThread().getName());
        AsyncContext asyncContext = servletRequest.startAsync();
        // 设置异步超时
        asyncContext.setTimeout(10000);
        // 设置监听器
        asyncContext.addListener(new ServletAsyncListener());
        new Thread(() -> {
            System.out.println("异步线程启动:" + Thread.currentThread().getName());
            PrintWriter writer = null;
            try {
//                Thread.sleep(8000);
                ServletResponse response = asyncContext.getResponse();
                response.setContentType("application/json;charset=UTF-8");
                writer = response.getWriter();
                HashMap<String, Object> map = new HashMap<>();
                map.put("id", 100L);
                map.put("message", "Hello World!");
                writer.println(JSON.toJsonString(map));
                asyncContext.complete();
                System.out.println("异步线程结束");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
        System.out.println("接收请求线程结束");
    }

    /**
     * @param request 请求
     * @desc 异步测试
     */
    @RequestMapping(value = "/asyncContext1", method = RequestMethod.GET)
    public void asyncContext1(HttpServletRequest request) {
        System.out.println("接收请求线程开始:" + Thread.currentThread().getName());
        AsyncContext asyncContext = request.startAsync();
        // 设置异步超时
        asyncContext.setTimeout(10000);
        // 设置监听器
        asyncContext.addListener(new ServletAsyncListener());
        new Thread(new Work(asyncContext)).start();
        System.out.println("接收请求线程结束");
    }
}

/**
 * @desc 任务
 */
public class Work implements Runnable {

    private AsyncContext asyncContext;

    public Work(AsyncContext asyncContext) {
        this.asyncContext = asyncContext;
    }

    @Override
    public void run() {
        System.out.println("异步线程启动:" + Thread.currentThread().getName());
        try {
            ServletResponse response = asyncContext.getResponse();
            response.setContentType("application/json;charset=UTF-8");
            PrintWriter writer = response.getWriter();
            HashMap<String, Object> map = new HashMap<>();
            map.put("id", 100L);
            map.put("message","Hello World!");
            writer.println(JSON.toJsonString(map));
        } catch (IOException e) {
            e.printStackTrace();
        }
        asyncContext.complete();
        System.out.println("异步线程结束");
    }
}

AsyncContext监听器

AsyncListener 是 Servlet 提供与 AsyncContext 关联的事件监听器。在触发事件时会回调监听器中对应的方法。

public interface AsyncListener extends EventListener {

    /**
     * 在容器处理了对 AsyncContext#complete() 的调用后，触发此事件
     */
    void onComplete(AsyncEvent event) throws IOException;

    /**
     * 异步操作超时触发此事件，在容器因超时而采取任何操作之前触发
     */
    void onTimeout(AsyncEvent event) throws IOException;

    /**
     * 异步操作期间发生了错误触发此事件，在容器因错误而采取任何操作之前触发
     */
    void onError(AsyncEvent event) throws IOException;

    /**
     * 在添加此侦听器的 AsyncContext 完成后，
     * 对 ServletRequest.startAsync() 进行了新调用，则会触发此事件。
     * PS:这个事件好像永远不会触发，异步开启中的状态不可以再开启，开启会报错
     */
    void onStartAsync(AsyncEvent event) throws IOException;
}

Spring MVC集成AsyncContext

统一异步处理基类

创建一个 Controller 抽象基类和处理方法，统一在此方法里将业务转异步处理。

所有 Controller 继续此基类，调用处理方法进行处理。

package com.gxitsky.async.design.controller;

import com.gxitsky.async.design.listener.ServletAsyncListener;
import com.gxitsky.utils.JSON;
import org.springframework.web.context.request.RequestAttributes;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;

import javax.annotation.Nullable;
import javax.servlet.AsyncContext;
import javax.servlet.ServletResponse;
import javax.servlet.http.HttpServletRequest;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.HashMap;
import java.util.concurrent.*;

/**
 * @author gxing
 * @desc Controller基类
 * @date 2022/7/28
 */
public abstract class BaseController {

    private static ExecutorService exececutor;
    private static final int corePoolSize = 5;
    private static final int maximumPoolSize = 200;
    private static final int keepAliveTime = 10 * 1000;
    private static final int maxQueueTask = 5 * 1000;
    private static final long TIME_OUT = 10 * 1000;

    static {
        exececutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime,
                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(maxQueueTask));
    }

    /**
     * @param servletRequest 请求
     * @param callable       业务
     * @desc 统一异步处理业务
     * @author gxing
     * @date 2022/7/30 16:35
     */
    public <T> void handler(HttpServletRequest servletRequest, Callable<T> callable) {
//        System.out.println("主线程开始:" + Thread.currentThread().getName() + ":" + DateUtil.format(new Date(), DateUtil.PATTERN_YMDHMS_SLASH));
        AsyncContext asyncContext = servletRequest.startAsync();
        asyncContext.setTimeout(TIME_OUT);
        asyncContext.addListener(new ServletAsyncListener());
        exececutor.execute(() -> {
//            System.out.println("异步线程开启:" + Thread.currentThread().getName());
            try {
//                Thread.sleep(5000);
                T result = callable.call();
                ServletResponse response = asyncContext.getResponse();
                response.setContentType("application/json;charset=UTF-8");
                PrintWriter writer = response.getWriter();
                writer.println(JSON.toJsonString(result));
            } catch (Exception e) {
                throw new RuntimeException(e);
            } finally {
                asyncContext.complete();
            }
//            System.out.println("异步线程结束:" + Thread.currentThread().getName());
        });
//        System.out.println("主线程结束:" + DateUtil.format(new Date(), DateUtil.PATTERN_YMDHMS_SLASH));
    }
}

业务集成使用示例

1. 业务 Controller 继承 BaseController，将业务统一交给异步方法处理

   /**
    * @author gxing
    * @desc 消息
    * @date 2022/7/28
    */
   @RestController
   @RequestMapping("/message")
   public class MessageController extends BaseController {
   
       @Autowired
       private MessageService messageService;
   
       /**
        * @param servletRequest 请求
        * @desc 查详情
        * @author gxing
        * @date 2022/7/30 16:43
        */
       @GetMapping("/detail")
       public void detail(HttpServletRequest servletRequest) {
           handler(servletRequest, () -> messageService.detail());
       }
   }

2. Service 接口和实现类与通常的写法没区别。

   业务接口：

   public interface MessageService {
       Message detail();
   }

   业务接口实现：

   @Service
   public class MessageServiceImpl implements MessageService {
   
       @Override
       public Message detail() {
               Message message = new Message();
               message.setId(100L);
               message.setBody("Hello World!");
               return message;
       }
   }