Tomcat NIO(9)-IO线程-Overall流程和关键类

在上一篇文章里我们主要介绍了 tomcat NIO 中 poller 线程的阻塞与唤醒，根据以前文章当 poller 线程监测到连接有数据可读事件的时候，会把原始 socket 的包装对象委托到 tomcat io 线程池中处理，包括解析请求行，请求头，调用 servlet API，处理异步等等，在这里我们主要介绍 tomcat io 线程。

对于tomcat io线程我们主要介绍：

IO 线程 overall 流程
IO 线程主要涉及的类以及作用

IO线程overall流程

对于 tomcat io 线程来说，overall 调用序列图如下：

SocketProcessor.run()--> ConnectionHandler.process()--> AbstractProcessorLight.process()--> Http11Processor.service--> CoyoteAdapter.service() 一直到 container 调用标准 servlet API。
涉及的关键类有 SocketProcessor，ConnectionHandler AbstractProcessorLight， Http11Processor，CoyoteAdapter

SocketProcessor的核心代码逻辑如下

protected void doRun() {
    NioChannel socket = socketWrapper.getSocket();
    SelectionKey key = socket.getIOChannel().keyFor(socket.getSocketWrapper().getPoller().getSelector());
    Poller poller = NioEndpoint.this.poller;
    if (poller == null) {
        socketWrapper.close();
        return;
    }
    try {
        int handshake = -1;
        try {
            if (key != null) {
                if (socket.isHandshakeComplete()) {
                    handshake = 0;
                } else if (event == SocketEvent.STOP || event == SocketEvent.DISCONNECT ||event == SocketEvent.ERROR) {
                    handshake = -1;
                } else {
                    handshake = socket.handshake(key.isReadable(), key.isWritable());
                    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;
            if (event == null) {
                state = getHandler().process(socketWrapper, SocketEvent.OPEN_READ);
            } else {
                state = getHandler().process(socketWrapper, event);
            }
            if (state == SocketState.CLOSED) {
                poller.cancelledKey(key, socketWrapper);
            }
        } else if (handshake == -1 ) {
            poller.cancelledKey(key, socketWrapper);
        } else if (handshake == SelectionKey.OP_READ){
            socketWrapper.registerReadInterest();
        } else if (handshake == SelectionKey.OP_WRITE){
            socketWrapper.registerWriteInterest();
        }
    } catch (CancelledKeyException cx) {
        poller.cancelledKey(key, socketWrapper);
    } catch (VirtualMachineError vme) {
        ExceptionUtils.handleThrowable(vme);
    } catch (Throwable t) {
        log.error(sm.getString("endpoint.processing.fail"), t);
        poller.cancelledKey(key, socketWrapper);
    } finally {
        socketWrapper = null;
        event = null;
        if (running && !paused && processorCache != null) {
            processorCache.push(this);
        }
    }
}

首先会处理 handshake，如果 handshake 没有问题则返回 handshake 的结果为 0。
如果 handshake 过程处理正常没问题，则会通过调用 getHandler().process(socketWrapper, event) 方法从而来间接触发 ConnectionHandler 实例的 process() 方法，并返回期望原始 socket 的状态 SocketState 枚举。
如果返回的 SocketState 为 CLOSED ，则调用 poller.cancelledKey() 方法，会把原始 sockte 关闭。
最后会把 SocketProcessor 实例回收到缓存 processorCache 中，以便下次使用不需要重新创建对象，从而提高效率。
另外 ConnectionHandler 是 global 对象，也就是说所有的连接处理均由这个对象处理。根据以前文章，该实例中有一个 Map 对象，key 为SocketWrapper 对象类型，对应的 value 为 Http11Processor 类型。也就是说为连接中的每一个请求(request)都去分配了相应处理类 Http11Processor 实例，可以保存连接上请求的状态信息(例如解析请求行，请求头等数据)。

ConnectionHandler的核心代码逻辑如下

private final Map<S,Processor> connections = new ConcurrentHashMap<>();
private final RecycledProcessors recycledProcessors = new RecycledProcessors(this);

public SocketState process(SocketWrapperBase<S> wrapper, SocketEvent status) {
    if (getLog().isDebugEnabled()) {
        getLog().debug(sm.getString("abstractConnectionHandler.process",
                wrapper.getSocket(), status));
    }
    if (wrapper == null) {
        return SocketState.CLOSED;
    }

    S socket = wrapper.getSocket();
    Processor processor = connections.get(socket);
    if (getLog().isDebugEnabled()) {
        getLog().debug(sm.getString("abstractConnectionHandler.connectionsGet",
                processor, socket));
    }

    if (SocketEvent.TIMEOUT == status && (processor == null || !processor.isAsync() || !processor.checkAsyncTimeoutGeneration())) {
        return SocketState.OPEN;
    }

    if (processor != null) {
        getProtocol().removeWaitingProcessor(processor);
    } else if (status == SocketEvent.DISCONNECT || status == SocketEvent.ERROR) {
        return SocketState.CLOSED;
    }

    ContainerThreadMarker.set();

    try {
        if (processor == null) {
            String negotiatedProtocol = wrapper.getNegotiatedProtocol();
            if (negotiatedProtocol != null && negotiatedProtocol.length() > 0) {
                UpgradeProtocol upgradeProtocol =
                        getProtocol().getNegotiatedProtocol(negotiatedProtocol);
                if (upgradeProtocol != null) {
                    processor = upgradeProtocol.getProcessor(
                            wrapper, getProtocol().getAdapter());
                } else if (negotiatedProtocol.equals("http/1.1")) {
                    // Explicitly negotiated the default protocol.
                    // Obtain a processor below.
                } else {
                    if (getLog().isDebugEnabled()) {
                        getLog().debug(sm.getString(
                            "abstractConnectionHandler.negotiatedProcessor.fail",
                            negotiatedProtocol));
                    }
                    return SocketState.CLOSED;
                }
            }
        }
        if (processor == null) {
            processor = recycledProcessors.pop();
            if (getLog().isDebugEnabled()) {
                getLog().debug(sm.getString("abstractConnectionHandler.processorPop",
                        processor));
            }
        }
        if (processor == null) {
            processor = getProtocol().createProcessor();
            register(processor);
        }

        processor.setSslSupport(
                wrapper.getSslSupport(getProtocol().getClientCertProvider()));

        connections.put(socket, processor);

        SocketState state = SocketState.CLOSED;
        do {
            state = processor.process(wrapper, status);

            if (state == SocketState.UPGRADING) {
                UpgradeToken upgradeToken = processor.getUpgradeToken();
                ByteBuffer leftOverInput = processor.getLeftoverInput();
                if (upgradeToken == null) {
                    UpgradeProtocol upgradeProtocol = getProtocol().getUpgradeProtocol("h2c");
                    if (upgradeProtocol != null) {
                        processor = upgradeProtocol.getProcessor(
                                wrapper, getProtocol().getAdapter());
                        wrapper.unRead(leftOverInput);
                        connections.put(socket, processor);
                    } else {
                        if (getLog().isDebugEnabled()) {
                            getLog().debug(sm.getString("abstractConnectionHandler.negotiatedProcessor.fail","h2c"));
                        }
                        return SocketState.CLOSED;
                    }
                } else {
                    HttpUpgradeHandler httpUpgradeHandler = upgradeToken.getHttpUpgradeHandler();
                    release(processor);
                    processor = getProtocol().createUpgradeProcessor(wrapper, upgradeToken);
                    if (getLog().isDebugEnabled()) {
                        getLog().debug(sm.getString("abstractConnectionHandler.upgradeCreate",processor, wrapper));
                    }
                    wrapper.unRead(leftOverInput);
                    wrapper.setUpgraded(true);
                    connections.put(socket, processor);
                    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 (httpUpgradeHandler instanceof InternalHttpUpgradeHandler) {
                        if (((InternalHttpUpgradeHandler) httpUpgradeHandler).hasAsyncIO()) {
                            state = SocketState.LONG;
                        }
                    }
                }
            }
        } while ( state == SocketState.UPGRADING);

        if (state == SocketState.LONG) {
            longPoll(wrapper, processor);
            if (processor.isAsync()) {
                getProtocol().addWaitingProcessor(processor);
            }
        } else if (state == SocketState.OPEN) {
            connections.remove(socket);
            release(processor);
            wrapper.registerReadInterest();
        } 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.
        } 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 != SocketEvent.OPEN_WRITE) {
                longPoll(wrapper, processor);
            }
        } else if (state == SocketState.SUSPENDED) {
            // Don't add sockets back to the poller.
            // The resumeProcessing() method will add this socket
            // to the poller.
        } else {
            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(processor);
            }
        }
        return state;
    } catch(java.net.SocketException e) {
        getLog().debug(sm.getString("abstractConnectionHandler.socketexception.debug"), e);
    } catch (java.io.IOException e) {
        getLog().debug(sm.getString("abstractConnectionHandler.ioexception.debug"), e);
    } catch (ProtocolException e) {
        getLog().debug(sm.getString("abstractConnectionHandler.protocolexception.debug"), e);
    }
    catch (OutOfMemoryError oome) {
        getLog().error(sm.getString("abstractConnectionHandler.oome"), oome);
    } catch (Throwable e) {
        ExceptionUtils.handleThrowable(e);
        getLog().error(sm.getString("abstractConnectionHandler.error"), e);
    } finally {
        ContainerThreadMarker.clear();
    }
    connections.remove(socket);
    release(processor);
    return SocketState.CLOSED;
}

该实例中有一个 Map 对象，key 为原始 scoket 的包装对象 SocketWrapper 类型，value 为 Http11Processor 类型。为每个连接中的请求都分配处理类 Http11Processor 实例，可以保存连接上的请求状态信息(例如解析请求行，请求头等数据)。
该实例有 recycledProcessors 对象，用来保持已经回收的 Http11Processor 实例，避免下次使用重新创建对象，提高效率。
该实例核心方法为 process() 方法，代码比较多，这里总结关键点。
该实例寻找 Map 中原始 socket 的包装对象关联的连接处理类 Http11Processor 实例，如果没有则创建新实例，并将 socket 包装对象与其关联到 Map 里，下次同一个连接中有数据的时候可以重用，而且也可以保存连接状态信息在 Http11Processor 实例中。
该实例调用 Http11Processor 实例的 process 方法并返回 SocketState。
如果返回的状态是代表 upgrade 协议(例如websocket连接等)，则处理 upgrade 协议，这里不对 upgrade 协议详细展开。
如果回的状态为 SocketState.LONG ，则代表要么是数据(请求行/请求头)没有解析完(因为 client 端没有发送完请求行/请求头数据)，要么是执行了 servlet 的异步请求。
对于 SocketState.LONG 的返回值调用如下： protected void longPoll(SocketWrapperBase<?> socket, Processor processor) { if (!processor.isAsync()) { // This is currently only used with HTTP // Either: // - this is an upgraded connection // - the request line/headers have not been completely // read socket.registerReadInterest(); } } //SocketWrapper public void registerReadInterest() { getPoller().add(this, SelectionKey.OP_READ); }
longPoll() 方法在 servlet 非异步请求(请求行/请求头数据没有解析完)的情况，对 socket 包装对象注册 OP_READ 事件，并添加到 poller 线程的事件队列里，让 poller 线程继续监听 client 端可读事件。
对于 SocketState.OPEN 的返回值调用： connections.remove(socket); release(processor); wrapper.registerReadInterest();
SocketState.OPEN 一般代表 servlet API 调用正常，返回 OPEN 表示该连接为长连接，不关闭原始 socket 。所以在 Map中会去移除 socket 和Http11Processor 的对应关系，来释放当前 Http11Processor 实例以便后续重用。由于是长连接，所以和异步处理方式一样，对 socket 包装对象注册 OP_READ 事件，并添加到 poller 线程事件队列中，让 poller 线程继续监听 client 端可读事件。
在最后的 else 分支中代表返回的期望状态为 CLOSED ，表示该连接需要关闭，则在 Map 中移除 socket 和 Http11Processor 的对应关系，然后会释放当前 Http11Processor 实例以便后续重用。根据上面 ConnectionHanlder 的分析，如果返回的 SocketState 枚举的结果为 CLOSED，则会去调用 poller.cancelledKey() 方法，从而把原始 socket 关闭。

目前先写到这里，下一篇文章里我们继续介绍 tomcat io 线程涉及的其他关键类。