Tomcat源码分析(四)-Connector

布满荆棘的人生 2022-05-14 07:46 182阅读 0赞

### **一、Connector内部分析** ###

![70][]

*Connector的内部结构*

Connector的作用:底层使用Socket连接，接收请求并把请求封装成Request和Response。

(Connector实现了Http协议和TCP/IP协议，Request和Response内部是由Http协议封装)

(Connector在Catalina调用load方法创建，生命周期方法在Service中调用。)

封装完后------》交给Container处理------》返回给Connector------》Connector使用Socket将结果返回给客户端。

ProtocolHandler  包含三个组件-------》1、Endpoint(AbstractEndpoint)  处理底层Socket的网络连接 ----用来实现TCP/IP协议

Acceptor 用于监听请求，AsyncTimeout用于检查异步request的超时,Handler用于处理接收到的Socket.

-------》2、Processor  将Endpoint接收到的Endpoint封装成Request  ----实现HTTP协议

-------》3、Adapter   将Request封装成交给Container处理。 ----请求转到Servlet容器处理

Connector的构造方法：

org.apache.catalina.connector.connector类
    
       public Connector() {
            this(null);
        }
    
        public Connector(String protocol) {
            setProtocol(protocol);
            // Instantiate protocol handler
            ProtocolHandler p = null;
            try {
                Class<?> clazz = Class.forName(protocolHandlerClassName);
                                                       //创建ProtocolHandler对象实例 
                p = (ProtocolHandler) clazz.getDeclaredConstructor().newInstance();
            } catch (Exception e) {
                log.error(sm.getString(
                        "coyoteConnector.protocolHandlerInstantiationFailed"), e);
            } finally {
                this.protocolHandler = p;
            }
    
            if (!Globals.STRICT_SERVLET_COMPLIANCE) {
                URIEncoding = "UTF-8";
                URIEncodingLower = URIEncoding.toLowerCase(Locale.ENGLISH);
            }
        }
    
       
            setProtocol(protocol);方法
           /**
         * Set the Coyote protocol which will be used by the connector.
         *
         * @param protocol The Coyote protocol name
         */
        public void setProtocol(String protocol) {
             //根据协议类型 设置protocolHandlerClassName
            if (AprLifecycleListener.isAprAvailable()) {
                if ("HTTP/1.1".equals(protocol)) {
                    setProtocolHandlerClassName
                        ("org.apache.coyote.http11.Http11AprProtocol");
                } else if ("AJP/1.3".equals(protocol)) {
                    setProtocolHandlerClassName
                        ("org.apache.coyote.ajp.AjpAprProtocol");
                } else if (protocol != null) {
                    setProtocolHandlerClassName(protocol);
                } else {
                    setProtocolHandlerClassName
                        ("org.apache.coyote.http11.Http11AprProtocol");
                }
            } else {
                if ("HTTP/1.1".equals(protocol)) {
                    setProtocolHandlerClassName
                        ("org.apache.coyote.http11.Http11NioProtocol");
                } else if ("AJP/1.3".equals(protocol)) {
                    setProtocolHandlerClassName
                        ("org.apache.coyote.ajp.AjpNioProtocol");
                } else if (protocol != null) {
                    setProtocolHandlerClassName(protocol);
                }
            }
    
        }
    
    
          /**
         * Coyote Protocol handler class name.
         * Defaults to the Coyote HTTP/1.1 protocolHandler.
         */
        protected String protocolHandlerClassName =
            "org.apache.coyote.http11.Http11NioProtocol";
    
           /**
         * Set the class name of the Coyote protocol handler which will be used
         * by the connector.
         *
         * @param protocolHandlerClassName The new class name
         */
        public void setProtocolHandlerClassName(String protocolHandlerClassName) {
    
            this.protocolHandlerClassName = protocolHandlerClassName;
    
        }

Connector（通过构造方法）创建对象实例的时候同时创建了ProtocolHandler对象实例。

用 setProtocol(protocol); 设置好的protocolHandlerClassName的类(默认是org.apache.coyote.http11.Http11NioProtocol)

然后用 Class<?> clazz = Class.forName(protocolHandlerClassName);创建ProtocolHandler对象

上面代码中的 Apr是 Apache Portable Runtime的缩写 -->Apache可移植运行库

主要为上层的应用程序提供一个可以跨越多操作系统平台使用的底层支持接口库。在早期 的Apache版本中，应用程序本身必须能够处理各种具体操作系统平台的细节，并针对不同的平台调用不同的处理函数。

目前APR主要还是由Apache使用，不过由于APR的较好的移植性，因此一些需要进行移植的C程序也开始使用APR

如果使用Apr需要先安装,安装后Tomcat可以检测的出来。

AprLifecycleListener.isAprAvailable() 判断Apr是否可用。

** 1.1 Connector的生命周期方法:**

org.apache.catalina.connector.connector类:
     @Override
        protected void initInternal() throws LifecycleException {
    
            super.initInternal();
    
            // Initialize adapter
            adapter = new CoyoteAdapter(this);
            protocolHandler.setAdapter(adapter);
    
            // Make sure parseBodyMethodsSet has a default
            if( null == parseBodyMethodsSet ) {
                setParseBodyMethods(getParseBodyMethods());
            }
    
            if (protocolHandler.isAprRequired() &&
                    !AprLifecycleListener.isAprAvailable()) {
                throw new LifecycleException(
                        sm.getString("coyoteConnector.protocolHandlerNoApr",
                                getProtocolHandlerClassName()));
            }
    
            try {
                protocolHandler.init();
            } catch (Exception e) {
                throw new LifecycleException
                    (sm.getString
                     ("coyoteConnector.protocolHandlerInitializationFailed"), e);
            }
        }
    
    
        /**
         * Begin processing requests via this Connector.
         *
         * @exception LifecycleException if a fatal startup error occurs
         */
        @Override
        protected void startInternal() throws LifecycleException {
    
            // Validate settings before starting
            if (getPort() < 0) {
                throw new LifecycleException(sm.getString(
                        "coyoteConnector.invalidPort", Integer.valueOf(getPort())));
            }
    
            setState(LifecycleState.STARTING);
    
            try {
                protocolHandler.start();
            } catch (Exception e) {
                String errPrefix = "";
                if(this.service != null) {
                    errPrefix += "service.getName(): \"" + this.service.getName() + "\"; ";
                }
    
                throw new LifecycleException
                    (errPrefix + " " + sm.getString
                     ("coyoteConnector.protocolHandlerStartFailed"), e);
            }
        }
    
    
      @Override
        protected void destroyInternal() throws LifecycleException {
            try {
                protocolHandler.destroy();
            } catch (Exception e) {
                throw new LifecycleException
                    (sm.getString
                     ("coyoteConnector.protocolHandlerDestroyFailed"), e);
            }
    
            if (getService() != null) {
                getService().removeConnector(this);      //service移除Connector
            }
    
            super.destroyInternal();
        }

**   1.2 ProtocolHandler类:**

![70 1][]

ProtocolHandler下面有三种协议类型：AJP,Http,Spdy。

AJP协议-----》(英文全称Apache JServ Protocol),Apache的定向包协议，是长连接，用于客服端(前端服务器)建立                                                  连接 通信。

Spdy协议------》应用不广泛，SPDY（读作“SPeeDY”）是[Google][]开发的基于[TCP][]的应用                                                                                        层协议，用以最小化网络延迟，提升网络速度，优化用户的网络使用体验。SPDY并不是一种用于                                                  替代HTTP的协议，而是对[HTTP][]协议的增强.

Http11NioProtocol的四个周期方法都在父类AbstractProtocol类实现--》 代码如下:

org.apache.coyote.AbstractProtocol类:
       /*
         * NOTE: There is no maintenance of state or checking for valid transitions
         * within this class. It is expected that the connector will maintain state
         * and prevent invalid state transitions.
         */
    
        @Override
        public void init() throws Exception {
            if (getLog().isInfoEnabled())
                getLog().info(sm.getString("abstractProtocolHandler.init",
                        getName()));
    
            if (oname == null) {
                // Component not pre-registered so register it
                oname = createObjectName();
                if (oname != null) {
                    Registry.getRegistry(null, null).registerComponent(this, oname,
                        null);
                }
            }
    
            if (this.domain != null) {
                try {
                    tpOname = new ObjectName(domain + ":" +
                            "type=ThreadPool,name=" + getName());
                    Registry.getRegistry(null, null).registerComponent(endpoint,
                            tpOname, null);
                } catch (Exception e) {
                    getLog().error(sm.getString(
                            "abstractProtocolHandler.mbeanRegistrationFailed",
                            tpOname, getName()), e);
                }
                rgOname=new ObjectName(domain +
                        ":type=GlobalRequestProcessor,name=" + getName());
                Registry.getRegistry(null, null).registerComponent(
                        getHandler().getGlobal(), rgOname, null );
            }
    
            String endpointName = getName();
            endpoint.setName(endpointName.substring(1, endpointName.length()-1));
    
            try {
                endpoint.init();
            } catch (Exception ex) {
                getLog().error(sm.getString("abstractProtocolHandler.initError",
                        getName()), ex);
                throw ex;
            }
        }
    
    
        @Override
        public void start() throws Exception {
            if (getLog().isInfoEnabled())
                getLog().info(sm.getString("abstractProtocolHandler.start",
                        getName()));
            try {
                endpoint.start();
            } catch (Exception ex) {
                getLog().error(sm.getString("abstractProtocolHandler.startError",
                        getName()), ex);
                throw ex;
            }
        }

** **protocolHandlerClassName**默认配置 --》org.apache.coyote.http11.Http11NioProtocol类：**

/**
     * Abstract the protocol implementation, including threading, etc.
     * Processor is single threaded and specific to stream-based protocols,
     * will not fit Jk protocols like JNI.
     *
     * @author Remy Maucherat
     * @author Costin Manolache
     */
    public class Http11NioProtocol extends AbstractHttp11JsseProtocol<NioChannel> {
    
        private static final Log log = LogFactory.getLog(Http11NioProtocol.class);
    
    
        @Override
        protected Log getLog() { return log; }
    
    
        @Override
        protected AbstractEndpoint.Handler getHandler() {
            return cHandler;
        }
    
     
        /**// Constants类的
         * // public static final int DEFAULT_CONNECTION_LINGER = -1;
         * // public static final int DEFAULT_CONNECTION_TIMEOUT = 60000;
         * // public static final boolean DEFAULT_TCP_NO_DELAY = true;
         **/
        public Http11NioProtocol() {
            endpoint=new NioEndpoint();                         //创建Endpoint
            cHandler = new Http11ConnectionHandler(this);      //创建Http11ConnectionHandler
            ((NioEndpoint) endpoint).setHandler(cHandler);
                 //父类AbstractProtocol的endpoint.setSoLinger(soLinger);              
            setSoLinger(Constants.DEFAULT_CONNECTION_LINGER);
                 //父类AbstractProtocol的endpoint.setSoTimeout(timeout);
            setSoTimeout(Constants.DEFAULT_CONNECTION_TIMEOUT);
                  //父类AbstractProtocol的endpoint.setTcpNoDelay(tcpNoDelay);
            setTcpNoDelay(Constants.DEFAULT_TCP_NO_DELAY);
        }
    
    
        public NioEndpoint getEndpoint() {
            return ((NioEndpoint)endpoint);
        }
    
    。。。省略部分代码
    }

**1.3 AbstractEndpoint类(endpoint):**

**       **![70 2][]

AbstractEndpoint类 分析：

org.apache.tomcat.util.net.AbstractEndpoint
     
      //1、init方法
     public final void init() throws Exception {
            testServerCipherSuitesOrderSupport();
            if (bindOnInit) {
                bind();
                bindState = BindState.BOUND_ON_INIT;
            }
        }
    
        protected void testServerCipherSuitesOrderSupport() {
            // Only test this feature if the user explicitly requested its use.
            if(!"".equals(getUseServerCipherSuitesOrder().trim()) && !JreCompat.isJre8Available()) {
                throw new UnsupportedOperationException(
                        sm.getString("endpoint.jsse.cannotHonorServerCipherOrder"));
            }
        }
    
           //2、start方法
         public final void start() throws Exception {
            if (bindState == BindState.UNBOUND) {
                bind();
                bindState = BindState.BOUND_ON_START;
            }
            startInternal();
        }
    
    
        //bind()具体实现方法在子类里 如:--》org.apache.tomcat.util.net.NioEndpoint
          /**
         * Initialize the endpoint.
         */
        @Override
        public void bind() throws Exception {
    
            serverSock = ServerSocketChannel.open();
            socketProperties.setProperties(serverSock.socket());
            InetSocketAddress addr = (getAddress()!=null?new InetSocketAddress(getAddress(),getPort()):new InetSocketAddress(getPort()));
            serverSock.socket().bind(addr,getBacklog());
            serverSock.configureBlocking(true); //mimic APR behavior
            serverSock.socket().setSoTimeout(getSocketProperties().getSoTimeout());
    
            // Initialize thread count defaults for acceptor, poller
            if (acceptorThreadCount == 0) {
                // FIXME: Doesn't seem to work that well with multiple accept threads
                acceptorThreadCount = 1;
            }
            if (pollerThreadCount <= 0) {
                //minimum one poller thread
                pollerThreadCount = 1;
            }
            stopLatch = new CountDownLatch(pollerThreadCount);
    
            // Initialize SSL if needed
            if (isSSLEnabled()) {
                SSLUtil sslUtil = handler.getSslImplementation().getSSLUtil(this);
    
                sslContext = sslUtil.createSSLContext();
                sslContext.init(wrap(sslUtil.getKeyManagers()),
                        sslUtil.getTrustManagers(), null);
    
                SSLSessionContext sessionContext =
                    sslContext.getServerSessionContext();
                if (sessionContext != null) {
                    sslUtil.configureSessionContext(sessionContext);
                }
                // Determine which cipher suites and protocols to enable
                enabledCiphers = sslUtil.getEnableableCiphers(sslContext);
                enabledProtocols = sslUtil.getEnableableProtocols(sslContext);
            }
    
            if (oomParachute>0) reclaimParachute(true);
            selectorPool.open();
        }
    
    
        //startInternal()具体实现方法在子类里--》org.apache.tomcat.util.net.NioEndpoint
          /**
         * Start the NIO endpoint, creating acceptor, poller threads.
         */
        @Override
        public void startInternal() throws Exception {
    
            if (!running) {
                running = true;
                paused = false;
    
                processorCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
                        socketProperties.getProcessorCache());
                eventCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
                                socketProperties.getEventCache());
                nioChannels = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
                        socketProperties.getBufferPool());
    
                // Create worker collection   //创建Executor
                if ( getExecutor() == null ) {
                    createExecutor();
                }
    
                initializeConnectionLatch();
    
                // Start poller threads      //启动poller线程
                pollers = new Poller[getPollerThreadCount()];
                for (int i=0; i<pollers.length; i++) {
                    pollers[i] = new Poller();
                    Thread pollerThread = new Thread(pollers[i], getName() + "-ClientPoller-"+i);
                    pollerThread.setPriority(threadPriority);
                    pollerThread.setDaemon(true);
                    pollerThread.start();
                }
    
                startAcceptorThreads();  //启动Acceptor线程
            }
        }
    
      
       org.apache.tomcat.util.net.AbstractEndpoint类:
    
       protected final void startAcceptorThreads() {
            int count = getAcceptorThreadCount();
            acceptors = new Acceptor[count];
    
            for (int i = 0; i < count; i++) {
                acceptors[i] = createAcceptor();
                String threadName = getName() + "-Acceptor-" + i;
                acceptors[i].setThreadName(threadName);
                Thread t = new Thread(acceptors[i], threadName);  //acceptors线程在子类实现
                t.setPriority(getAcceptorThreadPriority());
                t.setDaemon(getDaemon());
                t.start();
            }
        }

**1.3.1 selectorPool.open();**

在 org.apache.tomcat.util.net.NioEndpoint类具体实现了bind()方法，而在bind()方法里面有 selectorPool.open();

private NioSelectorPool selectorPool = new NioSelectorPool();

selectorPool.open();具体实现：

org.apache.tomcat.util.net.NioSelectorPool类: --> Thread safe non blocking selector pool
    
    
       public void open() throws IOException {
            enabled = true;
            getSharedSelector();
            if (SHARED) {
                blockingSelector = new NioBlockingSelector();
                blockingSelector.open(getSharedSelector());
            }
    
        }
    
       //getSharedSelector方法:
        protected volatile Selector SHARED_SELECTOR;
        protected static final boolean SHARED =   
     Boolean.parseBoolean(System.getProperty("org.apache.tomcat.util.net.NioSelectorShared", "true"));
    
       protected Selector getSharedSelector() throws IOException {
            if (SHARED && SHARED_SELECTOR == null) {
                synchronized ( NioSelectorPool.class ) {
                    if ( SHARED_SELECTOR == null )  {
                        synchronized (Selector.class) {
                            // Selector.open() isn't thread safe
                            // http://bugs.sun.com/view_bug.do?bug_id=6427854
                            // Affects 1.6.0_29, fixed in 1.7.0_01
                            SHARED_SELECTOR = Selector.open();  //打开selector
                        }
                        log.info("Using a shared selector for servlet write/read");
                    }
                }
            }
            return  SHARED_SELECTOR;
        }
    
        
    
         java.nio.channels.selector类
        /**
         * Opens a selector.
         *
         * <p> The new selector is created by invoking the {@link
         * java.nio.channels.spi.SelectorProvider#openSelector openSelector} method
         * of the system-wide default {@link
         * java.nio.channels.spi.SelectorProvider} object.  </p>
         *
         * @return  A new selector
         *
         * @throws  IOException
         *          If an I/O error occurs
         */
        public static Selector open() throws IOException {
            return SelectorProvider.provider().openSelector();
        }

![70 3][]

上面的org.apache.tomcat.util.net.NioSelectorPool类: 的open方法:

blockingSelector.open(getSharedSelector());方法实现如下:

--》org.apache.tomcat.util.net.NioBlockingSelector类:
    
        protected Selector sharedSelector;
    
        protected BlockPoller poller;
    
     public void open(Selector selector) {
            sharedSelector = selector;
            poller = new BlockPoller();
            poller.selector = sharedSelector;
            poller.setDaemon(true);
            poller.setName("NioBlockingSelector.BlockPoller-"+(++threadCounter));
            poller.start();   //启动poller
        }
    
    
    
     --》org.apache.tomcat.util.net.NioBlockingSelector类 下的内部类BlockPoller :
    
      protected static class BlockPoller extends Thread {
            protected volatile boolean run = true;
            protected Selector selector = null;
            protected final SynchronizedQueue<Runnable> events = new SynchronizedQueue<>();
            public void disable() { run = false; selector.wakeup();}
            protected final AtomicInteger wakeupCounter = new AtomicInteger(0);
    
            public void cancelKey(final SelectionKey key) {
                Runnable r = new RunnableCancel(key);
                events.offer(r);
                wakeup();
            }
    
            ...省略部分代码
    
            @Override
            public void run() {
                while (run) {
                    try {
                        events();
                        int keyCount = 0;
                        try {
                            int i = wakeupCounter.get();
                            if (i>0)
                                keyCount = selector.selectNow();
                            else {
                                wakeupCounter.set(-1);
                                keyCount = selector.select(1000);
                            }
                            wakeupCounter.set(0);
                            if (!run) break;
                        }catch ( NullPointerException x ) {
                            //sun bug 5076772 on windows JDK 1.5
                            if (selector==null) throw x;
                            if ( log.isDebugEnabled() ) log.debug("Possibly encountered sun bug 5076772 on windows JDK 1.5",x);
                            continue;
                        } catch ( CancelledKeyException x ) {
                            //sun bug 5076772 on windows JDK 1.5
                            if ( log.isDebugEnabled() ) log.debug("Possibly encountered sun bug 5076772 on windows JDK 1.5",x);
                            continue;
                        } catch (Throwable x) {
                            ExceptionUtils.handleThrowable(x);
                            log.error("",x);
                            continue;
                        }
    
                        Iterator<SelectionKey> iterator = keyCount > 0 ? selector.selectedKeys().iterator() : null;
    
                        // Walk through the collection of ready keys and dispatch
                        // any active event.
                        while (run && iterator != null && iterator.hasNext()) {
                            SelectionKey sk = iterator.next();
                            KeyAttachment attachment = (KeyAttachment)sk.attachment();
                            try {
                                attachment.access();
                                iterator.remove();
                                sk.interestOps(sk.interestOps() & (~sk.readyOps()));
                                if ( sk.isReadable() ) {
                                    countDown(attachment.getReadLatch());
                                }
                                if (sk.isWritable()) {
                                    countDown(attachment.getWriteLatch());
                                }
                            }catch (CancelledKeyException ckx) {
                                sk.cancel();
                                countDown(attachment.getReadLatch());
                                countDown(attachment.getWriteLatch());
                            }
                        }//while
                    }catch ( Throwable t ) {
                        log.error("",t);
                    }
                }
                events.clear();
                // If using a shared selector, the NioSelectorPool will also try and
                // close the selector. Try and avoid the ClosedSelectorException
                // although because multiple threads are involved there is always
                // the possibility of an Exception here.
                if (selector.isOpen()) {
                    try {
                        // Cancels all remaining keys
                        selector.selectNow();
                    }catch( Exception ignore ) {
                        if (log.isDebugEnabled())log.debug("",ignore);
                    }
                }
                try {
                    selector.close();
                }catch( Exception ignore ) {
                    if (log.isDebugEnabled())log.debug("",ignore);
                }
            }
    
            public void countDown(CountDownLatch latch) {
                if ( latch == null ) return;
                latch.countDown();
            }
    
    
            ...省略部分代码
        }

NioSelectorPool中维护了一个BlockPoller线程，这个线程就是基于Selector进行NIO的逻辑。以执行servlet后，得到response，往socket中写数据为例，最终写的过程调用NioBlockingSelector的write方法。

**1.3.2 reclaimParachute(true)**

在 org.apache.tomcat.util.net.NioEndpoint类具体实现了bind()方法，而在bind()方法里面有

if (oomParachute>0) reclaimParachute(true);
            selectorPool.open();

reclaimParachute(true) 方法实现如下:

org.apache.tomcat.util.net.NioEndpoint类：
    
    
        /**
         * The size of the OOM parachute.
         */
        private int oomParachute = 1024*1024;
        /**
         * The oom parachute, when an OOM error happens,
         * will release the data, giving the JVM instantly
         * a chunk of data to be able to recover with.
         */
        private byte[] oomParachuteData = null;
    
    
     //oomParachuteData当发生OOM错误时，oom Parachute将释放数据，使JVM立即获得一大块数据以便能
     //够恢复。
     //在Acceptor类 catch (OutOfMemoryError oom) 的时候，也就是内存溢出的时候会清空
     //  oomParachuteData数据
     //  force为true,oomParachute大于0，那么new 一个oomParachuteData实例
     //Runtime.getRuntime().freeMemory() 判断了 运行时的剩余内存是否足够
        protected boolean reclaimParachute(boolean force) {
            if ( oomParachuteData != null ) return true;
            if ( oomParachute > 0 && ( force || (Runtime.getRuntime().freeMemory() > 
          (oomParachute*2))) )
                oomParachuteData = new byte[oomParachute];
            return oomParachuteData != null;
        }

oomParachute : 官网--》

The NIO connector implements an OutOfMemoryError strategy called parachute.It holds a chunk of data as a byte array. In case of an OOM, this chunk of data is released and the error is reported. This will give the VM enough room to clean up. The oomParachute represents the size in bytes of the parachute(the byte array). The default value is 1024\*1024(1MB). Please note, this only works for OOM errors regarding the Java Heap space, and there is absolutely no guarantee that you will be able to recover at all. If you have an OOM outside of the Java Heap, then this parachute trick will not help

NIO connector 实现了一个OutOfMemoryError策略 叫做parachute。它拥有一个块的数据作为一个字节数组。如果  OOM错误 出现内存溢出异常，这个数据块被释放，并报告错误。这会给VM足够的空间来清理。oomParachute代表parachute（字节数组）的大小（以字节为单位）。默认值是 1024 \* 1024（1MB）。请注意，这仅适用于关于Java堆空间的OOM错误，也不是绝对保证，你将能够恢复所有。如果你有一个OOM 在Java堆之外，那么这个parachute也不会起作用。

OOM ---->指的是 OutOfMemory

----》也就是初始化 bind()的时候 初始化ServerSocketChannel的时候 (ServerSocketChannel用来为Acceptor接收请求                        做好准备),使用每个NIO通道在启动时都会根据参数包含这样一个parachute块，默认大小是1M的byte数组，发现                           OOM产生时，该数组会被设为null，同时清理一些组件的缓存 。

**1.4 Acceptor和Poller分析**

在org.apache.tomcat.util.net.NioEndpoint 类里面 startInternal方法 启动poller线程和Acceptor来处理请求。

1、NioEndpoint 的内部类**Poller**

org.apache.tomcat.util.net.NioEndpoint类:
    
     /**
         * Poller class.
         */
        public class Poller implements Runnable {
    
            private Selector selector;
            private final SynchronizedQueue<PollerEvent> events =
                    new SynchronizedQueue<>();
    
            private volatile boolean close = false;
            private long nextExpiration = 0;//optimize expiration handling
    
            private AtomicLong wakeupCounter = new AtomicLong(0);
    
            private volatile int keyCount = 0;
    
            public Poller() throws IOException {
                synchronized (Selector.class) {
                    // Selector.open() isn't thread safe
                    // http://bugs.sun.com/view_bug.do?bug_id=6427854
                    // Affects 1.6.0_29, fixed in 1.7.0_01
                    this.selector = Selector.open();
                }
            }
    
    。。。。。省略部分代码
    
             /**
             * The background thread that listens for incoming TCP/IP connections and
             * hands them off to an appropriate processor.
             */
            @Override
            public void run() {
                // Loop until destroy() is called
                while (true) {
                    try {
                        // Loop if endpoint is paused
                        while (paused && (!close) ) {
                            try {
                                Thread.sleep(100);
                            } catch (InterruptedException e) {
                                // Ignore
                            }
                        }
    
                        boolean hasEvents = false;
    
                        // Time to terminate?
                        if (close) {
                            events();
                            timeout(0, false);
                            try {
                                selector.close();
                            } catch (IOException ioe) {
                                log.error(sm.getString(
                                        "endpoint.nio.selectorCloseFail"), ioe);
                            }
                            break;
                        } else {
                            hasEvents = events();
                        }
                        try {
                            if ( !close ) {
                                if (wakeupCounter.getAndSet(-1) > 0) {
                                    //if we are here, means we have other stuff to do
                                    //do a non blocking select
                                    keyCount = selector.selectNow();
                                } else {
                                    keyCount = selector.select(selectorTimeout);
                                }
                                wakeupCounter.set(0);
                            }
                            if (close) {
                                events();
                                timeout(0, false);
                                try {
                                    selector.close();
                                } catch (IOException ioe) {
                                    log.error(sm.getString(
                                            "endpoint.nio.selectorCloseFail"), ioe);
                                }
                                break;
                            }
                        } catch (Throwable x) {
                            ExceptionUtils.handleThrowable(x);
                            log.error("",x);
                            continue;
                        }
                        //either we timed out or we woke up, process events first
                        if ( keyCount == 0 ) hasEvents = (hasEvents | events());
    
                        Iterator<SelectionKey> iterator =
                            keyCount > 0 ? selector.selectedKeys().iterator() : null;
                        // Walk through the collection of ready keys and dispatch
                        // any active event.              
                                       
                        while (iterator != null && iterator.hasNext()) {
                            SelectionKey sk = iterator.next();
                            KeyAttachment attachment = (KeyAttachment)sk.attachment();
                            // Attachment may be null if another thread has called
                            // cancelledKey()
                            if (attachment == null) {
                                iterator.remove();
                            } else {
                                attachment.access();
                                iterator.remove();
                                processKey(sk, attachment);
                            }
                        }//while
    
                        //process timeouts
                        timeout(keyCount,hasEvents);
                        if ( oomParachute > 0 && oomParachuteData == null ) checkParachute();
                    } catch (OutOfMemoryError oom) {
                        try {
                            oomParachuteData = null;
                            releaseCaches();
                            log.error("", oom);
                        }catch ( Throwable oomt ) {
                            try {
                                System.err.println(oomParachuteMsg);
                                oomt.printStackTrace();
                            }catch (Throwable letsHopeWeDontGetHere){
                                ExceptionUtils.handleThrowable(letsHopeWeDontGetHere);
                            }
                        }
                    }
                }//while
    
                stopLatch.countDown();
            }
    
    。。。。。。省略部分代码。。
          }

上面代码 通过 selector.selectedKeys().iterator() 接收到请求数据，

/\*\*  
         \* The background thread that listens for incoming TCP/IP connections and  
         \* hands them off to an appropriate processor.  
         \*/  
        @Override  
        public void run() \{\}

主要是 后台线程 监听TCP/IP连接 并且处理他们交给一个processor

上面代码 里面有一个 processKey(sk, attachment);方法 代码如下：

org.apache.tomcat.util.net.NioEndpoint类:
    
     protected boolean processKey(SelectionKey sk, KeyAttachment attachment) {
                boolean result = true;
                try {
                    if ( close ) {
                        cancelledKey(sk, SocketStatus.STOP);
                    } else if ( sk.isValid() && attachment != null ) {
                        attachment.access();//make sure we don't time out valid sockets
                        if (sk.isReadable() || sk.isWritable() ) {
                            if ( attachment.getSendfileData() != null ) {
                                processSendfile(sk,attachment, false);
                            } else {
                                if ( isWorkerAvailable() ) {
                                    unreg(sk, attachment, sk.readyOps());
                                    boolean closeSocket = false;
                                    // Read goes before write
                                    if (sk.isReadable()) {      //运行processSocket
                                        if (!processSocket(attachment, SocketStatus.OPEN_READ, true)) {
                                            closeSocket = true;
                                        }
                                    }
                                    if (!closeSocket && sk.isWritable()) {
                                        if (!processSocket(attachment, SocketStatus.OPEN_WRITE, true)) {
                                            closeSocket = true;
                                        }
                                    }
                                    if (closeSocket) {
                                        cancelledKey(sk,SocketStatus.DISCONNECT);
                                    }
                                } else {
                                    result = false;
                                }
                            }
                        }
                    } else {
                        //invalid key
                        cancelledKey(sk, SocketStatus.ERROR);
                    }
                } catch ( CancelledKeyException ckx ) {
                    cancelledKey(sk, SocketStatus.ERROR);
                } catch (Throwable t) {
                    ExceptionUtils.handleThrowable(t);
                    log.error("",t);
                }
                return result;
            }

上面代码   把具体处理交给了processSocket()方法：

processSocket方法代码如下：

org.apache.tomcat.util.net.NioEndpoint类:
    
      protected boolean processSocket(KeyAttachment attachment, SocketStatus status, boolean dispatch) {
            try {
                if (attachment == null) {
                    return false;
                }
                SocketProcessor sc = processorCache.pop();
                if ( sc == null ) sc = new SocketProcessor(attachment, status);
                else sc.reset(attachment, status);
                Executor executor = getExecutor();
                if (dispatch && executor != null) {
                    executor.execute(sc);
                } else {
                    sc.run();
                }
            } catch (RejectedExecutionException ree) {
                log.warn(sm.getString("endpoint.executor.fail", attachment.getSocket()), ree);
                return false;
            } catch (Throwable t) {
                ExceptionUtils.handleThrowable(t);
                // This means we got an OOM or similar creating a thread, or that
                // the pool and its queue are full
                log.error(sm.getString("endpoint.process.fail"), t);
                return false;
            }
            return true;
        }

上面代码运行  sc.run();  代码如下--》

org.apache.tomcat.util.net.NioEndpoint类:
     @Override
            public void run() {
                NioChannel socket = ka.getSocket();
    
                // Upgraded connections need to allow multiple threads to access the
                // connection at the same time to enable blocking IO to be used when
                // NIO has been configured
                if (ka.isUpgraded() && SocketStatus.OPEN_WRITE == status) {
                    synchronized (ka.getWriteThreadLock()) {
                        doRun();
                    }
                } else {
                    synchronized (socket) {
                        doRun();
                    }
                }
            }
    
      private void doRun() {
                NioChannel socket = ka.getSocket();
                SelectionKey key = socket.getIOChannel().keyFor(socket.getPoller().getSelector());
    
                try {
                    int handshake = -1;
    
                    try {
                        if (key != null) {
                            // For STOP there is no point trying to handshake as the
                            // Poller has been stopped.
                            if (socket.isHandshakeComplete() ||
                                    status == SocketStatus.STOP) {
                                handshake = 0;
                            } else {
                                handshake = socket.handshake(
                                        key.isReadable(), key.isWritable());
                                // The handshake process reads/writes from/to the
                                // socket. status may therefore be OPEN_WRITE once
                                // the handshake completes. However, the handshake
                                // happens when the socket is opened so the status
                                // must always be OPEN_READ after it completes. It
                                // is OK to always set this as it is only used if
                                // the handshake completes.
                                status = SocketStatus.OPEN_READ;
                            }
                        }
                    } catch (IOException x) {
                        handshake = -1;
                        if (log.isDebugEnabled()) log.debug("Error during SSL handshake",x);
                    } catch (CancelledKeyException ckx) {
                        handshake = -1;
                    }
                    if (handshake == 0) {
                        SocketState state = SocketState.OPEN;
                        // Process the request from this socket
                        if (status == null) {
                            state = handler.process(ka, SocketStatus.OPEN_READ);
                        } else {
                            state = handler.process(ka, status);
                        }
                        if (state == SocketState.CLOSED) {
                            close(socket, key, SocketStatus.ERROR);
                        }
                    } else if (handshake == -1 ) {
                        close(socket, key, SocketStatus.DISCONNECT);
                    } else {
                        ka.getPoller().add(socket,handshake);
                    }
                } catch (CancelledKeyException cx) {
                    socket.getPoller().cancelledKey(key, null);
                } catch (OutOfMemoryError oom) {
                    try {
                        oomParachuteData = null;
                        log.error("", oom);
                        socket.getPoller().cancelledKey(key,SocketStatus.ERROR);
                        releaseCaches();
                    } catch (Throwable oomt) {
                        try {
                            System.err.println(oomParachuteMsg);
                            oomt.printStackTrace();
                        } catch (Throwable letsHopeWeDontGetHere){
                            ExceptionUtils.handleThrowable(letsHopeWeDontGetHere);
                        }
                    }
                } catch (VirtualMachineError vme) {
                    ExceptionUtils.handleThrowable(vme);
                } catch (Throwable t) {
                    log.error("", t);
                    socket.getPoller().cancelledKey(key,SocketStatus.ERROR);
                } finally {
                    ka = null;
                    status = null;
                    //return to cache
                    if (running && !paused) {
                        processorCache.push(this);
                    }
                }
            }

// For STOP there is no point trying to handshake as the  
                        // Poller has been stopped.  
                        if (socket.isHandshakeComplete() ||  
                                status == SocketStatus.STOP) \{  
                            handshake = 0;  
                        \}

如果socket握手完成或者Poller已经stop,那么就会到 handler.process(）方法，也就是交给Handler来处理。

而Handler-->

![70 4][]

Http11ConnectionHandler类没有实现Handler的process()方法，交给了它的父类AbstractConnectionHandler

// --------------------  Connection handler --------------------

protected static class Http11ConnectionHandler  
            extends AbstractConnectionHandler<NioChannel,Http11NioProcessor>  
            implements Handler \{\}

org.apache.coyote.AbstractProtocol类
    
    
    
        // ------------------------------------------- Connection handler base class
    
        protected abstract static class AbstractConnectionHandler<S,P extends Processor<S>>
                implements AbstractEndpoint.Handler {
    
    。。。省略部分代码。。
    
    public SocketState process(SocketWrapper<S> wrapper,
                    SocketStatus status) {
                if (wrapper == null) {
                    // Nothing to do. Socket has been closed.
                    return SocketState.CLOSED;
                }
    
                S socket = wrapper.getSocket();
                if (socket == null) {
                    // Nothing to do. Socket has been closed.
                    return SocketState.CLOSED;
                }
    
                Processor<S> processor = connections.get(socket);
                if (status == SocketStatus.DISCONNECT && processor == null) {
                    // Nothing to do. Endpoint requested a close and there is no
                    // longer a processor associated with this socket.
                    return SocketState.CLOSED;
                }
    
                wrapper.setAsync(false);
                ContainerThreadMarker.set();
    
                try {
                    if (processor == null) {
                        processor = recycledProcessors.pop();
                    }
                    if (processor == null) {
                        processor = createProcessor();
                    }
    
                    initSsl(wrapper, processor);
    
                    SocketState state = SocketState.CLOSED;
                    Iterator<DispatchType> dispatches = null;
                    do {
                        if (dispatches != null) {
                            // Associate the processor with the connection as
                            // these calls may result in a nested call to process()
                            connections.put(socket, processor);
                            DispatchType nextDispatch = dispatches.next();
                            if (processor.isUpgrade()) {
                                state = processor.upgradeDispatch(
                                        nextDispatch.getSocketStatus());
                            } else {
                                state = processor.asyncDispatch(
                                        nextDispatch.getSocketStatus());
                            }
                        } else if (processor.isComet()) {
                            state = processor.event(status);
                        } else if (processor.isUpgrade()) {
                            state = processor.upgradeDispatch(status);
                        } else if (status == SocketStatus.DISCONNECT) {
                            // Comet and upgrade need to see DISCONNECT but the
                            // others don't. NO-OP and let socket close.
                        } else if (processor.isAsync() || state == SocketState.ASYNC_END) {
                            state = processor.asyncDispatch(status);
                            if (state == SocketState.OPEN) {
                                // release() won't get called so in case this request
                                // takes a long time to process, remove the socket from
                                // the waiting requests now else the async timeout will
                                // fire
                                getProtocol().endpoint.removeWaitingRequest(wrapper);
                                // There may be pipe-lined data to read. If the data
                                // isn't processed now, execution will exit this
                                // loop and call release() which will recycle the
                                // processor (and input buffer) deleting any
                                // pipe-lined data. To avoid this, process it now.
                                state = processor.process(wrapper);
                            }
                        } else if (status == SocketStatus.OPEN_WRITE) {
                            // Extra write event likely after async, ignore
                            state = SocketState.LONG;
                        } else {
                            state = processor.process(wrapper);
                        }
    
                        if (state != SocketState.CLOSED && processor.isAsync()) {
                            state = processor.asyncPostProcess();
                        }
    
                        if (state == SocketState.UPGRADING) {
                            // Get the HTTP upgrade handler
                            UpgradeToken upgradeToken = processor.getUpgradeToken();
                            HttpUpgradeHandler httpUpgradeHandler = upgradeToken.getHttpUpgradeHandler();
                            // Retrieve leftover input
                            ByteBuffer leftoverInput = processor.getLeftoverInput();
                            // Release the Http11 processor to be re-used
                            release(wrapper, processor, false, false);
                            // Create the upgrade processor
                            processor = createUpgradeProcessor(
                                    wrapper, leftoverInput, upgradeToken);
                            // Mark the connection as upgraded
                            wrapper.setUpgraded(true);
                            // Associate with the processor with the connection
                            connections.put(socket, processor);
                            // Initialise the upgrade handler (which may trigger
                            // some IO using the new protocol which is why the lines
                            // above are necessary)
                            // This cast should be safe. If it fails the error
                            // handling for the surrounding try/catch will deal with
                            // it.
                            if (upgradeToken.getInstanceManager() == null) {
                                httpUpgradeHandler.init((WebConnection) processor);
                            } else {
                                ClassLoader oldCL = upgradeToken.getContextBind().bind(false, null);
                                try {
                                    httpUpgradeHandler.init((WebConnection) processor);
                                } finally {
                                    upgradeToken.getContextBind().unbind(false, oldCL);
                                }
                            }
                        }
                        if (getLog().isDebugEnabled()) {
                            getLog().debug("Socket: [" + wrapper +
                                    "], Status in: [" + status +
                                    "], State out: [" + state + "]");
                        }
                        if (dispatches == null || !dispatches.hasNext()) {
                            // Only returns non-null iterator if there are
                            // dispatches to process.
                            dispatches = wrapper.getIteratorAndClearDispatches();
                        }
                    } while (state == SocketState.ASYNC_END ||
                            state == SocketState.UPGRADING ||
                            dispatches != null && state != SocketState.CLOSED);
    
                    if (state == SocketState.LONG) {
                        // In the middle of processing a request/response. Keep the
                        // socket associated with the processor. Exact requirements
                        // depend on type of long poll
                        connections.put(socket, processor);
                        longPoll(wrapper, processor);
                    } else if (state == SocketState.OPEN) {
                        // In keep-alive but between requests. OK to recycle
                        // processor. Continue to poll for the next request.
                        connections.remove(socket);
                        release(wrapper, processor, false, true);
                    } else if (state == SocketState.SENDFILE) {
                        // Sendfile in progress. If it fails, the socket will be
                        // closed. If it works, the socket either be added to the
                        // poller (or equivalent) to await more data or processed
                        // if there are any pipe-lined requests remaining.
                        connections.put(socket, processor);
                    } else if (state == SocketState.UPGRADED) {
                        // Don't add sockets back to the poller if this was a
                        // non-blocking write otherwise the poller may trigger
                        // multiple read events which may lead to thread starvation
                        // in the connector. The write() method will add this socket
                        // to the poller if necessary.
                        if (status != SocketStatus.OPEN_WRITE) {
                            longPoll(wrapper, processor);
                        }
                    } else {
                        // Connection closed. OK to recycle the processor. Upgrade
                        // processors are not recycled.
                        connections.remove(socket);
                        if (processor.isUpgrade()) {
                            UpgradeToken upgradeToken = processor.getUpgradeToken();
                            HttpUpgradeHandler httpUpgradeHandler = upgradeToken.getHttpUpgradeHandler();
                            InstanceManager instanceManager = upgradeToken.getInstanceManager();
                            if (instanceManager == null) {
                                httpUpgradeHandler.destroy();
                            } else {
                                ClassLoader oldCL = upgradeToken.getContextBind().bind(false, null);
                                try {
                                    httpUpgradeHandler.destroy();
                                } finally {
                                    try {
                                        instanceManager.destroyInstance(httpUpgradeHandler);
                                    } catch (Throwable e) {
                                        ExceptionUtils.handleThrowable(e);
                                        getLog().error(sm.getString("abstractConnectionHandler.error"), e);
                                    }
                                    upgradeToken.getContextBind().unbind(false, oldCL);
                                }
                            }
                        } else {
                            release(wrapper, processor, true, false);
                        }
                    }
                    return state;
                } catch(java.net.SocketException e) {
                    // SocketExceptions are normal
                    getLog().debug(sm.getString(
                            "abstractConnectionHandler.socketexception.debug"), e);
                } catch (java.io.IOException e) {
                    // IOExceptions are normal
                    getLog().debug(sm.getString(
                            "abstractConnectionHandler.ioexception.debug"), e);
                }
                // Future developers: if you discover any other
                // rare-but-nonfatal exceptions, catch them here, and log as
                // above.
                catch (Throwable e) {
                    ExceptionUtils.handleThrowable(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.
                    getLog().error(
                            sm.getString("abstractConnectionHandler.error"), e);
                } finally {
                    ContainerThreadMarker.clear();
                }
    
                // Make sure socket/processor is removed from the list of current
                // connections
                connections.remove(socket);
                // Don't try to add upgrade processors back into the pool
                if (processor !=null && !processor.isUpgrade()) {
                    release(wrapper, processor, true, false);
                }
                return SocketState.CLOSED;
            }
    
    。。。。。省略部分代码。。
    }

然后交给 Processor 处理   Processor<S> processor = connections.get(socket);

如： state = processor.process(wrapper); 代码如下：

![70 5][]

org.apache.coyote.http11.AbstractHttp11Processor类：
    
    /**
         * Process pipelined HTTP requests using the specified input and output
         * streams.
         *
         * @param socketWrapper Socket from which the HTTP requests will be read
         *               and the HTTP responses will be written.
         *
         * @throws IOException error during an I/O operation
         */
        @Override
        public SocketState process(SocketWrapper<S> socketWrapper)
            throws IOException {
            RequestInfo rp = request.getRequestProcessor();
            rp.setStage(org.apache.coyote.Constants.STAGE_PARSE);
    
            // Setting up the I/O
            setSocketWrapper(socketWrapper);
            getInputBuffer().init(socketWrapper, endpoint);
            getOutputBuffer().init(socketWrapper, endpoint);
    
            // Flags
            keepAlive = true;
            comet = false;
            openSocket = false;
            sendfileInProgress = false;
            readComplete = true;
            if (endpoint.getUsePolling()) {
                keptAlive = false;
            } else {
                keptAlive = socketWrapper.isKeptAlive();
            }
    
            if (disableKeepAlive()) {
                socketWrapper.setKeepAliveLeft(0);
            }
    
            while (!getErrorState().isError() && keepAlive && !comet && !isAsync() &&
                    upgradeToken == null && !endpoint.isPaused()) {
    
                // Parsing the request header
                try {
                    setRequestLineReadTimeout();
    
                    if (!getInputBuffer().parseRequestLine(keptAlive)) {
                        if (handleIncompleteRequestLineRead()) {
                            break;
                        }
                    }
    
                    if (endpoint.isPaused()) {
                        // 503 - Service unavailable
                        response.setStatus(503);
                        setErrorState(ErrorState.CLOSE_CLEAN, null);
                    } else {
                        keptAlive = true;
                        // Set this every time in case limit has been changed via JMX
                        request.getMimeHeaders().setLimit(endpoint.getMaxHeaderCount());
                        request.getCookies().setLimit(getMaxCookieCount());
                        // Currently only NIO will ever return false here
                        if (!getInputBuffer().parseHeaders()) {
                            // We've read part of the request, don't recycle it
                            // instead associate it with the socket
                            openSocket = true;
                            readComplete = false;
                            break;
                        }
                        if (!disableUploadTimeout) {
                            setSocketTimeout(connectionUploadTimeout);
                        }
                    }
                } catch (IOException e) {
                    if (getLog().isDebugEnabled()) {
                        getLog().debug(
                                sm.getString("http11processor.header.parse"), e);
                    }
                    setErrorState(ErrorState.CLOSE_NOW, e);
                    break;
                } catch (Throwable t) {
                    ExceptionUtils.handleThrowable(t);
                    UserDataHelper.Mode logMode = userDataHelper.getNextMode();
                    if (logMode != null) {
                        String message = sm.getString(
                                "http11processor.header.parse");
                        switch (logMode) {
                            case INFO_THEN_DEBUG:
                                message += sm.getString(
                                        "http11processor.fallToDebug");
                                //$FALL-THROUGH$
                            case INFO:
                                getLog().info(message, t);
                                break;
                            case DEBUG:
                                getLog().debug(message, t);
                        }
                    }
                    // 400 - Bad Request
                    response.setStatus(400);
                    setErrorState(ErrorState.CLOSE_CLEAN, t);
                    getAdapter().log(request, response, 0);
                }
    
                if (!getErrorState().isError()) {
                    // Setting up filters, and parse some request headers
                    rp.setStage(org.apache.coyote.Constants.STAGE_PREPARE);
                    try {
                        prepareRequest();
                    } catch (Throwable t) {
                        ExceptionUtils.handleThrowable(t);
                        if (getLog().isDebugEnabled()) {
                            getLog().debug(sm.getString(
                                    "http11processor.request.prepare"), t);
                        }
                        // 500 - Internal Server Error
                        response.setStatus(500);
                        setErrorState(ErrorState.CLOSE_CLEAN, t);
                        getAdapter().log(request, response, 0);
                    }
                }
    
                if (maxKeepAliveRequests == 1) {
                    keepAlive = false;
                } else if (maxKeepAliveRequests > 0 &&
                        socketWrapper.decrementKeepAlive() <= 0) {
                    keepAlive = false;
                }
    
                // Process the request in the adapter
                if (!getErrorState().isError()) {
                    try {
                        rp.setStage(org.apache.coyote.Constants.STAGE_SERVICE);
                        getAdapter().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 && !getErrorState().isError() && !isAsync() &&
                                statusDropsConnection(response.getStatus())) {
                            setErrorState(ErrorState.CLOSE_CLEAN, null);
                        }
                        setCometTimeouts(socketWrapper);
                    } catch (InterruptedIOException e) {
                        setErrorState(ErrorState.CLOSE_NOW, e);
                    } catch (HeadersTooLargeException e) {
                        getLog().error(sm.getString("http11processor.request.process"), e);
                        // The response should not have been committed but check it
                        // anyway to be safe
                        if (response.isCommitted()) {
                            setErrorState(ErrorState.CLOSE_NOW, e);
                        } else {
                            response.reset();
                            response.setStatus(500);
                            setErrorState(ErrorState.CLOSE_CLEAN, e);
                            response.setHeader("Connection", "close"); // TODO: Remove
                        }
                    } catch (Throwable t) {
                        ExceptionUtils.handleThrowable(t);
                        getLog().error(sm.getString("http11processor.request.process"), t);
                        // 500 - Internal Server Error
                        response.setStatus(500);
                        setErrorState(ErrorState.CLOSE_CLEAN, t);
                        getAdapter().log(request, response, 0);
                    }
                }
    
                // Finish the handling of the request
                rp.setStage(org.apache.coyote.Constants.STAGE_ENDINPUT);
    
                if (!isAsync() && !comet) {
                    if (getErrorState().isError()) {
                        // 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.
                        getInputBuffer().setSwallowInput(false);
                    } else {
                        // Need to check this again here in case the response was
                        // committed before the error that requires the connection
                        // to be closed occurred.
                        checkExpectationAndResponseStatus();
                    }
                    endRequest();
                }
    
                rp.setStage(org.apache.coyote.Constants.STAGE_ENDOUTPUT);
    
                // If there was an error, make sure the request is counted as
                // and error, and update the statistics counter
                if (getErrorState().isError()) {
                    response.setStatus(500);
                }
    
                if (!isAsync() && !comet || getErrorState().isError()) {
                    request.updateCounters();
                    if (getErrorState().isIoAllowed()) {
                        getInputBuffer().nextRequest();
                        getOutputBuffer().nextRequest();
                    }
                }
    
                if (!disableUploadTimeout) {
                    if(endpoint.getSoTimeout() > 0) {
                        setSocketTimeout(endpoint.getSoTimeout());
                    } else {
                        setSocketTimeout(0);
                    }
                }
    
                rp.setStage(org.apache.coyote.Constants.STAGE_KEEPALIVE);
    
                if (breakKeepAliveLoop(socketWrapper)) {
                    break;
                }
            }
    
            rp.setStage(org.apache.coyote.Constants.STAGE_ENDED);
    
            if (getErrorState().isError() || endpoint.isPaused()) {
                return SocketState.CLOSED;
            } else if (isAsync() || comet) {
                return SocketState.LONG;
            } else if (isUpgrade()) {
                return SocketState.UPGRADING;
            } else {
                if (sendfileInProgress) {
                    return SocketState.SENDFILE;
                } else {
                    if (openSocket) {
                        if (readComplete) {
                            return SocketState.OPEN;
                        } else {
                            return SocketState.LONG;
                        }
                    } else {
                        return SocketState.CLOSED;
                    }
                }
            }
        }

如：上面代码的-》getAdapter().service(request, response);   开始把交给Adapter来处理，并在adapter里处理request请求.

![70 6][]

org.apache.catalina.connector.CoyoteAdapter类:
    
        /**
         * Service method.
         */
        @SuppressWarnings("deprecation")
        @Override
        public void service(org.apache.coyote.Request req,
                            org.apache.coyote.Response res)
            throws Exception {
    
            Request request = (Request) req.getNote(ADAPTER_NOTES);
            Response response = (Response) res.getNote(ADAPTER_NOTES);
    
            if (request == null) {
    
                // Create objects
                request = connector.createRequest();
                request.setCoyoteRequest(req);
                response = connector.createResponse();
                response.setCoyoteResponse(res);
    
                // Link objects
                request.setResponse(response);
                response.setRequest(request);
    
                // Set as notes
                req.setNote(ADAPTER_NOTES, request);
                res.setNote(ADAPTER_NOTES, response);
    
                // Set query string encoding
                req.getParameters().setQueryStringEncoding
                    (connector.getURIEncoding());
    
            }
    
            if (connector.getXpoweredBy()) {
                response.addHeader("X-Powered-By", POWERED_BY);
            }
    
            boolean comet = false;
            boolean async = false;
            boolean postParseSuccess = false;
    
            try {
                // Parse and set Catalina and configuration specific
                // request parameters
                req.getRequestProcessor().setWorkerThreadName(THREAD_NAME.get());
                postParseSuccess = postParseRequest(req, request, res, response);
                if (postParseSuccess) {
                    //check valves if we support async
                    request.setAsyncSupported(connector.getService().getContainer().getPipeline().isAsyncSupported());
                    // Calling the container
                    connector.getService().getContainer().getPipeline().getFirst().invoke(request, response);
    
                    if (request.isComet()) {
                        if (!response.isClosed() && !response.isError()) {
                            comet = true;
                            res.action(ActionCode.COMET_BEGIN, null);
                            if (request.getAvailable() || (request.getContentLength() > 0 && (!request.isParametersParsed()))) {
                                // Invoke a read event right away if there are available bytes
                                event(req, res, SocketStatus.OPEN_READ);
                            }
                        } else {
                            // Clear the filter chain, as otherwise it will not be reset elsewhere
                            // since this is a Comet request
                            request.setFilterChain(null);
                        }
                    }
                }
    
                if (request.isAsync()) {
                    async = true;
                    ReadListener readListener = req.getReadListener();
                    if (readListener != null && request.isFinished()) {
                        // Possible the all data may have been read during service()
                        // method so this needs to be checked here
                        ClassLoader oldCL = null;
                        try {
                            oldCL = request.getContext().bind(false, null);
                            if (req.sendAllDataReadEvent()) {
                                req.getReadListener().onAllDataRead();
                            }
                        } finally {
                            request.getContext().unbind(false, oldCL);
                        }
                    }
                    Throwable throwable =
                            (Throwable) request.getAttribute(RequestDispatcher.ERROR_EXCEPTION);
    
                    // If an async request was started, is not going to end once
                    // this container thread finishes and an error occurred, trigger
                    // the async error process
                    if (!request.isAsyncCompleting() && throwable != null) {
                        request.getAsyncContextInternal().setErrorState(throwable, true);
                    }
                } else if (!comet) {
                    request.finishRequest();
                    response.finishResponse();
                }
            } catch (IOException e) {
                // Ignore
            } finally {
                AtomicBoolean error = new AtomicBoolean(false);
                res.action(ActionCode.IS_ERROR, error);
    
                if (request.isAsyncCompleting() && error.get()) {
                    // Connection will be forcibly closed which will prevent
                    // completion happening at the usual point. Need to trigger
                    // call to onComplete() here.
                    res.action(ActionCode.ASYNC_POST_PROCESS,  null);
                    async = false;
                }
    
                // Access log
                if (!async && !comet) {
                    if (postParseSuccess) {
                        // Log only if processing was invoked.
                        // If postParseRequest() failed, it has already logged it.
                        // If context is null this was the start of a comet request
                        // that failed and has already been logged.
                        request.getMappingData().context.logAccess(
                                request, response,
                                System.currentTimeMillis() - req.getStartTime(),
                                false);
                    }
                }
    
                req.getRequestProcessor().setWorkerThreadName(null);
    
                // Recycle the wrapper request and response
                if (!comet && !async) {
                    request.recycle();
                    response.recycle();
                } else {
                    // Clear converters so that the minimum amount of memory
                    // is used by this processor
                    request.clearEncoders();
                    response.clearEncoders();
                }
            }
    
        }

上面代码中  ：

// Calling the container  
     connector.getService().getContainer().getPipeline().getFirst().invoke(request, response);

也就是调用container类的管道 。

调用后 返回了Request request和Response response的值

invoke()方法 例如实现StandardContextValue类的invoke方法： 代码如下

@Override
        public final void invoke(Request request, Response response)
            throws IOException, ServletException {
    
            // Disallow any direct access to resources under WEB-INF or META-INF
            MessageBytes requestPathMB = request.getRequestPathMB();
            if ((requestPathMB.startsWithIgnoreCase("/META-INF/", 0))
                    || (requestPathMB.equalsIgnoreCase("/META-INF"))
                    || (requestPathMB.startsWithIgnoreCase("/WEB-INF/", 0))
                    || (requestPathMB.equalsIgnoreCase("/WEB-INF"))) {
                response.sendError(HttpServletResponse.SC_NOT_FOUND);
                return;
            }
    
            // Select the Wrapper to be used for this Request
            Wrapper wrapper = request.getWrapper();
            if (wrapper == null || wrapper.isUnavailable()) {
                response.sendError(HttpServletResponse.SC_NOT_FOUND);
                return;
            }
    
            // Acknowledge the request
            try {
                response.sendAcknowledgement();
            } catch (IOException ioe) {
                container.getLogger().error(sm.getString(
                        "standardContextValve.acknowledgeException"), ioe);
                request.setAttribute(RequestDispatcher.ERROR_EXCEPTION, ioe);
                response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
                return;
            }
    
            if (request.isAsyncSupported()) {
                request.setAsyncSupported(wrapper.getPipeline().isAsyncSupported());
            }
            wrapper.getPipeline().getFirst().invoke(request, response);
        }

**Acceptor**

而 org.apache.tomcat.util.net.NioEndpoint类的内部Acceptor实现了AbstractEndpoint.Acceptor代码如下:

org.apache.tomcat.util.net.NioEndpoint类:
    
     // --------------------------------------------------- Acceptor Inner Class
        /**
         * The background thread that listens for incoming TCP/IP connections and
         * hands them off to an appropriate processor.
         */
        protected class Acceptor extends AbstractEndpoint.Acceptor {
    
            @Override
            public void run() {
    
                int errorDelay = 0;
    
                // Loop until we receive a shutdown command
                while (running) {
    
                    // Loop if endpoint is paused
                    while (paused && running) {
                        state = AcceptorState.PAUSED;
                        try {
                            Thread.sleep(50);
                        } catch (InterruptedException e) {
                            // Ignore
                        }
                    }
    
                    if (!running) {
                        break;
                    }
                    state = AcceptorState.RUNNING;
    
                    try {
                        //if we have reached max connections, wait
                        countUpOrAwaitConnection();
    
                        SocketChannel socket = null;
                        try {
                            // Accept the next incoming connection from the server
                            // socket
                            socket = serverSock.accept();
                        } catch (IOException ioe) {
                            //we didn't get a socket
                            countDownConnection();
                            // Introduce delay if necessary
                            errorDelay = handleExceptionWithDelay(errorDelay);
                            // re-throw
                            throw ioe;
                        }
                        // Successful accept, reset the error delay
                        errorDelay = 0;
    
                        // setSocketOptions() will add channel to the poller
                        // if successful
                        if (running && !paused) {
                            if (!setSocketOptions(socket)) {
                                countDownConnection();
                                closeSocket(socket);
                            }
                        } else {
                            countDownConnection();
                            closeSocket(socket);
                        }
                    } catch (SocketTimeoutException sx) {
                        // Ignore: Normal condition
                    } catch (IOException x) {
                        if (running) {
                            log.error(sm.getString("endpoint.accept.fail"), x);
                        }
                    } catch (OutOfMemoryError oom) {
                        try {
                            oomParachuteData = null;
                            releaseCaches();
                            log.error("", oom);
                        }catch ( Throwable oomt ) {
                            try {
                                try {
                                    System.err.println(oomParachuteMsg);
                                    oomt.printStackTrace();
                                }catch (Throwable letsHopeWeDontGetHere){
                                    ExceptionUtils.handleThrowable(letsHopeWeDontGetHere);
                                }
                            }catch (Throwable letsHopeWeDontGetHere){
                                ExceptionUtils.handleThrowable(letsHopeWeDontGetHere);
                            }
                        }
                    } catch (Throwable t) {
                        ExceptionUtils.handleThrowable(t);
                        log.error(sm.getString("endpoint.accept.fail"), t);
                    }
                }
                state = AcceptorState.ENDED;
            }
        }
    
         
         //上面的setSocketOptions方法 实现
    
            /**
         * Process the specified connection.
         */
        protected boolean setSocketOptions(SocketChannel socket) {
            // Process the connection
            try {
                //disable blocking, APR style, we are gonna be polling it
                socket.configureBlocking(false);
                Socket sock = socket.socket();
                socketProperties.setProperties(sock);
    
                NioChannel channel = nioChannels.pop();
                if ( channel == null ) {
                    // SSL setup
                    if (sslContext != null) {
                        SSLEngine engine = createSSLEngine();
                        int appbufsize = engine.getSession().getApplicationBufferSize();
                        NioBufferHandler bufhandler = new NioBufferHandler(Math.max(appbufsize,socketProperties.getAppReadBufSize()),
                                                                           Math.max(appbufsize,socketProperties.getAppWriteBufSize()),
                                                                           socketProperties.getDirectBuffer());
                        channel = new SecureNioChannel(socket, engine, bufhandler, selectorPool);
                    } else {
                        // normal tcp setup
                        NioBufferHandler bufhandler = new NioBufferHandler(socketProperties.getAppReadBufSize(),
                                                                           socketProperties.getAppWriteBufSize(),
                                                                           socketProperties.getDirectBuffer());
    
                        channel = new NioChannel(socket, bufhandler);
                    }
                } else {
                    channel.setIOChannel(socket);
                    if ( channel instanceof SecureNioChannel ) {
                        SSLEngine engine = createSSLEngine();
                        ((SecureNioChannel)channel).reset(engine);
                    } else {
                        channel.reset();
                    }
                }
                getPoller0().register(channel);
            } catch (Throwable t) {
                ExceptionUtils.handleThrowable(t);
                try {
                    log.error("",t);
                } catch (Throwable tt) {
                    ExceptionUtils.handleThrowable(tt);
                }
                // Tell to close the socket
                return false;
            }
            return true;
        }

代码里面的  // Accept the next incoming connection from the server socket  
       socket = serverSock.accept();    用Socket来连接接收数据

setSocketOptions方法里的  getPoller0().register(channel);代码如下:

上面的setSocketOptions方法里的  getPoller0().register(channel);
    
         /**
             * Registers a newly created socket with the poller.
             *
             * @param socket    The newly created socket
             */
            public void register(final NioChannel socket) {
                socket.setPoller(this);
                KeyAttachment ka = new KeyAttachment(socket);
                ka.setPoller(this);
                ka.setTimeout(getSocketProperties().getSoTimeout());
                ka.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
                ka.setSecure(isSSLEnabled());
                PollerEvent r = eventCache.pop();
                ka.interestOps(SelectionKey.OP_READ);//this is what OP_REGISTER turns into.
                if ( r==null) r = new PollerEvent(socket,ka,OP_REGISTER);
                else r.reset(socket,ka,OP_REGISTER);
                addEvent(r);
            }

也就是在Acceptor里面 调用了getPoller0().register(channel);注册 代码实现如下：

org.apache.tomcat.util.net.NioEndpoint类:
    
    
    /**
             * Registers a newly created socket with the poller.
             *
             * @param socket    The newly created socket
             */
            public void register(final NioChannel socket) {
                socket.setPoller(this);
                KeyAttachment ka = new KeyAttachment(socket);
                ka.setPoller(this);
                ka.setTimeout(getSocketProperties().getSoTimeout());
                ka.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
                ka.setSecure(isSSLEnabled());
                PollerEvent r = eventCache.pop();
                ka.interestOps(SelectionKey.OP_READ);//this is what OP_REGISTER turns into.
                if ( r==null) r = new PollerEvent(socket,ka,OP_REGISTER);
                else r.reset(socket,ka,OP_REGISTER);
                addEvent(r);
            }

PollerEvent，用于轮询器事件的可缓存对象，以避免GC

在Poller类 的run()方法里面 ：

if (close) \{  
                        events();  
                        timeout(0, false);  
                        try \{  
                            selector.close();  
                        \} catch (IOException ioe) \{  
                            log.error(sm.getString(  
                                    "endpoint.nio.selectorCloseFail"), ioe);  
                        \}  
                        break;  
                    \} else \{  
                        hasEvents = events();  
                    \}

如果没有close,执行  hasEvents = events();

NioEndpoint类：
    
    
     /**
    
      * Processes events in the event queue of the Poller.
             *
             * @return <code>true</code> if some events were processed,
             *   <code>false</code> if queue was empty
             */
            public boolean events() {
                boolean result = false;
    
                PollerEvent pe = null;
                for (int i = 0, size = events.size(); i < size && (pe = events.poll()) != null; i++ ) {
                    result = true;
                    try {
                        pe.run();
                        pe.reset();
                        if (running && !paused) {
                            eventCache.push(pe);
                        }
                    } catch ( Throwable x ) {
                        log.error("",x);
                    }
                }
    
                return result;
            }

pe.run();运行了PollerEvent 类的run方法：

/**
         *
         * PollerEvent, cacheable object for poller events to avoid GC
         */
        public static class PollerEvent implements Runnable {
    
            private NioChannel socket;
            private int interestOps;
            private KeyAttachment key;
    
            public PollerEvent(NioChannel ch, KeyAttachment k, int intOps) {
                reset(ch, k, intOps);
            }
    
            public void reset(NioChannel ch, KeyAttachment k, int intOps) {
                socket = ch;
                interestOps = intOps;
                key = k;
            }
    
            public void reset() {
                reset(null, null, 0);
            }
    
            @Override
            public void run() {
                if ( interestOps == OP_REGISTER ) {
                    try {
                        socket.getIOChannel().register(socket.getPoller().getSelector(), SelectionKey.OP_READ, key);
                    } catch (Exception x) {
                        log.error("", x);
                    }
                } else {
                    final SelectionKey key = socket.getIOChannel().keyFor(socket.getPoller().getSelector());
                    try {
                        if (key == null) {
                            // The key was cancelled (e.g. due to socket closure)
                            // and removed from the selector while it was being
                            // processed. Count down the connections at this point
                            // since it won't have been counted down when the socket
                            // closed.
                            socket.getPoller().getEndpoint().countDownConnection();
                        } else {
                            final KeyAttachment att = (KeyAttachment) key.attachment();
                            if ( att!=null ) {
                                att.access();//to prevent timeout
                                //we are registering the key to start with, reset the fairness counter.
                                int ops = key.interestOps() | interestOps;
                                att.interestOps(ops);
                                key.interestOps(ops);
                            } else {
                                socket.getPoller().cancelledKey(key, SocketStatus.ERROR);
                            }
                        }
                    } catch (CancelledKeyException ckx) {
                        try {
                            socket.getPoller().cancelledKey(key, SocketStatus.DISCONNECT);
                        } catch (Exception ignore) {}
                    }
                }//end if
            }//run
    
            @Override
            public String toString() {
                return super.toString()+"[intOps="+this.interestOps+"]";
            }
        }

为Socket的SocketChannel 注册了Selector等。

而这时的        if ( interestOps == OP\_REGISTER ) \{...\}

也就是判断了interestOps是否是 OP\_REGISTER，而在getPoller0().register(channel)方法里面

if ( r==null) r = new PollerEvent(socket,ka,OP\_REGISTER);  
            else r.reset(socket,ka,OP\_REGISTER);

为PollerEvent设置类OP\_REGISTER(  public static final int OP\_REGISTER = 0x100; //register interest op)

**    2、Processor和Adapter分析**

Processor用于处理应用层协议(如：Http协议)

![70 7][]

上面的 upgrade是在Servlet3.1知乎新增的，用于处理Http的升级协议(如WebSocket协议)。当下面的Processor处理之后，Socket的状态是UPGRADE,那么Endpoint的Handler将会创建并调用org.apache.coyote.http11.upgrade包下面的Processor进行处理。

**Adapter **

![70 8][]

Processor在其process方法中会调用Adapter的service方法来处理请求，Adapter的Service方法调用Container管道中的invoke方法处理请求，在处理之前把org.apache.coyote下面的Request和Response封装成org.apache.catalina.connector下的Request和Response,  而代码里面的 判断是否启动了Comet （长连接模式）和是否启动了异步。

也就是前面分析的 CoyoteAdapter类的 process方法

// Calling the container  
     connector.getService().getContainer().getPipeline().getFirst().invoke(request, response);

Service中是保存的是最顶层的容器，当调用Service管道的invoke方法时，管道会逐层调用各层容器中Value的invoke方法。直到最后调用Wrapper的管道的BaseValue来处理Filter和Servlet.

[70]: /images/20220514/52904a3ca39a46f3a27e5850624741fd.png
[70 1]: /images/20220514/2baab66ce84a461f8a41683ad6e82b25.png
[Google]: https://baike.baidu.com/item/Google
[TCP]: https://baike.baidu.com/item/TCP/33012
[HTTP]: https://baike.baidu.com/item/HTTP
[70 2]: /images/20220514/162e2d1c36e2462398d288035cc0a3f0.png
[70 3]: /images/20220514/c41651f39a504d75898870a8fce03087.png
[70 4]: /images/20220514/91f66090602d4f268cb3a1f24bbad0f6.png
[70 5]: /images/20220514/86f9cdc181ba49988f9c30fdbdacfc38.png
[70 6]: /images/20220514/a2eea75017cd49a4b82a4722879b21b8.png
[70 7]: /images/20220514/b0c6229b576f48ca9bb119b6cfd57949.png
[70 8]: /images/20220514/f9563a9ed1364fa0afe1af076990eb0c.png