Servlet 和基于 Servlet 的容器(例如Tomcat)默认处理请求和处理业务是同一条线程,即有请求进来后,分配一条线程接收并处理请求,并继续执行业务处理,直到所有处理结束才完成响应,线程才会释放。这使得 Servlet 对业务方法的调用变成一种阻塞调用,效率较低。
Servlet3.0 提供了AsyncContex 来异步处理请求,将请求线程与业务处理线程分离。
异步处理请求
异步处理请求
Servlet 容器基本上都有线程池来维护请求线程,会有最大线程数限制。例如,Spring Boot默认嵌入的 Tomcat 默认最大线程数是 200。
业务处理会有复杂业务和简单业务,当并发量高时可能会因为复杂业务长时间占用线程而耗完容器线程池中的所有线程,继而导致有简单业务的请求进来也无法快速处理,影响系统性能,产生性能瓶颈。
需要将请求线程与业务处理线程分离。步骤大概如下:
- 请求线程接收请求完成后,交给异步线程处理业务。
- 异步线程处理业务完成后,回调容器的线程完成响应。
请求线程与业务处理线程分离,请求线程接收完请求后就交给业务线程处理业务,请求线程不会阻塞可以继续接收其它请求,业务线程处理即使耗时长些也不影响请求的接收。
假如正在处理业务的线程数量超过请求线程池的数量,这时进来一个简单的请求仍可以快速响应。通常处理业务的线程也会由线程池来维护,最大业务线程数会远远大于接收请求的线程池,并使用一个较大的任务队列维护排队的任务。
下面演示请求同步阻塞 和 异步处理示例。以 SpringBoot 默认嵌入的 Tomcat 为例。
异步请求示例
SpringBoot 默认嵌入的 Tomcat 的最大线程数默认是 200。为了便于测试,使用下面配置将最大线程数改为 1 。
1
| server.tomcat.threads.max=1
|
模拟复杂业务耗时较长,让线程睡眠 8 秒。使用浏览器或 Postman 同时发两个或多个请求,观察结果。
请求线程同步阻塞示例代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
| @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("--------------------------------"); } }
|
执行结果:
1 2 3 4 5 6 7 8
| 接收请求线程开始: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 --------------------------------
|
同时发两个请求,从结果可以看到,第二个请求是在第一个请求完成后开始的,在第一个请求未完成前,第二个请求被一直阻塞。
请求线程异步处理示例:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27
| @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)); } }
|
执行结果:
1 2 3 4 5 6 7 8
| 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应用
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94
| @RestController @RequestMapping("/asyncServlet") public class AsyncServletController {
@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)); }
@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 {
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("接收请求线程结束"); }
@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("接收请求线程结束"); } }
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 关联的事件监听器。在触发事件时会回调监听器中对应的方法。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24
| public interface AsyncListener extends EventListener {
void onComplete(AsyncEvent event) throws IOException;
void onTimeout(AsyncEvent event) throws IOException;
void onError(AsyncEvent event) throws IOException;
void onStartAsync(AsyncEvent event) throws IOException; }
|
Spring MVC集成AsyncContext
统一异步处理基类
创建一个 Controller 抽象基类和处理方法,统一在此方法里将业务转异步处理。
所有 Controller 继续此基类,调用处理方法进行处理。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67
| 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.*;
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)); }
public <T> void handler(HttpServletRequest servletRequest, Callable<T> callable) {
AsyncContext asyncContext = servletRequest.startAsync(); asyncContext.setTimeout(TIME_OUT); asyncContext.addListener(new ServletAsyncListener()); exececutor.execute(() -> {
try {
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(); }
});
} }
|
业务集成使用示例
业务 Controller 继承 BaseController,将业务统一交给异步方法处理
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23
|
@RestController @RequestMapping("/message") public class MessageController extends BaseController {
@Autowired private MessageService messageService;
@GetMapping("/detail") public void detail(HttpServletRequest servletRequest) { handler(servletRequest, () -> messageService.detail()); } }
|
Service 接口和实现类与通常的写法没区别。
业务接口:
1 2 3
| public interface MessageService { Message detail(); }
|
业务接口实现:
1 2 3 4 5 6 7 8 9 10 11
| @Service public class MessageServiceImpl implements MessageService {
@Override public Message detail() { Message message = new Message(); message.setId(100L); message.setBody("Hello World!"); return message; } }
|