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

Posted 2021-06-16 学习王国

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

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

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

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

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

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

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

      

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

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

将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三种模式之一

我们先看一下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进行关联监听

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

不用急，我们继续往下看上图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调用方法的处理类，暂不用管）

最后我们再来看看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源码解析就到此结束了，其实这里面还有很多东西可以去研究，比如说老板最多能接多少客人（

private int maxConnections = 10000;

），一休最多能有多少小弟（

protected int maxThreads = 200;

），这些在endpoint初始化的时候就已经决定了，如果在面试中能够说出来这些默认值应该可以把面试官唬的翻白眼吧，今天的tomcat源码解析先到这了，后面还会做各种框架的源码解析，敬请期待哦！！！