快捷導航

詳解Tomcat是如何實現(xiàn)異步Servlet的

更新時間：2019年10月15日 09:20:21 作者：木木匠

這篇文章主要介紹了詳解Tomcat是如何實現(xiàn)異步Servlet的，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧

前言

通過我之前的Tomcat系列文章，相信看我博客的同學對Tomcat應(yīng)該有一個比較清晰的了解了，在前幾篇博客我們討論了Tomcat在SpringBoot框架中是如何啟動的，討論了Tomcat的內(nèi)部組件是如何設(shè)計以及請求是如何流轉(zhuǎn)的，那么我們這邊博客聊聊Tomcat的異步Servlet，Tomcat是如何實現(xiàn)異步Servlet的以及異步Servlet的使用場景。

手擼一個異步的Servlet

我們直接借助SpringBoot框架來實現(xiàn)一個Servlet,這里只展示Servlet代碼：

@WebServlet(urlPatterns = "/async",asyncSupported = true)
@Slf4j
public class AsyncServlet extends HttpServlet {

 ExecutorService executorService =Executors.newSingleThreadExecutor();

 @Override
  protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
  //開啟異步,獲取異步上下文
  final AsyncContext ctx = req.startAsync();
  // 提交線程池異步執(zhí)行
  executorService.execute(new Runnable() {


   @Override
   public void run() {
    try {
     log.info("async Service 準備執(zhí)行了");
     //模擬耗時任務(wù)
     Thread.sleep(10000L);
     ctx.getResponse().getWriter().print("async servlet");
     log.info("async Service 執(zhí)行了");
    } catch (IOException e) {
     e.printStackTrace();
    } catch (InterruptedException e) {
     e.printStackTrace();
    }
    //最后執(zhí)行完成后完成回調(diào)。
    ctx.complete();
   }
  });
 }

上面的代碼實現(xiàn)了一個異步的Servlet,實現(xiàn)了 doGet 方法注意在SpringBoot中使用需要再啟動類加上 @ServletComponentScan 注解來掃描Servlet。既然代碼寫好了，我們來看看實際運行效果。

我們發(fā)送一個請求后，看到頁面有響應(yīng)，同時，看到請求時間花費了10.05s,那么我們這個Servlet算是能正常運行啦。有同學肯定會問，這不是異步servlet嗎？你的響應(yīng)時間并沒有加快，有什么用呢？對，我們的響應(yīng)時間并不能加快，還是會取決于我們的業(yè)務(wù)邏輯，但是我們的異步servlet請求后，依賴于業(yè)務(wù)的異步執(zhí)行，我們可以立即返回，也就是說，Tomcat的線程可以立即回收，默認情況下，Tomcat的核心線程是10，最大線程數(shù)是200,我們能及時回收線程，也就意味著我們能處理更多的請求，能夠增加我們的吞吐量，這也是異步Servlet的主要作用。

異步Servlet的內(nèi)部原理

了解完異步Servlet的作用后，我們來看看，Tomcat是如何是先異步Servlet的。其實上面的代碼，主要核心邏輯就兩部分， final AsyncContext ctx = req.startAsync(); 和 ctx.complete(); 那我們來看看他們究竟做了什么？

 public AsyncContext startAsync(ServletRequest request,
   ServletResponse response) {
  if (!isAsyncSupported()) {
   IllegalStateException ise =
     new IllegalStateException(sm.getString("request.asyncNotSupported"));
   log.warn(sm.getString("coyoteRequest.noAsync",
     StringUtils.join(getNonAsyncClassNames())), ise);
   throw ise;
  }

  if (asyncContext == null) {
   asyncContext = new AsyncContextImpl(this);
  }

  asyncContext.setStarted(getContext(), request, response,
    request==getRequest() && response==getResponse().getResponse());
  asyncContext.setTimeout(getConnector().getAsyncTimeout());

  return asyncContext;
 }

我們發(fā)現(xiàn) req.startAsync(); 只是保存了一個異步上下文，同時設(shè)置一些基礎(chǔ)信息，比如 Timeout ,順便提一下，這里設(shè)置的默認超時時間是30S，也就是說你的異步處理邏輯超過30S后就會報錯，這個時候執(zhí)行 ctx.complete(); 就會拋出IllegalStateException 異常。

我們來看看 ctx.complete(); 的邏輯

 public void complete() {
  if (log.isDebugEnabled()) {
   logDebug("complete ");
  }
  check();
  request.getCoyoteRequest().action(ActionCode.ASYNC_COMPLETE, null);
 }
//類：AbstractProcessor 
 public final void action(ActionCode actionCode, Object param) {
 case ASYNC_COMPLETE: {
   clearDispatches();
   if (asyncStateMachine.asyncComplete()) {
    processSocketEvent(SocketEvent.OPEN_READ, true);
   }
   break;
  } 
 }
 //類：AbstractProcessor 
protected void processSocketEvent(SocketEvent event, boolean dispatch) {
  SocketWrapperBase<?> socketWrapper = getSocketWrapper();
  if (socketWrapper != null) {
   socketWrapper.processSocket(event, dispatch);
  }
 }
 //類：AbstractEndpoint
public boolean processSocket(SocketWrapperBase<S> socketWrapper,
   SocketEvent event, boolean dispatch) {
  //省略部分代碼
   SocketProcessorBase<S> sc = null;
   if (processorCache != null) {
    sc = processorCache.pop();
   }
   if (sc == null) {
    sc = createSocketProcessor(socketWrapper, event);
   } else {
    sc.reset(socketWrapper, event);
   }
   Executor executor = getExecutor();
   if (dispatch && executor != null) {
    executor.execute(sc);
   } else {
    sc.run();
   }
 
  return true;
 }

所以，這里最終會調(diào)用 AbstractEndpoint 的 processSocket 方法，之前看過我前面博客的同學應(yīng)該有印象， EndPoint 是用來接受和處理請求的，接下來就會交給 Processor 去進行協(xié)議處理。

類：AbstractProcessorLight
public SocketState process(SocketWrapperBase<?> socketWrapper, SocketEvent status)
   throws IOException {
  //省略部分diam
  SocketState state = SocketState.CLOSED;
  Iterator<DispatchType> 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;
 }

這部分是重點， AbstractProcessorLight 會根據(jù) SocketEvent 的狀態(tài)來判斷是不是要去調(diào)用 service(socketWrapper) ,該方法最終會去調(diào)用到容器，從而完成業(yè)務(wù)邏輯的調(diào)用，我們這個請求是執(zhí)行完成后調(diào)用的，肯定不能進容器了，不然就是死循環(huán)了，這里通過 isAsync() 判斷，就會進入 dispatch(status) ,最終會調(diào)用 CoyoteAdapter 的 asyncDispatch 方法

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();//這里執(zhí)行瀏覽器響應(yīng)，寫入數(shù)據(jù)
      if (request.isFinished() && req.sendAllDataReadEvent() &&
        readListener != null) {
       readListener.onAllDataRead();
      }
     } catch (Throwable t) {
      
     } finally {
      request.getContext().unbind(false, oldCL);
     }
    } 
    }
   }
   //這里判斷異步正在進行，說明這不是一個完成方法的回調(diào)，是一個正常異步請求，繼續(xù)調(diào)用容器。
   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;
 }

上面的代碼就是 ctx.complete() 執(zhí)行最終的方法了（當然省略了很多細節(jié)），完成了數(shù)據(jù)的輸出，最終輸出到瀏覽器。

這里有同學可能會說，我知道異步執(zhí)行完后，調(diào)用 ctx.complete() 會輸出到瀏覽器，但是，第一次doGet請求執(zhí)行完成后，Tomcat是怎么知道不用返回到客戶端的呢？關(guān)鍵代碼在 CoyoteAdapter 中的 service 方法，部分代碼如下：

 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 {
    //輸出數(shù)據(jù)到客戶端
    request.finishRequest();
    response.finishResponse();
   if (!async) {
    updateWrapperErrorCount(request, response);
    //銷毀request和response
    request.recycle();
    response.recycle();
   }

這部分代碼在調(diào)用完 Servlet 后，會通過 request.isAsync() 來判斷是否是異步請求，如果是異步請求，就設(shè)置 async = true 。如果是非異步請求就執(zhí)行輸出數(shù)據(jù)到客戶端邏輯，同時銷毀 request 和 response 。這里就完成了請求結(jié)束后不響應(yīng)客戶端的操作。

為什么說Spring Boot的@EnableAsync注解不是異步Servlet

因為之前準備寫本篇文章的時候就查詢過很多資料，發(fā)現(xiàn)很多資料寫SpringBoot異步編程都是依賴于 @EnableAsync 注解，然后在 Controller 用多線程來完成業(yè)務(wù)邏輯，最后匯總結(jié)果，完成返回輸出。這里拿一個掘金大佬的文章來舉例《新手也能看懂的 SpringBoot 異步編程指南》，這篇文章寫得很通俗易懂，非常不錯，從業(yè)務(wù)層面來說，確實是異步編程，但是有一個問題，拋開業(yè)務(wù)的并行處理來說，針對整個請求來說，并不是異步的，也就是說不能立即釋放Tomcat的線程，從而不能達到異步Servlet的效果。這里我參考上文也寫了一個demo，我們來驗證下，為什么它不是異步的。

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

 @GetMapping("/hello")
 public String test() {
  try {
   log.info("testAsynch Start");
   CompletableFuture<String> test1 = service.test1();
   CompletableFuture<String> test2 = service.test2();
   CompletableFuture<String> 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<String> test1() throws InterruptedException {
  Thread.sleep(3000L);
  return CompletableFuture.completedFuture("test1");
 }

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

 @Async("asyncExecutor")
 public CompletableFuture<String> 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;
 }

這里我運行下，看看效果

這里我請求之后，在調(diào)用容器執(zhí)行業(yè)務(wù)邏輯之前打了一個斷點，然后在返回之后的同樣打了一個斷點，在 Controller 執(zhí)行完之后，請求才回到了 CoyoteAdapter 中，并且判斷 request.isAsync() ,根據(jù)圖中看到，是為 false ,那么接下來就會執(zhí)行 request.finishRequest() 和 response.finishResponse() 來執(zhí)行響應(yīng)的結(jié)束，并銷毀請求和響應(yīng)體。很有趣的事情是，我實驗的時候發(fā)現(xiàn)，在執(zhí)行 request.isAsync() 之前，瀏覽器的頁面上已經(jīng)出現(xiàn)了響應(yīng)體，這是SpringBoot框架已經(jīng)通過 StringHttpMessageConverter 類中的 writeInternal 方法已經(jīng)進行輸出了。

以上分析的核心邏輯就是，Tomcat的線程執(zhí)行 CoyoteAdapter 調(diào)用容器后，必須要等到請求返回，然后再判斷是否是異步請求，再處理請求，然后執(zhí)行完畢后，線程才能進行回收。而我一最開始的異步Servlet例子，執(zhí)行完doGet方法后，就會立即返回，也就是會直接到 request.isAsync() 的邏輯，然后整個線程的邏輯執(zhí)行完畢，線程被回收。

聊聊異步Servlet的使用場景

分析了這么多，那么異步Servlet的使用場景有哪些呢？其實我們只要抓住一點就可以分析了，就是異步Servlet提高了系統(tǒng)的吞吐量，可以接受更多的請求。假設(shè)web系統(tǒng)中Tomcat的線程不夠用了，大量請求在等待，而此時Web系統(tǒng)應(yīng)用層面的優(yōu)化已經(jīng)不能再優(yōu)化了，也就是無法縮短業(yè)務(wù)邏輯的響應(yīng)時間了，這個時候，如果想讓減少用戶的等待時間，提高吞吐量，可以嘗試下使用異步Servlet。

舉一個實際的例子：比如做一個短信系統(tǒng)，短信系統(tǒng)對實時性要求很高，所以要求等待時間盡可能短，而發(fā)送功能我們實際上是委托運營商去發(fā)送的，也就是說我們要調(diào)用接口，假設(shè)并發(fā)量很高，那么這個時候業(yè)務(wù)系統(tǒng)調(diào)用我們的發(fā)送短信功能，就有可能把我們的Tomcat線程池用完，剩下的請求就會在隊列中等待，那這個時候，短信的延時就上去了，為了解決這個問題，我們可以引入異步Servlet,接受更多的短信發(fā)送請求，從而減少短信的延時。

總結(jié)

這篇文章我從手寫一個異步Servlet來開始，分析了異步Servlet的作用，以及Tomcat內(nèi)部是如何實現(xiàn)異步Servlet的，然后我也根據(jù)互聯(lián)網(wǎng)上流行的SpringBoot異步編程來進行說明，其在Tomcat內(nèi)部并不是一個異步的Servlet。最后，我談到了異步Servlet的使用場景，分析了什么情況下可以嘗試異步Servlet。

以上就是本文的全部內(nèi)容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: