详解Tomcat是如何实现异步Servlet的

类：AbstractProcessorLight
public SocketState process(SocketWrapperBase<&＃63;> socketWrapper, SocketEvent status)
   throws IOException {
  //省略部分diam
  SocketState state = SocketState.CLOSED;
  Iterator dispatches = null;
  do {
   if (dispatches != null) {
    DispatchType nextDispatch = dispatches.next();
    state = dispatch(nextDispatch.getSocketStatus());
   } else if (status == SocketEvent.DISCONNECT) {
   
   } else if (isAsync() || isUpgrade() || state == SocketState.ASYNC_END) {
    state = dispatch(status);
    if (state == SocketState.OPEN) {
     state = service(socketWrapper);
    }
   } else if (status == SocketEvent.OPEN_WRITE) {
    state = SocketState.LONG;
   } else if (status == SocketEvent.OPEN_READ){
    state = service(socketWrapper);
   } else {
    state = SocketState.CLOSED;
   }

  } while (state == SocketState.ASYNC_END ||
    dispatches != null && state != SocketState.CLOSED);

  return state;
 }

public boolean asyncDispatch(org.apache.coyote.Request req, org.apache.coyote.Response res,
   SocketEvent status) throws Exception {
  //省略部分代码
  Request request = (Request) req.getNote(ADAPTER_NOTES);
  Response respOnse= (Response) res.getNote(ADAPTER_NOTES);
  boolean success = true;
  AsyncContextImpl asyncCOnImpl= request.getAsyncContextInternal();
  try {
   if (!request.isAsync()) {
    response.setSuspended(false);
   }

   if (status==SocketEvent.TIMEOUT) {
    if (!asyncConImpl.timeout()) {
     asyncConImpl.setErrorState(null, false);
    }
   } else if (status==SocketEvent.ERROR) {
    
   }

   if (!request.isAsyncDispatching() && request.isAsync()) {
    WriteListener writeListener = res.getWriteListener();
    ReadListener readListener = req.getReadListener();
    if (writeListener != null && status == SocketEvent.OPEN_WRITE) {
     ClassLoader oldCL = null;
     try {
      oldCL = request.getContext().bind(false, null);
      res.onWritePossible();//这里执行浏览器响应，写入数据
      if (request.isFinished() && req.sendAllDataReadEvent() &&
        readListener != null) {
       readListener.onAllDataRead();
      }
     } catch (Throwable t) {
      
     } finally {
      request.getContext().unbind(false, oldCL);
     }
    } 
    }
   }
   //这里判断异步正在进行，说明这不是一个完成方法的回调，是一个正常异步请求，继续调用容器。
   if (request.isAsyncDispatching()) {
    connector.getService().getContainer().getPipeline().getFirst().invoke(
      request, response);
    Throwable t = (Throwable) request.getAttribute(RequestDispatcher.ERROR_EXCEPTION);
    if (t != null) {
     asyncConImpl.setErrorState(t, true);
    }
   }
   //注意，这里，如果超时或者出错，request.isAsync()会返回false，这里是为了尽快的输出错误给客户端。
   if (!request.isAsync()) {
    //这里也是输出逻辑
    request.finishRequest();
    response.finishResponse();
   }
   //销毁request和response
   if (!success || !request.isAsync()) {
    updateWrapperErrorCount(request, response);
    request.recycle();
    response.recycle();
   }
  }
  return success;
 }

 postParseSuccess = postParseRequest(req, request, res, response);
   //省略部分代码
   if (postParseSuccess) {
    request.setAsyncSupported(
      connector.getService().getContainer().getPipeline().isAsyncSupported());
    connector.getService().getContainer().getPipeline().getFirst().invoke(
      request, response);
   }
   if (request.isAsync()) {
    async = true;
    } else {
    //输出数据到客户端
    request.finishRequest();
    response.finishResponse();
   if (!async) {
    updateWrapperErrorCount(request, response);
    //销毁request和response
    request.recycle();
    response.recycle();
   }

@RestController
@Slf4j
public class TestController {
 @Autowired
 private TestService service;

 @GetMapping("/hello")
 public String test() {
  try {
   log.info("testAsynch Start");
   CompletableFuture test1 = service.test1();
   CompletableFuture test2 = service.test2();
   CompletableFuture test3 = service.test3();
   CompletableFuture.allOf(test1, test2, test3);
   log.info("test1=====" + test1.get());
   log.info("test2=====" + test2.get());
   log.info("test3=====" + test3.get());
  } catch (InterruptedException e) {
   e.printStackTrace();
  } catch (ExecutionException e) {
   e.printStackTrace();
  }
  return "hello";
 }
@Service
public class TestService {
 @Async("asyncExecutor")
 public CompletableFuture test1() throws InterruptedException {
  Thread.sleep(3000L);
  return CompletableFuture.completedFuture("test1");
 }

 @Async("asyncExecutor")
 public CompletableFuture test2() throws InterruptedException {
  Thread.sleep(3000L);
  return CompletableFuture.completedFuture("test2");
 }

 @Async("asyncExecutor")
 public CompletableFuture test3() throws InterruptedException {
  Thread.sleep(3000L);
  return CompletableFuture.completedFuture("test3");
 }
}
@SpringBootApplication
@EnableAsync
public class TomcatdebugApplication {

 public static void main(String[] args) {
  SpringApplication.run(TomcatdebugApplication.class, args);
 }

 @Bean(name = "asyncExecutor")
 public Executor asyncExecutor() {
  ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
  executor.setCorePoolSize(3);
  executor.setMaxPoolSize(3);
  executor.setQueueCapacity(100);
  executor.setThreadNamePrefix("AsynchThread-");
  executor.initialize();
  return executor;
 }