tomcat趣谈之NIO源码详解（这回你总该懂了吧）

vlambda
2021-06-16

tomcat趣谈之NIO源码详解（这回你总该懂了吧）

小故事：在讲tomcat的nio之前，我先给大家讲一个小故事，故事的主人翁是一休和他的老板，一休大四培训完以后兴致勃勃的就出去找工作，在找了一两个月投了几百封简历以后终于找到了一家小的创业公司，整个公司加上他总共两个人，磨洋工是不可能磨洋工的，因为就连呼吸老板都能感觉得到，老板是一个三十七八岁的中年大叔，以前也是写代码的，听他自己说是厌倦了上班的枯燥，想做自己生活的主人，说的煞有介事的，不知真假，第一次见面，老板就满脸堆笑的对一休说，虽然我这里工资不多，但是干的活你放心绝对不会少的，这对你以后的职业生涯来说是很有好处的，跟着我保证你吃香喝辣的，不出两年就让你升任ceo...

公司的业务是接外包项目，老板工作了多年有一些人脉和资源，负责拉客，一休就负责伺候这些客人，虽然工资不多，但是每天也过的挺充实的，老板虽然不写代码，但是好歹也写过十几年，为了充分发挥一休的工作才能，对一休提出的问题也是知无不言，甚至还会让一休发散思维多想想如何提高工作效率，刚开始一休还是比较开心的，慢慢的，老板拉的客人越来越多了，一休开始渐渐应付不过来了，于是开始每天向老板吐槽，吐槽了三个月以后老板突然良心发现给他招了几个小弟，美其明曰兑现他当初许下的CEO承诺，有了小弟以后一休就不写代码了，负责给客人分小弟的号牌，从此过上了CEO的幸福生活！！！

我们现在开始进入正题，首先来一张tomcat的基本架构图：今天要讲的主角就是红色箭头标记的Connector

这张是Connector的详细内部结构：

tomcat趣谈之NIO源码详解（这回你总该懂了吧）

一：ProtocolHandler:支持nio协议，tomcat支持多种协议（可配）

如下是tomcat的配置文件server.xml中配置Http11NioProtocol协议的示例

<Connector connectionTimeout="20000" maxThreads="1000" port="8080"protocol="org.apache.coyote.http11.Http11NioProtocol"/>

tomcat趣谈之NIO源码详解（这回你总该懂了吧）

二 Endpoint:NioEndpoint（上图new出来的）封装了请求和处理的实现，在调用start方法时候绑定，后面也会讲到start方法

tomcat趣谈之NIO源码详解（这回你总该懂了吧）

三：Acceptor:实现了Runnable接口，接收外部的请求并封装成channel

tomcat趣谈之NIO源码详解（这回你总该懂了吧）

将scoket封装成channel交给poller数组中的一个进行注册

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.poll();
 //中间省略。。。
        getPoller0().register(channel);

四 Poller: poll，epoll，selector三种模式之一

tomcat趣谈之NIO源码详解（这回你总该懂了吧）

我们先看一下poller的注册渠道方法：注意最后一行

public void register(final NioChannel socket) {
    socket.setPoller(this);
    KeyAttachment key = keyCache.poll();
    final KeyAttachment ka = key!=null?key:new KeyAttachment(socket);
    ka.reset(this,socket,getSocketProperties().getSoTimeout());
    ka.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
    ka.setSecure(isSSLEnabled());
    PollerEvent r = eventCache.poll();
    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);封装成pollevent 画重点
}

再看一下它的run方法：

@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);

再看一下代码中的events()方法：r.run

public boolean events() {
    boolean result = false;
    Runnable r = null;
    while ( (r = events.poll()) != null ) {
        result = true;
        try {
            r.run();
            if ( r instanceof PollerEvent ) {
                ((PollerEvent)r).reset();
                eventCache.offer((PollerEvent)r);
            }
        } catch ( Throwable x ) {
            log.error("",x);
        }
    }
    return result;
}

五最后我们来看一下r也就是封装成的pollevent到底是何方神圣：将通道channel和自己的selector进行关联监听

tomcat趣谈之NIO源码详解（这回你总该懂了吧）

那么问题来了，这里只是对通道进行关联监听，怎么实现最后的方法调用呢？

不用急，我们继续往下看上图poller的events()方法之后马上接着一个timeout方法，我们接着看下timeout方法是干嘛的：

protected void timeout(int keyCount, boolean hasEvents) {
    long now = System.currentTimeMillis();
    // This method is called on every loop of the Poller. Don't process
    // timeouts on every loop of the Poller since that would create too
    // much load and timeouts can afford to wait a few seconds.
    // However, do process timeouts if any of the following are true:
    // - the selector simply timed out (suggests there isn't much load)
    // - the nextExpiration time has passed
    // - the server socket is being closed
    if ((keyCount > 0 || hasEvents) && (now < nextExpiration) && !close) {
        return;
    }
    //timeout
    Set<SelectionKey> keys = selector.keys();
    int keycount = 0;
    try {
        for (Iterator<SelectionKey> iter = keys.iterator(); iter.hasNext();) {
            SelectionKey key = iter.next();
            keycount++;
            try {
                KeyAttachment ka = (KeyAttachment) key.attachment();
                if ( ka == null ) {
                    cancelledKey(key, SocketStatus.ERROR,false); //we don't support any keys without attachments
                } else if ( ka.getError() ) {
                    cancelledKey(key, SocketStatus.ERROR,true);//TODO this is not yet being used
                } else if (ka.isComet() && ka.getCometNotify() ) {
                    ka.setCometNotify(false);
                    reg(key,ka,0);//avoid multiple calls, this gets reregistered after invocation
                    //if (!processSocket(ka.getChannel(), SocketStatus.OPEN_CALLBACK)) processSocket(ka.getChannel(), SocketStatus.DISCONNECT);
                    if (!processSocket(ka.getChannel(), SocketStatus.OPEN_READ, true))

最后一行的processSocket方法：将监听的渠道转化成架构图中对应的SocketProcessor（就是tomcat调用方法的处理类，暂不用管）

tomcat趣谈之NIO源码详解（这回你总该懂了吧）

最后我们再来看看EndPoint的启动方法：首先它会绑定端口，其次就是会初始化poller和acceptor，这三者的关系我会在最后做一个总结，你只需要先了解个大概

public void startInternal() throws Exception {

    if (!running) {
        running = true;
        paused = false;
        // Create worker collection
        if ( getExecutor() == null ) {
            createExecutor();
        }
        initializeConnectionLatch();
        // Start poller threads
        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();
    }
}

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);
        t.setPriority(getAcceptorThreadPriority());
        t.setDaemon(getDaemon());
        t.start();
    }
}

总结：

整个Endpoint就好比是一个公司，acceptor就好比是老板，它负责从外面接各种活，然后就把活交给一休，也就是poller来处理，那一休后面的小弟又是什么呢，其实就是endpoint启动时候创建的线程池

public void startInternal() throws Exception {
    if (!running) {
        running = true;
        paused = false;
        // Create worker collection
        if ( getExecutor() == null ) {
            createExecutor();
        }

那小弟又是如何安排活的呢：上图封装channel通道为tomcat的处理类时，藏得比较深，难怪老板拖了三个月

if ( dispatch && getExecutor()!=null ) getExecutor().execute(sc);

其实老板可以是多个，一休类似的员工也可以是多个，就好比一个创业公司上市了有了多个股东，但是这里面的多个员工在endpoint是多个老板共用的哦，哎可怜的一休！！！

protected Poller[] pollers = null;

无论是acceptor还是poller都是实现了runnable接口，也在endpoint启动的时候跟着start了，那在看他们的关系的时候就要动态的去看他们的run方法及一些关联的字段比如存放pollerevent的eventCache，注意它的offer和put方法的使用地方，读者可以根据总结返回去再看一下截图中的源码细节

那么这一期的tomcat的nio源码解析就到此结束了，其实这里面还有很多东西可以去研究，比如说老板最多能接多少客人（