Tomcat源码解析之Web请求与处理

（5）NioEndPoint.NioSocketWrapper，是Socket的封装类，增强类，将Socket与其他对象建立关联。

 public static class NioSocketWrapper extends SocketWrapperBase {
 		private final NioSelectorPool pool;

        private Poller poller = null; // 轮询的Poller 
        private int interestOps = 0;
        private CountDownLatch readLatch = null;
        private CountDownLatch writeLatch = null;
        private volatile SendfileData sendfileData = null;
        private volatile long lastRead = System.currentTimeMillis();
        private volatile long lastWrite = lastRead;
        private volatile boolean closed = false;

（6）NioEndPoint.SocketProcessor（Worker）继承了Runnable接口，负责对socket的g各种事件进行处理。读事件、写事件、停止时间、超时事件、断连事件、错误时间、连接失败事件。

SocketProcessor的doRun方法，会根据SocketState进行处理，SocketState 为STOP、DISCONNECT或者ERROR的时候就进行关闭，SocketWrapperBase对应的selector事件，得到指定的Handler处理器进行处理。

@Override
 protected void doRun() {
     NioChannel socket = socketWrapper.getSocket();
     SelectionKey key = socket.getIOChannel().keyFor(socket.getPoller().getSelector());

     try {
         int handshake = -1;

         try {
             if (key != null) {
                 if (socket.isHandshakeComplete()) {
                     // 是否已经握手成功，不需要TLS(加密)握手，就让处理器对socket和event的组合进行处理。
                     handshake = 0;
                 } else if (event == SocketEvent.STOP || event == SocketEvent.DISCONNECT ||
                         event == SocketEvent.ERROR) {
                     // 不能够完成TLS握手，就把他认为是TLS握手失败。
                     handshake = -1;
                 } 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.
                     // 握手从/向socket读/写时，握手一旦完成状态应该为OPEN_WRITE，
                     // 握手是在套接字打开时发生的，因此在完成后状态必须始终为OPEN_READ
                     // 始终设置此选项是可以的，因为它仅在握手完成时使用。
                     event = SocketEvent.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 (event == null) {
                 // 调用处理器进行处理。
                 // NioEndPoint的默认Handler是Http11的
                 // 这里的Handler是AbstractProtocol.ConnectionHandler
                 // 这个Handler的设置方法是：
                 // 首先在Connector类的构造函数中，将默认的ProtocolHandler设置为org.apache.coyote.http11.Http11NioProtocol
                 // AbstractHttp11Protocol的构造函数里面创建了Handler类ConnectionHandler
                 state = getHandler().process(socketWrapper, SocketEvent.OPEN_READ);
             } else {
                 state = getHandler().process(socketWrapper, event);
             }
             // 如果返回的状态是SocketState，那么就关掉连接
             if (state == SocketState.CLOSED) {
                 close(socket, key);
             }
         } else if (handshake == -1) {
             getHandler().process(socketWrapper, SocketEvent.CONNECT_FAIL);
             close(socket, key);
         } else if (handshake == SelectionKey.OP_READ) {
             // 如果是SelectionKey.OP_READ，也就是读事件的话，就将OP_READ时间设置到socketWrapper
             socketWrapper.registerReadInterest();
         } else if (handshake == SelectionKey.OP_WRITE) {
             // 如果是SelectionKey.OP_WRITE，也就是读事件的话，就将OP_WRITE事件设置到socketWrapper
             socketWrapper.registerWriteInterest();
         }

二、ConnectionHandler

（1）ConnectionHandler用于根据Socket连接找到相应的Engine处理器。

上面是SocketProcessor的doRun方法，执行了getHandler().process(socketWrapper, SocketEvent.OPEN_READ);;process方法是首先在Map缓存中查找当前socket是否存在对应的processor，如果不存在，再去可循环的处理器栈中查找是否存在，如果不存在就创建相应的Processor，然后将新创建的Processor与Socket建立映射，存在connection的Map中。在任何一个阶段得到Processor对象之后，会执行processor的process方法state = processor.process(wrapper, status);

protected static class ConnectionHandler implements AbstractEndpoint.Handler {

        private final AbstractProtocol proto;
        private final RequestGroupInfo global = new RequestGroupInfo();
        private final AtomicLong registerCount = new AtomicLong(0);
        // 终于找到了这个集合，给Socket和处理器建立连接
        // 对每个有效链接都会缓存进这里，用于连接选择一个合适的Processor实现以进行请求处理。
        private final Map connections = new ConcurrentHashMap<>();
        // 可循环的处理器栈
        private final RecycledProcessors recycledProcessors = new RecycledProcessors(this);

		
  		@Override
        public SocketState process(SocketWrapperBase wrapper, SocketEvent status) {
            if (getLog().isDebugEnabled()) {
                getLog().debug(sm.getString("abstractConnectionHandler.process",
                        wrapper.getSocket(), status));
            }
            if (wrapper == null) {
                // wrapper == null 表示Socket已经被关闭了，所以不需要做任何操作。
                return SocketState.CLOSED;
            }
            // 得到wrapper内的Socket对象
            S socket = wrapper.getSocket();
            // 从Map缓冲区中得到socket对应的处理器。
            Processor processor = connections.get(socket);
            if (getLog().isDebugEnabled()) {
                getLog().debug(sm.getString("abstractConnectionHandler.connectionsGet",
                        processor, socket));
            }

            // Timeouts are calculated on a dedicated thread and then
            // dispatched. Because of delays in the dispatch process, the
            // timeout may no longer be required. Check here and avoid
            // unnecessary processing.

            // 超时是在专用线程上计算的，然后被调度。
            // 因为调度过程中的延迟，可能不再需要超时。检查这里，避免不必要的处理。
            if (SocketEvent.TIMEOUT == status &&
                    (processor == null ||
                            !processor.isAsync() && !processor.isUpgrade() ||
                            processor.isAsync() && !processor.checkAsyncTimeoutGeneration())) {
                // This is effectively a NO-OP
                return SocketState.OPEN;
            }
            // 如果Map缓存存在该socket相关的处理器
            if (processor != null) {
                // Make sure an async timeout doesn't fire
                // 确保没有触发异步超时
                getProtocol().removeWaitingProcessor(processor);
            } else if (status == SocketEvent.DISCONNECT || status == SocketEvent.ERROR) {
                // Nothing to do. Endpoint requested a close and there is no
                // longer a processor associated with this socket.
                // SocketEvent事件是关闭，或者SocketEvent时间出错，此时不需要做任何操作。
                // Endpoint需要一个CLOSED的信号，并且这里不再有与这个socket有关联了
                return SocketState.CLOSED;
            }

            ContainerThreadMarker.set();

            try {
                // Map缓存不存在该socket相关的处理器
                if (processor == null) {
                    String negotiatedProtocol = wrapper.getNegotiatedProtocol();
                    // OpenSSL typically returns null whereas JSSE typically
                    // returns "" when no protocol is negotiated
                    // OpenSSL通常返回null，而JSSE通常在没有协议协商时返回""
                    if (negotiatedProtocol != null && negotiatedProtocol.length() > 0) {
                        // 获取协商协议
                        UpgradeProtocol upgradeProtocol = getProtocol().getNegotiatedProtocol(negotiatedProtocol);
                        if (upgradeProtocol != null) {
                            // 升级协议为空
                            processor = upgradeProtocol.getProcessor(wrapper, getProtocol().getAdapter());
                            if (getLog().isDebugEnabled()) {
                                getLog().debug(sm.getString("abstractConnectionHandler.processorCreate", processor));
                            }
                        } else if (negotiatedProtocol.equals("http/1.1")) {
                            // Explicitly negotiated the default protocol.
                            // Obtain a processor below.
                        } else {
                            // TODO:
                            // OpenSSL 1.0.2's ALPN callback doesn't support
                            // failing the handshake with an error if no
                            // protocol can be negotiated. Therefore, we need to
                            // fail the connection here. Once this is fixed,
                            // replace the code below with the commented out
                            // block.
                            if (getLog().isDebugEnabled()) {
                                getLog().debug(sm.getString("abstractConnectionHandler.negotiatedProcessor.fail",
                                        negotiatedProtocol));
                            }
                            return SocketState.CLOSED;
                            /*
                             * To replace the code above once OpenSSL 1.1.0 is
                             * used.
                            // Failed to create processor. This is a bug.
                            throw new IllegalStateException(sm.getString(
                                    "abstractConnectionHandler.negotiatedProcessor.fail",
                                    negotiatedProtocol));
                            */
                        }
                    }
                }
                // 经过上面的操作，processor还是null的话。
                if (processor == null) {
                    // 从recycledProcessors可循环processors中获取processor
                    processor = recycledProcessors.pop();
                    if (getLog().isDebugEnabled()) {
                        getLog().debug(sm.getString("abstractConnectionHandler.processorPop", processor));
                    }
                }
                if (processor == null) {
                    // 创建处理器
                    processor = getProtocol().createProcessor();
                    register(processor);
                    if (getLog().isDebugEnabled()) {
                        getLog().debug(sm.getString("abstractConnectionHandler.processorCreate", processor));
                    }
                }
                processor.setSslSupport(
                        wrapper.getSslSupport(getProtocol().getClientCertProvider()));

                // 将socket和processor建立关联。
                connections.put(socket, processor);

                SocketState state = SocketState.CLOSED;
                do {
                    // 调用processor的process方法。
                    state = processor.process(wrapper, status);

                    // processor的process方法返回升级状态
                    if (state == SocketState.UPGRADING) {
                        // Get the HTTP upgrade handler
                        // 得到HTTP的升级句柄
                        UpgradeToken upgradeToken = processor.getUpgradeToken();
                        // Retrieve leftover input
                        // 检索剩余输入
                        ByteBuffer leftOverInput = processor.getLeftoverInput();
                        if (upgradeToken == null) {
                            // Assume direct HTTP/2 connection
                            UpgradeProtocol upgradeProtocol = getProtocol().getUpgradeProtocol("h2c");
                            if (upgradeProtocol != null) {
                                // Release the Http11 processor to be re-used
                                release(processor);
                                // Create the upgrade processor
                                processor = upgradeProtocol.getProcessor(wrapper, getProtocol().getAdapter());
                                wrapper.unRead(leftOverInput);
                                // Associate with the processor with the connection
                                connections.put(socket, processor);
                            } else {
                                if (getLog().isDebugEnabled()) {
                                    getLog().debug(sm.getString(
                                            "abstractConnectionHandler.negotiatedProcessor.fail",
                                            "h2c"));
                                }
                                // Exit loop and trigger appropriate clean-up
                                state = SocketState.CLOSED;
                            }
                        } else {
                            HttpUpgradeHandler httpUpgradeHandler = upgradeToken.getHttpUpgradeHandler();
                            // Release the Http11 processor to be re-used
                            release(processor);
                            // Create the upgrade processor
                            processor = getProtocol().createUpgradeProcessor(wrapper, upgradeToken);
                            if (getLog().isDebugEnabled()) {
                                getLog().debug(sm.getString("abstractConnectionHandler.upgradeCreate",
                                        processor, wrapper));
                            }
                            wrapper.unRead(leftOverInput);
                            // 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);
                                }
                            }
                        }
                    }
                } while (state == SocketState.UPGRADING);	

@Override
    public SocketState service(SocketWrapperBase<?> socketWrapper)
            throws IOException {
		// 上面省略n行
		// 调用Coyote的service方法
		 getAdapter().service(request, response);
		 // 下面省略n行

@Override
    public SocketState service(SocketWrapperBase<?> socketWrapper)
            throws IOException {
		// 上面省略n行
		// 调用Coyote的service方法
		 getAdapter().service(request, response);
		 // 下面省略n行

@Override
    public SocketState service(SocketWrapperBase<?> socketWrapper)
            throws IOException {
		// 上面省略n行
		// 调用Coyote的service方法
		 getAdapter().service(request, response);
		 // 下面省略n行

@Override
    public final void invoke(Request request, Response response)
            throws IOException, ServletException {
        // 初始化我们需要的本地变量
        boolean unavailable = false;
        Throwable throwable = null;
        // This should be a Request attribute...
        long t1 = System.currentTimeMillis();
        // 原子类AtomicInteger，CAS操作，表示请求的数量。
        requestCount.incrementAndGet();
        StandardWrapper wrapper = (StandardWrapper) getContainer();
        Servlet servlet = null;
        Context context = (Context) wrapper.getParent();

        // 检查当前的Context应用是否已经被标注为不可以使用
        if (!context.getState().isAvailable()) {
            // 如果当前应用不可以使用的话，就报503错误。
            response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE,
                    sm.getString("standardContext.isUnavailable"));
            unavailable = true;
        }

        // 检查Servelt是否被标记为不可使用
        if (!unavailable && wrapper.isUnavailable()) {
            container.getLogger().info(sm.getString("standardWrapper.isUnavailable",
                    wrapper.getName()));
            long available = wrapper.getAvailable();
            if ((available > 0L) && (available < Long.MAX_VALUE)) {
                response.setDateHeader("Retry-After", available);
                response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE,
                        sm.getString("standardWrapper.isUnavailable",
                                wrapper.getName()));
            } else if (available == Long.MAX_VALUE) {
                response.sendError(HttpServletResponse.SC_NOT_FOUND,
                        sm.getString("standardWrapper.notFound",
                                wrapper.getName()));
            }
            unavailable = true;
        }
        // Servelt是第一次调用的时候初始化
        try {
            if (!unavailable) {
                // 如果此时Servelt还没有被初始化，就分配一个Servelt实例来处理request请求。
                servlet = wrapper.allocate();
            }
        /// 省略代码..........................................
        // // 给该request创建Filter过滤链。Filter过滤链执行完之后，会执行Servelt
        ApplicationFilterChain filterChain =
                ApplicationFilterFactory.createFilterChain(request, wrapper, servlet);

        // Call the filter chain for this request
        // NOTE: This also calls the servlet's service() method
        try {
            if ((servlet != null) && (filterChain != null)) {
                // Swallow output if needed
                if (context.getSwallowOutput()) {
                    try {
                        SystemLogHandler.startCapture();
                        if (request.isAsyncDispatching()) {
                            request.getAsyncContextInternal().doInternalDispatch();
                        } else {
                            // 调用过滤链
                            filterChain.doFilter(request.getRequest(),
                                    response.getResponse());
                        }
        /// 省略代码..........................................