tomcat的Http11Protocol，Http11Processor

  上文说到在JIoEndpoint类中处理请求最终是调用到内部接口Handler的process()方法，而Handler的实现类是Http11Protocol的内部类Http11ConnectionHandler，Http11ConnectionHandler又委托Http11Processor进行处理，tomcat在这个地方我觉得类的设计不是很优雅啊，内部类用的太多，导致阅读的时候有困难(好处是内部类持有外部类的的引用，不用使用显示的new)。

先把Http11ConnectionHandler的类附上:

    protected static class Http11ConnectionHandler implements Handler {

        protected Http11Protocol proto;
        protected AtomicLong registerCount = new AtomicLong(0);
        protected RequestGroupInfo global = new RequestGroupInfo();

        //无界线程安全队列，不会阻塞，自己加工的原因是processorCache的数量是有限的,为了线程和性能考虑
        protected ConcurrentLinkedQueue<Http11Processor> recycledProcessors = 
            new ConcurrentLinkedQueue<Http11Processor>() {
            protected AtomicInteger size = new AtomicInteger(0);
            public boolean offer(Http11Processor processor) {
                boolean offer = (proto.processorCache == -1) ? true : (size.get() < proto.processorCache);
                //avoid over growing our cache or add after we have stopped
                boolean result = false;
                if ( offer ) {
                    result = super.offer(processor);
                    if ( result ) {
                        size.incrementAndGet();
                    }
                }
                if (!result) unregister(processor);
                return result;
            }
            
            public Http11Processor poll() {
                Http11Processor result = super.poll();
                if ( result != null ) {
                    size.decrementAndGet();
                }
                return result;
            }
            
            public void clear() {
                Http11Processor next = poll();
                while ( next != null ) {
                    unregister(next);
                    next = poll();
                }
                super.clear();
                size.set(0);
            }
        };

        Http11ConnectionHandler(Http11Protocol proto) {
            this.proto = proto;
        }

        public boolean process(Socket socket) {
        	//processorCache里面没有processor
            Http11Processor processor = recycledProcessors.poll();
            try {

                if (processor == null) {
                    processor = createProcessor();
                }

                if (processor instanceof ActionHook) {
                    ((ActionHook) processor).action(ActionCode.ACTION_START, null);
                }

                if (proto.isSSLEnabled() && (proto.sslImplementation != null)) {
                    processor.setSSLSupport
                        (proto.sslImplementation.getSSLSupport(socket));
                } else {
                    processor.setSSLSupport(null);
                }
                
                processor.process(socket);
                return false;

            } catch(java.net.SocketException e) {
                // SocketExceptions are normal
                Http11Protocol.log.debug
                    (sm.getString
                     ("http11protocol.proto.socketexception.debug"), e);
            } catch (java.io.IOException e) {
                // IOExceptions are normal
                Http11Protocol.log.debug
                    (sm.getString
                     ("http11protocol.proto.ioexception.debug"), e);
            }
            // Future developers: if you discover any other
            // rare-but-nonfatal exceptions, catch them here, and log as
            // above.
            catch (Throwable e) {
                // any other exception or error is odd. Here we log it
                // with "ERROR" level, so it will show up even on
                // less-than-verbose logs.
                Http11Protocol.log.error
                    (sm.getString("http11protocol.proto.error"), e);
            } finally {
                //       if(proto.adapter != null) proto.adapter.recycle();
                //                processor.recycle();

                if (processor instanceof ActionHook) {
                    ((ActionHook) processor).action(ActionCode.ACTION_STOP, null);
                }
                recycledProcessors.offer(processor);
            }
            return false;
        }
}

  看process()方法前，我们看另外一个有意思的东西，ConcurrentLinkedQueue的子类，自己实现的原因是ConcurrentLinkedQueue是无界的，非阻塞的，而这个需要的是有界的，查了下jdk的concurrent包，没有现有的有界的，非阻塞的实现类，所有这里自己实现个，还有实现这个主要还和jmx的注册联系在一起，这里由于processorCache==-1，长度限制这个原因不存在。主要看process()这个方法

    首先，从ConcurrentLinkedQueue中取得一个Http11Processor，没有的话就创建一个。

    由于Http11Processor是ActionHook的实现，ActionHook的action方法被执行，这个以后再说，这里是   ActionCode.ACTION_START的事件的执行

    接着是SSL的一些的设置,在https的时候有用

    接着重点，调用Http11Processor的process的方法

    后面就是收拾残局了，什么错误啊，Http11Processor的回收了，ACTION_STOP的事件执行了

 

    转到Http11Processor的process()的方法:

 

   public void process(Socket theSocket)
        throws IOException {
        RequestInfo rp = request.getRequestProcessor();
        rp.setStage(org.apache.coyote.Constants.STAGE_PARSE);

        // Set the remote address
        remoteAddr = null;
        remoteHost = null;
        localAddr = null;
        localName = null;
        remotePort = -1;
        localPort = -1;

        // Setting up the I/O
        this.socket = theSocket;
        inputBuffer.setInputStream(socket.getInputStream());
        outputBuffer.setOutputStream(socket.getOutputStream());

        // Error flag
        error = false;
        keepAlive = true;

        int keepAliveLeft = maxKeepAliveRequests;
        int soTimeout = endpoint.getSoTimeout();

        // When using an executor, these values may return non-positive values
        int curThreads = endpoint.getCurrentThreadsBusy();
        int maxThreads = endpoint.getMaxThreads();
        if (curThreads > 0 && maxThreads > 0) {
            // Only auto-disable keep-alive if the current thread usage % can be
            // calculated correctly
            if ((curThreads*100)/maxThreads > 75) {
                keepAliveLeft = 1;
            }
        }
        
        try {
            socket.setSoTimeout(soTimeout);
        } catch (Throwable t) {
            log.debug(sm.getString("http11processor.socket.timeout"), t);
            error = true;
        }

        boolean keptAlive = false;

        //keepAlive:当处理完用户发起的 HTTP 请求后是否立即关闭 TCP 连接
        while (started && !error && keepAlive) {

            // Parsing the request header
            try {
                if (keptAlive) {
                    if (keepAliveTimeout > 0) {
                        socket.setSoTimeout(keepAliveTimeout);
                    }
                    else if (soTimeout > 0) {
                        socket.setSoTimeout(soTimeout);
                    }
                }
                
                //解析http header 的第一行数据
                inputBuffer.parseRequestLine();
                request.setStartTime(System.currentTimeMillis());
                keptAlive = true;
                if (disableUploadTimeout) {
                    socket.setSoTimeout(soTimeout);
                } else {
                    socket.setSoTimeout(timeout);
                }
              //解析http header 的除第一行数据以外的数据
                inputBuffer.parseHeaders();
            } catch (IOException e) {
                error = true;
                break;
            } catch (Throwable t) {
                if (log.isDebugEnabled()) {
                    log.debug(sm.getString("http11processor.header.parse"), t);
                }
                // 400 - Bad Request
                response.setStatus(400);
                error = true;
            }

            if (!error) {
                // Setting up filters, and parse some request headers
                rp.setStage(org.apache.coyote.Constants.STAGE_PREPARE);
                try {
                    prepareRequest();
                } catch (Throwable t) {
                    if (log.isDebugEnabled()) {
                        log.debug(sm.getString("http11processor.request.prepare"), t);
                    }
                    // 400 - Internal Server Error
                    response.setStatus(400);
                    error = true;
                }
            }

            if (maxKeepAliveRequests > 0 && --keepAliveLeft == 0)
                keepAlive = false;

            // Process the request in the adapter
            if (!error) {
                try {
                    rp.setStage(org.apache.coyote.Constants.STAGE_SERVICE);
                    adapter.service(request, response);
                    // Handle when the response was committed before a serious
                    // error occurred.  Throwing a ServletException should both
                    // set the status to 500 and set the errorException.
                    // If we fail here, then the response is likely already
                    // committed, so we can't try and set headers.
                    if(keepAlive && !error) { // Avoid checking twice.
                        error = response.getErrorException() != null ||
                                statusDropsConnection(response.getStatus());
                    }

                } catch (InterruptedIOException e) {
                    error = true;
                } catch (Throwable t) {
                    log.error(sm.getString("http11processor.request.process"), t);
                    // 500 - Internal Server Error
                    response.setStatus(500);
                    error = true;
                }
            }

            // Finish the handling of the request
            try {
                rp.setStage(org.apache.coyote.Constants.STAGE_ENDINPUT);
                // If we know we are closing the connection, don't drain input.
                // This way uploading a 100GB file doesn't tie up the thread 
                // if the servlet has rejected it.
                if(error)
                    inputBuffer.setSwallowInput(false);
                inputBuffer.endRequest();
            } catch (IOException e) {
                error = true;
            } catch (Throwable t) {
                log.error(sm.getString("http11processor.request.finish"), t);
                // 500 - Internal Server Error
                response.setStatus(500);
                error = true;
            }
            try {
                rp.setStage(org.apache.coyote.Constants.STAGE_ENDOUTPUT);
                outputBuffer.endRequest();
            } catch (IOException e) {
                error = true;
            } catch (Throwable t) {
                log.error(sm.getString("http11processor.response.finish"), t);
                error = true;
            }

            // If there was an error, make sure the request is counted as
            // and error, and update the statistics counter
            if (error) {
                response.setStatus(500);
            }
            request.updateCounters();

            rp.setStage(org.apache.coyote.Constants.STAGE_KEEPALIVE);

            // Don't reset the param - we'll see it as ended. Next request
            // will reset it
            // thrA.setParam(null);
            // Next request
            inputBuffer.nextRequest();
            outputBuffer.nextRequest();

        }

        rp.setStage(org.apache.coyote.Constants.STAGE_ENDED);

        // Recycle
        inputBuffer.recycle();
        outputBuffer.recycle();
        this.socket = null;
        // Recycle ssl info
        sslSupport = null;
    }

    Http11Processor很夸张的又引用到JIoEndpoint（感觉很乱，是否有循环有用的感觉？）

    这里有一个很有意思的变量keepAliveLeft，一开始的时候等于maxKeepAliveRequests，也就是创建Processor的时候传入的Http11Protocol的maxKeepAliveRequests，默认为100，当JIoEndpoint的curThreads(当前已经用的线程数)/maxThreads>0.75,maxKeepAliveRequests马上可怜的变成1，这里可以认为是过载保护。

   当开始标志，没有错误标志等设置后，就开始真正解析这个传入的socket了.

   在这之前实例化InternalInputBuffer，InternalOutputBuffer两个类，对应socket的输入和输出，是可以重复使用的高性能的类。

   首先调用InternalInputBuffer的parseRequestLine()方法解析http的请求行,即'请求方法 请求uri http版本'那一行

  接着解析请求头InternalInputBuffer的parseHeaders()方法干这个活

 

    public void parseHeaders()
        throws IOException {

        while (parseHeader()) {
        }

        parsingHeader = false;
        end = pos;

    }

   while循环解析每一行，请求头有个专门的类MimeHeaders存放，里面一般存放的是MessageBytes，简单的说一行就是MessageBytes，这个类有char，byte，String等表示，好处是很方便，任何三类都可以处理，对于涉及编码的来说

，char和byte可以非常方便的进行原生态的表示

   接着是调用prepareRequest()方法，其实在parseHeaders()中进行了http headers的解析，这里只是拿出来用而已,只是把org.apache.coyote.Request的属性设置好而已

 

    Http11Processor的process方法就到这里

    回到Http11ConnectionHandler的process()方法，最后面有代码:

        finally {
                //       if(proto.adapter != null) proto.adapter.recycle();
                //                processor.recycle();

                if (processor instanceof ActionHook) {
                    ((ActionHook) processor).action(ActionCode.ACTION_STOP, null);
                }
                recycledProcessors.offer(processor);
            }

    这里调用了Http11Processor的action的方法，我们看下这个方法：

 

    	
        if (actionCode == ActionCode.ACTION_COMMIT) {
        	
            // Commit current response  只有在response中触发这个分支
        	
        	
        	//一般来说现在response.isCommitted()是false的，但tomcat中很多地方都有这样的判断,主要是为了稳定性考虑(不存在多线程问题?)
            if (response.isCommitted())
                return;

            // Validate and write response headers
            prepareResponse();
            try {
                outputBuffer.commit();
            } catch (IOException e) {
                // Set error flag
                error = true;
            }

        } 

 

    第一个分支就是，注意  prepareResponse(); 这里是准备org.apache.coyote.Response方法了，这个和request的处理差不多，是设置http协议响应头的一些内容

 

  准备好的org.apache.coyote.Request和org.apache.coyote.Response在CoyoteAdapter里面会被用到，返回给tomcat的Connector组件，这个下回再说。

 

 

  哇，真长！