深度解读 Tomcat 中的 NIO 模型

I/O 复用模型
TOMCAT NIO

I/O 复用模型

Tomcat 的 NIO 是基于 I/O 复用来实现的：

内核级别

操作系统为代码所处的特权级别分了 4 个级别。不过现代操作系统只用到了 0 和 3 两个级别。0 和 3 的切换就是用户态和内核态的切换：

如果进程执行的是用户代码，当前进程处在用户态
如果进程执行的是内核里面的代码，当前进程处在内核态

同步非阻塞

I/O 复用模型，是 同步非阻塞 ：

非阻塞：I/O 读写，对应的是 recvfrom 操作，因为数据报文已经准备好，无需阻塞
同步：执行是在一个进程里面执行的
阻塞：阻塞在 select 上面，必须等到读就绪、写就绪等网络事件有时候
多路复用：
- 多路：N 个连接，每一个连接对应一个 channel，或者说多路就是多个 channel
- 复用：多个连接复用了一个线程或者少量线程 (在 Tomcat 中是 Math.min(2,Runtime.getRuntime().availableProcessors()) 个线程)

网络事件：

连接就绪
接收就绪
读就绪
写就绪

Selector复用器

I/O 复用主要是通过 Selector 复用器 来实现的：

TOMCAT NIO

对 IO 模型的支持

tomcat 从 6 以后开始支持 NIO 模型，实现是基于 JDK 的 java.nio 包。可以看到对 read body 和 response body 是 Blocking 的

NIO 配置

在 Connector 节点配置：

protocol="org.apache.coyote.http11.Http11NioProtocol"
maxConnections=10000
maxThreads=50

Http11NioProtocol 协议下默认最大连接数是 10000，也可以重新修改 maxConnections 的值，同时可以设置最大线程数 maxThreads，这里设置的最大线程数就是 Excutor 的线程池的大小

在 BIO 模式下实际上是没有 maxConnections，即使配置也不会生效

BIO 模式下的 maxConnections 是保持跟 maxThreads 大小一致，因为它是一请求一线程模式

组件

NioEndpoint 一共包含 LimitLatch、Acceptor、Poller、SocketProcessor、Excutor5 个部分：

LimitLatch ：连接控制器，它负责维护连接数的计算，nio 模式下默认是 10000，达到这个阈值后，就会拒绝连接请求
Acceptor ：负责接收连接，默认是 1 个线程来执行，将请求的事件注册到事件列表
Poller：来负责轮询，Poller 线程数量是 cpu 的核数 Math.min(2,Runtime.getRuntime().availableProcessors())
SocketProcessor：由 Poller 将就绪的事件生成 SocketProcessor ，同时交给 Excutor 去执行
Excutor ：线程池的大小就是在 Connector 节点配置的 maxThreads 的值。在 Excutor 的线程中：
1. 从 socket 中读取 http request
2. 解析成 HttpServletRequest 对象
3. 分派到相应的 servlet 并完成逻辑
4. 将 response 通过 socket 发回 client

在从 socket 中读数据和往 socket 中写数据的过程，并没有像典型的非阻塞的 NIO 的那样，注册 OP_READ 或 OP_WRITE 事件到主 Selector

而是 直接通过 socket 完成读写，这时是阻塞完成的

但是在 timeout 控制上，使用了 NIO 的 Selector 机制

这个 Selector 并不是 Poller 线程维护的主 Selector，而是 BlockPoller 线程中维护的 Selector，称之为辅 Selector

源码解读

NioEndpoint 执行序列图如下：

0: 初始化

无论是 BIO 还是 NIO，开始都会初始化连接限制，不可能无限增大，NIO 模式下默认是 10000：

public void startInternal() throws Exception {

        if (!running) {
                //省略代码...
                initializeConnectionLatch();
                //省略代码...
        }
}
protected LimitLatch initializeConnectionLatch() {
        if (maxConnections==-1) return null;
        if (connectionLimitLatch==null) {
                connectionLimitLatch = new LimitLatch(getMaxConnections());
        }
        return connectionLimitLatch;
}

1: 绑定 IP 地址及端口，将 ServerSocketChannel 设置为阻塞

Tomcat 的设计初衷主要是为了操作方便。这样这里就跟 BIO 模式下一样了。只不过在 BIO 下这里返回的是 Socket，NIO 下这里返回的是 SocketChannel：

public void bind() throws Exception {

        //省略代码...
        serverSock.socket().bind(addr,getBacklog());
        serverSock.configureBlocking(true); 
        //省略代码...
        selectorPool.open();
}

2: 启动接收线程

public void startInternal() throws Exception {
        if (!running) {
                //省略代码...
                startAcceptorThreads();
        }
}

//这个方法实际是在它的超类 AbstractEndpoint 里面    
protected final void startAcceptorThreads() {
        int count = getAcceptorThreadCount();
        acceptors = new Acceptor[count];

        for (int i = 0; i < count; i++) {
                acceptors[i] = createAcceptor();
                Thread t = new Thread(acceptors[i], getName() + "-Acceptor-" + i);
                t.setPriority(getAcceptorThreadPriority());
                t.setDaemon(getDaemon());
                t.start();
        }
}

3: ServerSocketChannel.accept() 接收新连接

protected class Acceptor extends AbstractEndpoint.Acceptor {
        @Override
        public void run() {
                while (running) {
                        try {
                                //省略代码...
                                SocketChannel socket = null;
                                try {                        
                                        socket = serverSock.accept();//接收新连接 
                                } catch (IOException ioe) {
                                        //省略代码...
                                        throw ioe;
                                }
                                //省略代码...
                                if (running && !paused) {
                                        if (!setSocketOptions(socket)) {
                                                //省略代码...
                                        }
                                } else {
                                        //省略代码...
                                }
                        } catch (SocketTimeoutException sx) {

                        } catch (IOException x) {
                                //省略代码...
                        } catch (OutOfMemoryError oom) {
                                //省略代码...
                        } catch (Throwable t) {
                                //省略代码...
                        }
                }

        }
}

4: 将接收到的链接通道设置为非阻塞

5: 构造 NioChannel 对象

6: register 注册到轮询线程

protected boolean setSocketOptions(SocketChannel socket) {        
        try {
                socket.configureBlocking(false);//将连接通道设置为非阻塞 
                Socket sock = socket.socket();
                socketProperties.setProperties(sock);

                NioChannel channel = nioChannels.poll();//构造 NioChannel 对象 
                //省略代码...
                getPoller0().register(channel);//register 注册到轮询线程 
        } catch (Throwable t) {
                //省略代码...
        }
        //省略代码...
}

7: 构造 PollerEvent，并添加到事件队列

protected ConcurrentLinkedQueue<Runnable> events = new ConcurrentLinkedQueue<Runnable>();
public void register(final NioChannel socket) {
        //省略代码...
        PollerEvent r = eventCache.poll();
        //省略代码...
        addEvent(r);
}

8: 启动轮询线程

public void startInternal() throws Exception {

        if (!running) {
                //省略代码...
                // 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();
                }
                //省略代码...
        }
}

9: 取出队列中新增的 PollerEvent 并注册到 Selector

public static class PollerEvent implements Runnable {
        //省略代码...
        @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 {
                        //省略代码...
                }//end if
        }//run
        //省略代码...
}

10: Selector.select()

public void run() {
        // Loop until destroy() is called
        while (true) {
                try {
                        //省略代码...
                        try {
                                if ( !close ) {
                                        if (wakeupCounter.getAndSet(-1) > 0) {
                                                keyCount = selector.selectNow();
                                        } else {
                                                keyCount = selector.select(selectorTimeout);
                                        }
                                        //省略代码...
                                }
                                //省略代码...
                        } catch ( NullPointerException x ) {
                                //省略代码...
                        } catch ( CancelledKeyException x ) {
                                //省略代码...
                        } catch (Throwable x) {
                                //省略代码...
                        }
                        //省略代码...

                        Iterator<SelectionKey> iterator =
                                keyCount > 0 ? selector.selectedKeys().iterator() : null;

                        while (iterator != null && iterator.hasNext()) {
                                SelectionKey sk = iterator.next();
                                KeyAttachment attachment = (KeyAttachment)sk.attachment();

                                if (attachment == null) {
                                        iterator.remove();
                                } else {
                                        attachment.access();
                                        iterator.remove();
                                        processKey(sk, attachment);//此方法跟下去就是把 SocketProcessor 交给 Excutor 去执行 
                                }
                        }//while

                        //省略代码...
                } catch (OutOfMemoryError oom) {
                        //省略代码...
                }
        }//while
        //省略代码...
}

11: 根据选择的 SelectionKey 构造 SocketProcessor 提交到请求处理线程

public boolean processSocket(NioChannel socket, SocketStatus status, boolean dispatch) {
        try {
                //省略代码...
                SocketProcessor sc = processorCache.poll();
                if ( sc == null )
                        sc = new SocketProcessor(socket,status);
                else
                        sc.reset(socket,status);
                if ( dispatch && getExecutor()!=null )
                        getExecutor().execute(sc);
                else
                        sc.run();
        } catch (RejectedExecutionException rx) {
                //省略代码...
        } catch (Throwable t) {
                //省略代码...
        }
        //省略代码...
}

NioBlockingSelector 和 BlockPoller

NioEndpoint 对象中维护了一个 NioSelecPool 对象，这个 NioSelectorPool 中又维护了一个 BlockPoller 线程，这个线程就是基于辅 Selector 进行 NIO 的逻辑：执行 servlet 后，得到 response，往 socket 中写数据为例，最终写的过程调用 NioBlockingSelector 的 write 方法。代码如下：

public int write(ByteBuffer buf, NioChannel socket, long writeTimeout,MutableInteger lastWrite) throws IOException {  
        SelectionKey key = socket.getIOChannel().keyFor(socket.getPoller().getSelector());  
        if ( key == null )
                throw new IOException("Key no longer registered");  
        KeyAttachment att = (KeyAttachment) key.attachment();  
        int written = 0;  
        boolean timedout = false;  
        int keycount = 1; //assume we can write  
        long time = System.currentTimeMillis(); //start the timeout timer  
        try {  
                while ( (!timedout) && buf.hasRemaining()) {  
                        if (keycount > 0) { //only write if we were registered for a write  
                                //直接往 socket 中写数据  
                                int cnt = socket.write(buf); //write the data  
                                lastWrite.set(cnt);  
                                if (cnt == -1)  
                                        throw new EOFException();  
                                written += cnt;  
                                //写数据成功，直接进入下一次循环，继续写  
                                if (cnt > 0) {  
                                        time = System.currentTimeMillis(); //reset our timeout timer  
                                        continue; //we successfully wrote, try again without a selector  
                                }  
                        }  
                        //如果写数据返回值 cnt 等于 0，通常是网络不稳定造成的写数据失败  
                        try {  
                                //开始一个倒数计数器   
                                if ( att.getWriteLatch()==null || att.getWriteLatch().getCount()==0)
                                        att.startWriteLatch(1);  
                                //将 socket 注册到辅 Selector，这里 poller 就是 BlockSelector 线程
                                poller.add(att,SelectionKey.OP_WRITE);  
                                //阻塞，直至超时时间唤醒，或者在还没有达到超时时间，在 BlockSelector 中唤醒  
                                att.awaitWriteLatch(writeTimeout,TimeUnit.MILLISECONDS);  
                        }catch (InterruptedException ignore) {  
                                Thread.interrupted();  
                        }  
                        if ( att.getWriteLatch()!=null && att.getWriteLatch().getCount()> 0) {  
                                keycount = 0;  
                        }else {  
                                //还没超时就唤醒，说明网络状态恢复，继续下一次循环，完成写 socket  
                                keycount = 1;  
                                att.resetWriteLatch();  
                        }  

                        if (writeTimeout > 0 && (keycount == 0))  
                                timedout = (System.currentTimeMillis() - time) >= writeTimeout;  
                } //while  
                if (timedout)   
                        throw new SocketTimeoutException();  
        } finally {  
                poller.remove(att,SelectionKey.OP_WRITE);  
                if (timedout && key != null) {  
                        poller.cancelKey(socket, key);  
                }  
        }  
        return written;  
}

当 socket.write() 返回 0 时，说明网络状态不稳定，这时将 socket 注册 OP_WRITE 事件到辅 Selector，由 BlockPoller 线程不断轮询这个辅 Selector，直到发现这个 socket 的写状态恢复了，通过那个倒数计数器，通知 Worker 线程继续写 socket 动作。看一下 BlockSelector 线程的代码逻辑：

public void run() {  
        while (run) {  
                try {  
                        ......  

                        Iterator iterator = keyCount > 0 ? selector.selectedKeys().iterator() : null;  
                        while (run && iterator != null && iterator.hasNext()) {  
                                SelectionKey sk = (SelectionKey) iterator.next();  
                                KeyAttachment attachment = (KeyAttachment)sk.attachment();  
                                try {  
                                        attachment.access();  
                                        iterator.remove(); ;  
                                        sk.interestOps(sk.interestOps() & (~sk.readyOps()));  
                                        if ( sk.isReadable() ) {  
                                                countDown(attachment.getReadLatch());  
                                        }  
                                        //发现 socket 可写状态恢复，将倒数计数器置位，通知 Worker 线程继续  
                                        if (sk.isWritable()) {  
                                                countDown(attachment.getWriteLatch());  
                                        }  
                                }catch (CancelledKeyException ckx) {  
                                        if (sk!=null) sk.cancel();  
                                        countDown(attachment.getReadLatch());  
                                        countDown(attachment.getWriteLatch());  
                                }  
                        }//while  
                }catch ( Throwable t ) {  
                        log.error("",t);  
                }  
        }  
        events.clear();  
        try {  
                selector.selectNow();//cancel all remaining keys  
        }catch( Exception ignore ) {  
                if (log.isDebugEnabled())log.debug("",ignore);  
        }  
}

使用这个辅 Selector 主要是减少线程间的切换，同时还可减轻主 Selector 的负担