• 博客访问: 432514
  • 博文数量: 36
  • 博客积分: 330
  • 博客等级: 一等列兵
  • 技术积分: 3002
  • 用 户 组: 普通用户
  • 注册时间: 2012-09-05 22:38
个人简介

java开发工程师,专注于内核源码,算法,数据结构。 qq:630501400

文章分类
文章存档

2014年(2)

2013年(25)

2012年(9)

微信关注

IT168企业级官微



微信号:IT168qiye



系统架构师大会



微信号:SACC2013

订阅
热词专题
Tomcat 对异步servlet的处理 2013-07-17 12:02:57

分类: Java

     这个是以前看的,由于最近工作中用到异步servlet,看了一下tomcat在对servlet3.0AsyncContext的实现过程,总结一下,使用异步servlet主要原因就是因为,在service方法中业务逻辑如果碰到io操作时间比较长的操作,这样这个service方法就会长时间占用tomcat容器线程池中的线程,这样是不利于其他请求的处理的,当线程池中的线程处理任务时,任务由于长时间io操作,肯定会阻塞线程处理其他任务,引入异步servlet的目的就是将容器线程池和业务线程池分离开。在处理大io的业务操作的时候,把这个操作移动到业务线程池中进行,释放容器线程,使得容器线程处理其他任务,在业务逻辑执行完毕之后,然后在通知tomcat容器线程池来继续后面的操作,这个操作应该是把处理结果commit到客户端或者是dispatch到其他servlet上。


    上图是原始servlet和容器线程池的模型,下图是异步servlet,容器线程池和业务线程池的模型,从图中可以看出,原始模型在处理业务逻辑的过程中会一直占有容器线程池,而异步servlet模型,可以看出在业务线程池处理的过程中,有一段时间容器线程池中的那个线程是空闲的,这种设计大大提高了容器的处理请求的能力。

    异步servlet的开启在service中开启,对于一般请求,在service方法之后,都会commit response结果到客户端,但是在异步servlet中这个commit是没有意义的,因为输出还没产生,在业务线程池中还未处理完毕,这时需要把当前处理环境保存起来,以便业务线程池处理完毕后,再次找到这个处理环境继续处理。

    参考下代码,重点看下普通servlet和异步servlet处理分支的逻辑

  1. // 当有请求过来的时候,或者是异步servlet,业务线程处理完毕,通知容器线程处理后续的逻辑,都会触发这个process函数
  2. public SocketState process(SocketWrapper<S> socket,
  3.                 SocketStatus status) {
  4. //可以看出来如果是异步servlet,会从connections中取回之前执行servlet时的上下文环境
  5.             Processor<S> processor = connections.remove(socket.getSocket());

  6.             if (status == SocketStatus.DISCONNECT && processor == null) {
  7.                 //nothing more to be done endpoint requested a close
  8.                 //and there are no object associated with this connection
  9.                 return SocketState.CLOSED;
  10.             }

  11.             socket.setAsync(false);
  12. // 如果没有取到processor,就意味着不是异步的servlet
  13.             try {
  14.                 if (processor == null) {
  15.                     processor = recycledProcessors.poll();
  16.                 }
  17.                 if (processor == null) {
  18.                     processor = createProcessor();
  19.                 }

  20.                 initSsl(socket, processor);

  21.                 SocketState state = SocketState.CLOSED;
  22.                 do {
  23.                     if (status == SocketStatus.DISCONNECT &&
  24.                             !processor.isComet()) {
  25.                         // Do nothing here, just wait for it to get recycled
  26.                         // Don't do this for Comet we need to generate an end
  27.                         // event (see BZ 54022)
  28.                     } else if (processor.isAsync() ||
  29.                             state == SocketState.ASYNC_END) {
  30. //异步servlet的后续处理,这里做直接返回给客户端结果,或者请求其他servlet,这里为什么是判断isAsync和state呢,因为complete操作和dispatch操作,可以认为dispatch操作最后processor的状态还是异步,但是complete就不是了只是前面循环返回的state是ASYNC_END,所以这里要判断isAsync或者state,isAsync代表dispatch的后续操作,ASYNC_END代表complete的后续操作
  31.                         state = processor.asyncDispatch(status);
  32.                     } else if (processor.isComet()) { //tomcat 对comet请求的处理
  33.                         state = processor.event(status);
  34.                     } else if (processor.isUpgrade()) { //tomcat对 websocket的处理
  35.                         state = processor.upgradeDispatch();
  36.                     } else {
  37.                         state = processor.process(socket); //普通servlet的处理
  38.                     }
  39.         //这里post操作很重要,目的是改变异步servlet的状态机,这里来判断是否业务线程池中的任务是否处理完毕,如果处理完毕,则state返回 ASYNC_END,如果未处理完毕,则表示当前循环中,等不到业务线程池处理任务完毕了,这样当前占用的容器线程池就需要被释放了,等待异步servlet通知容器线程池重新处理servlet
  40.                     if (state != SocketState.CLOSED && processor.isAsync()) {
  41.                         state = processor.asyncPostProcess();
  42.                     }

  43.                     if (state == SocketState.UPGRADING) {
  44.                         // Get the UpgradeInbound handler
  45.                         UpgradeInbound inbound = processor.getUpgradeInbound();
  46.                         // Release the Http11 processor to be re-used
  47.                         release(socket, processor, false, false);
  48.                         // Create the light-weight upgrade processor
  49.                         processor = createUpgradeProcessor(socket, inbound);
  50.                         inbound.onUpgradeComplete();
  51.                     }
  52.                 } while (state == SocketState.ASYNC_END ||
  53.                         state == SocketState.UPGRADING);
  54. //刚才的post操作如果返回的状态是ASYNC_END,那么说明在当前容器线程处理任务之前,业务线程已经处理完任务了,那么就不需要释放当前容器线程池了,只需要再次在当前容器线程中继续处理servlet就行,否则表示容器线程不需要等待业务线程了,直接释放容器线程吧。
  55. //state是long表示当前处理的servlet是异步servlet,需要保存当前处理上下文,需要把这个processor保存起来,再业务线程处理完毕之后,再从connnections中取得这个processor,见代码的第一行。
  56.                 if (state == SocketState.LONG) {
  57.                     // In the middle of processing a request/response. Keep the
  58.                     // socket associated with the processor. Exact requirements
  59.                     // depend on type of long poll
  60.                     longPoll(socket, processor);
  61.                 } else if (state == SocketState.OPEN) { //open表示Http1.1的长连接
  62.                     // In keep-alive but between requests. OK to recycle
  63.                     // processor. Continue to poll for the next request.
  64.                     release(socket, processor, false, true);
  65.                 } else if (state == SocketState.SENDFILE) {
  66.                     // Sendfile in progress. If it fails, the socket will be
  67.                     // closed. If it works, the socket will be re-added to the
  68.                     // poller
  69.                     release(socket, processor, false, false);
  70.                 } else if (state == SocketState.UPGRADED) {
  71.                     // Need to keep the connection associated with the processor
  72.                     longPoll(socket, processor);
  73.                 } else {
  74.                     // Connection closed. OK to recycle the processor.
  75.                     if (!(processor instanceof UpgradeProcessor)) {
  76.                         release(socket, processor, true, false);
  77.                     }
  78.                 }
  79.                 return state;


    上面代码中很重要的一点就是每个processor,也就是每个servlet的处理过程,这个processor里面有个状态机,来记录异步servlet的过程。


  1. DISPATCHED:普通servlet结束的状态
  2. STARTING:servlet开始异步时的状态
  3. STARTED:当前servlet已经开始异步,释放容器线程之前异步servlet并未结束的状态
  4. MUST_COMPLETE:释放容器线程之前,异步servlet已经结束的状态(complete函数)
  5. COMPLETING:异步servlet并未dispatch到其他servlet上,然后异步结束的状态
  6. TIMING_OUT:当前异步servlet已经超时的状态
  7. MUST_DISPATCH:释放容器线程之前,异步servlet dispatch到其他servlet上的状态
  8. DISPATCHING:异步servlet结束,dispatch到其他servlet上的状态
  9. ERROR:异步servlet异常的状态


    异步servlet结束的操作有两种一种是在异步servlet中调用complete函数,另一种是调用dispatch函数,但是业务线程的结束时间是有说法的,如果在容器线程池释放线程之前业务线程就结束了,那么是不会释放容器线程的,状态机回出现 MUST_COMPLETEMUST_DISPATCH这两种状态,相反如果释放容器线程时,业务线程还没有处理完毕,那么异步servlet结束时就会出现 COMPLETINGDISPATCHING这两种状态,同时dispatch函数在业务线程池中只是在AsyncContext中设置需要dispatch到其他servlet,但是并不会直接dispatch,而是等到容器线程池,在后面处理的时候再去真正的dispatch

    状态迁移图来源于tomcat源码注解

对于dispatch操作,在业务线程中只是设置一下要dispatch,然后激发一下状态机。

代码在AsyncContextImpl

  1. public void dispatch(ServletContext context, String path) {
  2.         if (log.isDebugEnabled()) {
  3.             logDebug("dispatch ");
  4.         }
  5.         check();
  6.         if (request.getAttribute(ASYNC_REQUEST_URI)==null) {
  7.             request.setAttribute(ASYNC_REQUEST_URI, request.getRequestURI());
  8.             request.setAttribute(ASYNC_CONTEXT_PATH, request.getContextPath());
  9.             request.setAttribute(ASYNC_SERVLET_PATH, request.getServletPath());
  10.             request.setAttribute(ASYNC_PATH_INFO, request.getPathInfo());
  11.             request.setAttribute(ASYNC_QUERY_STRING, request.getQueryString());
  12.         }
  13.         final RequestDispatcher requestDispatcher = context.getRequestDispatcher(path);
  14.         if (!(requestDispatcher instanceof AsyncDispatcher)) {
  15.             throw new UnsupportedOperationException(
  16.                     sm.getString("asyncContextImpl.noAsyncDispatcher"));
  17.         }
  18.         final AsyncDispatcher applicationDispatcher =
  19.                 (AsyncDispatcher) requestDispatcher;
  20.         final HttpServletRequest servletRequest =
  21.                 (HttpServletRequest) getRequest();
  22.         final HttpServletResponse servletResponse =
  23.                 (HttpServletResponse) getResponse();
  24.         //这个runnable暂时不执行,后面会invoke到
  25.         Runnable run = new Runnable() {
  26.             @Override
  27.             public void run() {
  28.                 //状态迁移到DISPATCHED
  29.                 request.getCoyoteRequest().action(ActionCode.ASYNC_DISPATCHED, null);
  30.                 try {
  31.                     applicationDispatcher.dispatch(servletRequest, servletResponse);//真正dispatch的函数
  32.                 }catch (Exception x) {
  33.                     //log.error("Async.dispatch",x);
  34.                     throw new RuntimeException(x);
  35.                 }
  36.             }
  37.         };
  38.         
  39.         this.dispatch = run;
  40.         // 迁移到 MUST_DISPATCH或者 DISPATCHING状态
  41.         this.request.getCoyoteRequest().action(ActionCode.ASYNC_DISPATCH, null);
  42.     }

最后在standardWrapperValve中可以看到上面那个runnable被执行,这个是地方是普通servlet请求,容器最终调用service函数的地方。

  1. if (request.isAsyncDispatching()) { //是DISPATCHING状态
  2.                         //TODO SERVLET3 - async
  3.                         ((AsyncContextImpl)request.getAsyncContext()).doInternalDispatch(); //执行runnable
  4.                     } else if (comet) {
  5.                         request.setComet(true);
  6.                         filterChain.doFilterEvent(request.getEvent());
  7.                     } else {
  8.                         filterChain.doFilter
  9.                             (request.getRequest(), response.getResponse());
  10.                     }

参考代码:
  tomcat-7.0.40


阅读(12477) | 评论(0) | 转发(0) |
给主人留下些什么吧!~~
评论热议
请登录后评论。

登录 注册