欢迎您的访问
专注架构,Java,数据结构算法,Python技术分享

Tomcat NIO 处理分析

Tomcat 的 Connector 有三种运行模式 bio、nio、apr ,先了解一下这三种的区别。

1、 bio(blocking I/O),顾名思义即阻塞式 I/O 操作,表示 Tomcat 使用的是传统的 Java I/O 操作(即java.io包及其子包)。Tomcat 在默认情况下,就是以 bio 模式运行的。一般而言 bio 模式是三种运行模式中性能最低的一种。
2、 nio(new I/O),是 Java SE 1.4 及后续版本提供的一种新的 I/O 操作方式(即java.nio包及其子包)。Java nio 是一个基于缓冲区并能提供非阻塞 I/O 操作的 Java API ,因此 nio 也被看成是 non-blocking I/O 的缩写。它拥有比传统 I/O 操作( bio )更好的并发运行性能。要让 Tomcat 以 nio 模式来运行只需要在 Tomcat 安装目录/conf/server.xml文件中将 Connector 节点的 protocol 配置成org.apache.coyote.http11.Http11NioProtocol即可。
3、 apr( Apache Portable Runtime/Apache 可移植运行时),是 Apache HTTP 服务器的支持库。可以简单地理解为 Tomcat 将以 JNI 的形式调 用Apache HTTP 服务器的核心动态链接库来处理文件读取或网络传输操作,从而大大提高 Tomcat 对静态文件的处理性能。 Tomcat apr 也是在 Tomcat 上运行高并发应用的首选模式。

写个 BIO 的 Socket 服务器还是比较容易的,无非是每 accept 一个 socket 之后就扔到一个线程中处理请求生成响应,这种方式可以改进的点就是增加线程池的支持,本文主要分析一下 Tomcat 中 NIO 处理方式的相关代码逻辑。

关键代码都是在org.apache.tomcat.util.net.NioEndpoint这个类里面,它是 Http11NioProtocol 中负责接收处理 socket 的主要组件,别看代码很长,仔细阅读会发现有很多共通的地方,如:

1、 都会对JDK中原有的API做一下扩展或者包装,如 ThreadPoolExecutor 是对java.util.concurrent.ThreadPoolExecutor的扩展,NioChannel 是对 ByteChannel 的扩展,KeyAttachment 则是对 NioChannel 的包装
2、 很多类设计成非 GC 的,方便缓存和重复使用,实现方式都是通过 ConcurrentLinkedQueue 类构造一个队列。比如 NioEndpoint 类里面的 ConcurrentLinkedQueue processorCache、ConcurrentLinkedQueue keyCache、ConcurrentLinkedQueue eventCache、ConcurrentLinkedQueue nioChannels 。Poller 类里面的 ConcurrentLinkedQueue events

先看下整个 Connector 组件结构图:

63_1.png

看过之前 Tomcat 启动文章的应该都知道,Connector 的启动会调用 Connector 类的 startInternal 方法,里面调用了 protocolHandler 的 start() ,该方法中将调用抽象的 endpoint 的 start() 方法,这个方法会调用到具体 Endpoint 类的 startInternal() ,所以代码分析先从 NioEndpoint 类的 startInternal 看起。

1.NioEndpoint 类核心组件的初始化

/** 
 * Start the NIO endpoint, creating acceptor, poller threads. 
 */  
@Override  
public void startInternal() throws Exception {  

    if (!running) {  
        running = true;  
        paused = false;  

        // Create worker collection  
        if ( getExecutor() == null ) {  
            // 构造线程池,用于后续执行SocketProcessor线程,这就是上图中的Worker。  
            createExecutor();  
        }  

        initializeConnectionLatch();  

        // Start poller threads  
        // 根据处理器数量构造一定数目的轮询器,即上图中的Poller  
        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();  
        }  

        // 创建接收者线程,即上图中的Acceptor  
        startAcceptorThreads();  
    }  
}  

startAcceptorThreads 调用的是父类org.apache.tomcat.util.net.AbstractEndpoint中的实现:

protected final void startAcceptorThreads() {  
    int count = getAcceptorThreadCount();  
    acceptors = new Acceptor[count];  

    for (int i = 0; i < count; i++) {  
        // 调用子类的createAcceptor方法,本例中即NioEndpoint类的createAcceptor方法  
        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();  
    }  
}

以上就是 Acceptor、Poller、Worker 等核心组件的初始化过程。

2.请求接收

核心组件初始化之后接着就是 Acceptor 线程接收 socket 连接,看下 Acceptor 的源码:


// --------------------------------------------------- Acceptor Inner Class  
/** 
 * 后台线程,用于监听TCP/IP连接以及将它们分发给相应的调度器处理。 
 * The background thread that listens for incoming TCP/IP connections and 
 * hands them off to an appropriate processor. 
 */  
protected class Acceptor extends AbstractEndpoint.Acceptor {  

    @Override  
    public void run() {  

        int errorDelay = 0;  

        // 循环遍历直到接收到关闭命令  
        // Loop until we receive a shutdown command  
        while (running) {  

            // Loop if endpoint is paused  
            while (paused && running) {  
                state = AcceptorState.PAUSED;  
                try {  
                    Thread.sleep(50);  
                } catch (InterruptedException e) {  
                    // Ignore  
                }  
            }  

            if (!running) {  
                break;  
            }  
            state = AcceptorState.RUNNING;  

            try {  
                // 如果已经达到最大连接数则让线程等待  
                //if we have reached max connections, wait  
                countUpOrAwaitConnection();  

                SocketChannel socket = null;  
                try {  
                    // 接收连接,这里用的阻塞模式。  
                    // Accept the next incoming connection from the server  
                    // socket  
                    socket = serverSock.accept();  
                } catch (IOException ioe) {  
                    //we didn't get a socket  
                    countDownConnection();  
                    // Introduce delay if necessary  
                    errorDelay = handleExceptionWithDelay(errorDelay);  
                    // re-throw  
                    throw ioe;  
                }  
                // Successful accept, reset the error delay  
                errorDelay = 0;  

                // 注意这个setSocketOptions方法  
                // 它将把上面接收到的socket添加到轮询器Poller中  
                // setSocketOptions() will add channel to the poller  
                // if successful  
                if (running && !paused) {  
                    if (!setSocketOptions(socket)) {  
                        countDownConnection();  
                        closeSocket(socket);  
                    }  
                } else {  
                    countDownConnection();  
                    closeSocket(socket);  
                }  
            } catch (SocketTimeoutException sx) {  
                // Ignore: Normal condition  
            } catch (IOException x) {  
                if (running) {  
                    log.error(sm.getString("endpoint.accept.fail"), x);  
                }  
            } catch (OutOfMemoryError oom) {  
                try {  
                    oomParachuteData = null;  
                    releaseCaches();  
                    log.error("", oom);  
                }catch ( Throwable oomt ) {  
                    try {  
                        try {  
                            System.err.println(oomParachuteMsg);  
                            oomt.printStackTrace();  
                        }catch (Throwable letsHopeWeDontGetHere){  
                            ExceptionUtils.handleThrowable(letsHopeWeDontGetHere);  
                        }  
                    }catch (Throwable letsHopeWeDontGetHere){  
                        ExceptionUtils.handleThrowable(letsHopeWeDontGetHere);  
                    }  
                }  
            } catch (Throwable t) {  
                ExceptionUtils.handleThrowable(t);  
                log.error(sm.getString("endpoint.accept.fail"), t);  
            }  
        }  
        state = AcceptorState.ENDED;  
    }  
}

3.Socket 参数设置

在 Acceptor 里接收到一个连接之后调用 setSocketOptions 方法设置 SocketChannel 的一些参数,然后将 SocketChannel 注册到 Poller 中。看下 setSocketOptions 的实现:

/** 
 * Process the specified connection. 
 */  
protected boolean setSocketOptions(SocketChannel socket) {  
    // Process the connection  
    try {  
        // 将SocketChannel配置为非阻塞模式  
        //disable blocking, APR style, we are gonna be polling it  
        socket.configureBlocking(false);  
        Socket sock = socket.socket();  
        // 设置Socket参数值(从server.xml的Connector节点上获取参数值)  
        // 比如Socket发送、接收的缓存大小、心跳检测等  
        socketProperties.setProperties(sock);  

        // 从NioChannel的缓存队列中取出一个NioChannel  
        // NioChannel是SocketChannel的一个的包装类  
        // 这里对上层屏蔽SSL和一般TCP连接的差异  
        NioChannel channel = nioChannels.poll();  

        // 缓存队列中没有则新建一个NioChannel  
        if ( channel == null ) {  
            // SSL setup  
            if (sslContext != null) {  
                SSLEngine engine = createSSLEngine();  
                int appbufsize = engine.getSession().getApplicationBufferSize();  
                NioBufferHandler bufhandler = new NioBufferHandler(Math.max(appbufsize,socketProperties.getAppReadBufSize()),  
                                                                   Math.max(appbufsize,socketProperties.getAppWriteBufSize()),  
                                                                   socketProperties.getDirectBuffer());  
                channel = new SecureNioChannel(socket, engine, bufhandler, selectorPool);  
            } else {  
                // normal tcp setup  
                NioBufferHandler bufhandler = new NioBufferHandler(socketProperties.getAppReadBufSize(),  
                                                                   socketProperties.getAppWriteBufSize(),  
                                                                   socketProperties.getDirectBuffer());  

                channel = new NioChannel(socket, bufhandler);  
            }  
        } else {  
            // 将SocketChannel关联到从缓存队列中获取的NioChannel上来  
            channel.setIOChannel(socket);  
            if ( channel instanceof SecureNioChannel ) {  
                SSLEngine engine = createSSLEngine();  
                ((SecureNioChannel)channel).reset(engine);  
            } else {  
                channel.reset();  
            }  
        }  
        // 将新接收到的SocketChannel注册到Poller中  
        getPoller0().register(channel);  
    } catch (Throwable t) {  
        ExceptionUtils.handleThrowable(t);  
        try {  
            log.error("",t);  
        } catch (Throwable tt) {  
            ExceptionUtils.handleThrowable(t);  
        }  
        // Tell to close the socket  
        return false;  
    }  
    return true;  
}

核心调用是最后的getPoller0().register(channel);它将配置好的 SocketChannel 包装成一个 PollerEvent ,然后加入到 Poller 的 events 缓存队列中。

4.读取事件注册

getPoller0 方法将轮询当前的 Poller 数组,从中取出一个 Poller 返回。( Poller 的初始化参见上述第1步:NioEndpoint 类核心组件的初始化)

/** 
 * Return an available poller in true round robin fashion 
 */  
public Poller getPoller0() {  
    // 最简单的轮询调度算法,poller的计数器不断加1再对poller数组取余数  
    int idx = Math.abs(pollerRotater.incrementAndGet()) % pollers.length;  
    return pollers[idx];  
}  

之后调用 Poller 对象的 register 方法:

        public void register(final NioChannel socket) {
            // 设置socket的Poller引用,便于后续处理
            socket.setPoller(this);
            // 从NioEndpoint的keyCache缓存队列中取出一个KeyAttachment
            KeyAttachment key = keyCache.poll();
            // KeyAttachment实际是NioChannel的包装类
            final KeyAttachment ka = key!=null?key:new KeyAttachment(socket);
            // 重置KeyAttachment对象中Poller、NioChannel等成员变量的引用
            ka.reset(this,socket,getSocketProperties().getSoTimeout());
            ka.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
            ka.setSecure(isSSLEnabled());

            // 从Poller的事件对象缓存中取出一个PollerEvent,并用socket初始化事件对象
            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);

            // 加入到Poller对象里的事件队列
            addEvent(r);
        }

看下 Poller 类里 addEvent 的代码:

        /**
         * Only used in this class. Will be made private in Tomcat 8.0.x
         * @deprecated
         */
        @Deprecated
        public void addEvent(Runnable event) {
            events.offer(event);
            if ( wakeupCounter.incrementAndGet() == 0 ) selector.wakeup();
        }

就两行,第一行从 event 对象添加到缓存队列中,第二行如果当前事件队列中没有事件,则唤醒处于阻塞状态的 selector 。

5.Poller 处理流程

上面讲的是从 Acceptor 中接收到的 Socket 以 PollerEvent 的形式包装并添加到 Poller 的事件缓存队列中,接下来看看另外一个核心组件 Poller 的处理过程:


    /**
     * Poller class.
     */
    public class Poller implements Runnable {

        // 这就是NIO中用到的选择器,可以看出每一个Poller都会关联一个Selector
        protected Selector selector;
        // 待处理的事件队列
        protected ConcurrentLinkedQueue events = new ConcurrentLinkedQueue();

        // 唤醒多路复用器的条件阈值
        protected AtomicLong wakeupCounter = new AtomicLong(0l);

        public Poller() throws IOException {
            // 对Selector的同步访问,通过调用Selector.open()方法创建一个Selector
            synchronized (Selector.class) {
                // Selector.open() isn't thread safe
                // http://bugs.sun.com/view_bug.do?bug_id=6427854
                // Affects 1.6.0_29, fixed in 1.7.0_01
                this.selector = Selector.open();
            }
        }

        // 通过addEvent方法将事件添加到Poller的事件队列中
        /**
         * Only used in this class. Will be made private in Tomcat 8.0.x
         * @deprecated
         */
        @Deprecated
        public void addEvent(Runnable event) {
            events.offer(event);
            // 如果队列中没有待处理的事件则唤醒处于阻塞状态的selector
            if ( wakeupCounter.incrementAndGet() == 0 ) selector.wakeup();
        }

        // 处理事件队列中的所有事件,如果事件队列是空的则返回false
        /**
         * Processes events in the event queue of the Poller.
         *
         * @return true if some events were processed,
         *   false if queue was empty
         */
        public boolean events() {
            boolean result = false;

            Runnable r = null;
            // 将Poller的事件队列中的事件逐个取出并执行相应的事件线程
            while ( (r = events.poll()) != null ) {
                result = true;
                try {
                    // 执行事件处理逻辑
                    // 这里将事件设计成线程是将具体的事件处理逻辑和事件框架分开 
                    r.run();
                    if ( r instanceof PollerEvent ) {
                        ((PollerEvent)r).reset();
                        // 事件处理完之后,将事件对象返回NIOEndpoint的事件对象缓存中
                        eventCache.offer((PollerEvent)r);
                    }
                } catch ( Throwable x ) {
                    log.error("",x);
                }
            }

            return result;
        }

        // 将socket包装成统一的事件对象PollerEvent,加入到待处理事件队列中
        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());
            // 从NIOEndpoint的事件对象缓存中取出一个事件对象
            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);
            // 将事件添加打Poller的事件队列中
            addEvent(r);
        }

        // Poller是一个线程,该线程同Acceptor一样会监听TCP/IP连接并将它们交给合适的处理器处理
        /**
         * The background thread that listens for incoming TCP/IP connections and
         * hands them off to an appropriate processor.
         */
        @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);
                        try {
                            selector.close();
                        } catch (IOException ioe) {
                            log.error(sm.getString(
                                    "endpoint.nio.selectorCloseFail"), ioe);
                        }
                        break;
                    } else {
                        // 执行事件队列中的事件线程
                        hasEvents = events();
                    }
                    try {
                        if ( !close ) {
                            if (wakeupCounter.getAndSet(-1) > 0) {
                                // 把wakeupCounter设成-1,这是与addEvent里的代码呼应,这样会唤醒selector
                                //if we are here, means we have other stuff to do
                                //do a non blocking select
                                // 以非阻塞方式查看selector是否有事件发生
                                keyCount = selector.selectNow();
                            } else {
                                // 查看selector是否有事件发生,超过指定时间则立即返回
                                keyCount = selector.select(selectorTimeout);
                            }
                            wakeupCounter.set(0);
                        }
                        if (close) {
                            // 执行事件队列中的事件线程
                            events();
                            timeout(0, false);
                            try {
                                selector.close();
                            } catch (IOException ioe) {
                                log.error(sm.getString(
                                        "endpoint.nio.selectorCloseFail"), ioe);
                            }
                            break;
                        }
                    } catch ( NullPointerException x ) {
                        //sun bug 5076772 on windows JDK 1.5
                        if ( log.isDebugEnabled() ) log.debug("Possibly encountered sun bug 5076772 on windows JDK 1.5",x);
                        if ( wakeupCounter == null || selector == null ) throw x;
                        continue;
                    } catch ( CancelledKeyException x ) {
                        //sun bug 5076772 on windows JDK 1.5
                        if ( log.isDebugEnabled() ) log.debug("Possibly encountered sun bug 5076772 on windows JDK 1.5",x);
                        if ( wakeupCounter == null || selector == null ) throw x;
                        continue;
                    } catch (Throwable x) {
                        ExceptionUtils.handleThrowable(x);
                        log.error("",x);
                        continue;
                    }
                    //either we timed out or we woke up, process events first
                    if ( keyCount == 0 ) hasEvents = (hasEvents | events());

                    Iterator iterator =
                        keyCount > 0 ? selector.selectedKeys().iterator() : null;
                    // 根据向selector中注册的key遍历channel中已经就绪的keys,并处理这些key
                    // Walk through the collection of ready keys and dispatch
                    // any active event.
                    while (iterator != null && iterator.hasNext()) {
                        SelectionKey sk = iterator.next();
                        // 这里的attachment方法返回的就是在register()方法中注册的
                        // 而KeyAttachment对象是对socket的包装
                        KeyAttachment attachment = (KeyAttachment)sk.attachment();
                        // Attachment may be null if another thread has called
                        // cancelledKey()
                        if (attachment == null) {
                            iterator.remove();
                        } else {
                            // 更新通道最近一次发生事件的时间
                            // 防止因超时没有事件发生而被剔除出selector
                            attachment.access();
                            iterator.remove();
                            // 具体处理通道的逻辑
                            processKey(sk, attachment);
                        }
                    }//while

                    //process timeouts
                    // 多路复用器每执行一遍完整的轮询便查看所有通道是否超时
                    // 对超时的通道将会被剔除出多路复用器
                    timeout(keyCount,hasEvents);
                    if ( oomParachute > 0 && oomParachuteData == null ) checkParachute();
                } catch (OutOfMemoryError oom) {
                    try {
                        oomParachuteData = null;
                        releaseCaches();
                        log.error("", oom);
                    }catch ( Throwable oomt ) {
                        try {
                            System.err.println(oomParachuteMsg);
                            oomt.printStackTrace();
                        }catch (Throwable letsHopeWeDontGetHere){
                            ExceptionUtils.handleThrowable(letsHopeWeDontGetHere);
                        }
                    }
                }
            }//while
            synchronized (this) {
                this.notifyAll();
            }
            stopLatch.countDown();

        }

        // 处理selector检测到的通道事件 
        protected boolean processKey(SelectionKey sk, KeyAttachment attachment) {
            boolean result = true;
            try {
                if ( close ) {
                    cancelledKey(sk, SocketStatus.STOP, attachment.comet);
                } else if ( sk.isValid() && attachment != null ) {
                    // 确保通道不会因超时而被剔除
                    attachment.access();//make sure we don't time out valid sockets
                    sk.attach(attachment);//cant remember why this is here
                    NioChannel channel = attachment.getChannel();
                    // 处理通道发生的读写事件
                    if (sk.isReadable() || sk.isWritable() ) {
                        if ( attachment.getSendfileData() != null ) {
                            processSendfile(sk,attachment, false);
                        } else {
                            if ( isWorkerAvailable() ) {
                                // 在通道上注销对已经发生事件的关注
                                unreg(sk, attachment, sk.readyOps());
                                boolean closeSocket = false;
                                // Read goes before write
                                if (sk.isReadable()) {
                                    // 具体的通道处理逻辑
                                    if (!processSocket(channel, SocketStatus.OPEN_READ, true)) {
                                        closeSocket = true;
                                    }
                                }
                                if (!closeSocket && sk.isWritable()) {
                                    if (!processSocket(channel, SocketStatus.OPEN_WRITE, true)) {
                                        closeSocket = true;
                                    }
                                }
                                if (closeSocket) {
                                    // 解除无效通道
                                    cancelledKey(sk,SocketStatus.DISCONNECT,false);
                                }
                            } else {
                                result = false;
                            }
                        }
                    }
                } else {
                    //invalid key
                    cancelledKey(sk, SocketStatus.ERROR,false);
                }
            } catch ( CancelledKeyException ckx ) {
                cancelledKey(sk, SocketStatus.ERROR,false);
            } catch (Throwable t) {
                ExceptionUtils.handleThrowable(t);
                log.error("",t);
            }
            return result;
        }

        // 这个unreg()很巧妙,防止了通道对同一个事件不断select的问题
        protected void unreg(SelectionKey sk, KeyAttachment attachment, int readyOps) {
            //this is a must, so that we don't have multiple threads messing with the socket
            reg(sk,attachment,sk.interestOps()& (~readyOps));
        }

        // 向NioChannel注册感兴趣的事件,具体代码看下面的PollerEvent类的说明
        protected void reg(SelectionKey sk, KeyAttachment attachment, int intops) {
            sk.interestOps(intops);
            attachment.interestOps(intops);
            attachment.setCometOps(intops);
        }

    }

6.PollerEvent 处理流程

Poller 处理的核心是启动执行事件队列中的 PollerEvent,接着从 selector 中遍历已经就绪的 key ,一旦发生了感兴趣的事件,则交由 processSocket 方法处理。PollerEvent 的作用是向 socket 注册或更新感兴趣的事件:

    /**
     *
     * PollerEvent, cacheable object for poller events to avoid GC
     */
    public static class PollerEvent implements Runnable {

    // 每个PollerEvent都会保存NioChannel的引用
        protected NioChannel socket;
        protected int interestOps;
        protected KeyAttachment key;
        public PollerEvent(NioChannel ch, KeyAttachment k, int intOps) {
            reset(ch, k, intOps);
        }

        public void reset(NioChannel ch, KeyAttachment k, int intOps) {
            socket = ch;
            interestOps = intOps;
            key = k;
        }

        public void reset() {
            reset(null, null, 0);
        }

        @Override
        public void run() {
            //socket第一次注册到selector中,完成对socket读事件的注册
            if ( interestOps == OP_REGISTER ) {
                try {
                    socket.getIOChannel().register(socket.getPoller().getSelector(), SelectionKey.OP_READ, key);
                } catch (Exception x) {
                    log.error("", x);
                }
            } else {
                // socket之前已经注册到了selector中,更新socket所感兴趣的事件
                final SelectionKey key = socket.getIOChannel().keyFor(socket.getPoller().getSelector());
                try {
                    boolean cancel = false;
                    if (key != null) {
                        final KeyAttachment att = (KeyAttachment) key.attachment();
                        if ( att!=null ) {
                            //handle callback flag
                            if (att.isComet() && (interestOps & OP_CALLBACK) == OP_CALLBACK ) {
                                att.setCometNotify(true);
                            } else {
                                att.setCometNotify(false);
                            }
                            interestOps = (interestOps & (~OP_CALLBACK));//remove the callback flag
                            // 刷新事件的最后访问时间,防止事件超时 
                            att.access();//to prevent timeout
                            //we are registering the key to start with, reset the fairness counter.
                            int ops = key.interestOps() | interestOps;
                            att.interestOps(ops);
                            key.interestOps(ops);
                        } else {
                            cancel = true;
                        }
                    } else {
                        cancel = true;
                    }
                    if ( cancel ) socket.getPoller().cancelledKey(key,SocketStatus.ERROR,false);
                }catch (CancelledKeyException ckx) {
                    try {
                        socket.getPoller().cancelledKey(key,SocketStatus.DISCONNECT,true);
                    }catch (Exception ignore) {}
                }
            }//end if
        }//run

        @Override
        public String toString() {
            return super.toString()+"[intOps="+this.interestOps+"]";
        }
    }

7.将 socket 交给 Worker 执行

在第5步的 Poller 处理流程的分析中看到它的 run 方法最后会调用 processKey() 处理 selector 检测到的通道事件,而在这个方法最后会调用 processSocket 来调用具体的通道处理逻辑,看下 processSocket 方法的实现:

    public boolean processSocket(NioChannel socket, SocketStatus status, boolean dispatch) {
        try {
            KeyAttachment attachment = (KeyAttachment)socket.getAttachment();
            if (attachment == null) {
                return false;
            }
            attachment.setCometNotify(false); //will get reset upon next reg
            // 从SocketProcessor的缓存队列中取出一个来处理socket
            SocketProcessor sc = processorCache.poll();
            if ( sc == null ) sc = new SocketProcessor(socket,status);
            else sc.reset(socket,status);
            // 将有事件发生的socket交给Worker处理 
            if ( dispatch && getExecutor()!=null ) getExecutor().execute(sc);
            else sc.run();
        } catch (RejectedExecutionException rx) {
            log.warn("Socket processing request was rejected for:"+socket,rx);
            return false;
        } catch (Throwable t) {
            ExceptionUtils.handleThrowable(t);
            // This means we got an OOM or similar creating a thread, or that
            // the pool and its queue are full
            log.error(sm.getString("endpoint.process.fail"), t);
            return false;
        }
        return true;
    }

Poller 通过 NioEndpoint 的协调,将发生事件的 socket 交给工作者线程 Worker 来进一步处理。整个事件框架的工作就到此结束,下面就是 Worker 的处理。

8.从 socket 中处理请求

在 Tomcat 6 版本的 NIO 处理实现中有一个 Worker 类,在 Tomcat 7 中把它去掉了,但工作者的职责还在,只是交由了上面看到的 SocketProcessor 这个类来担当,看下这个类的实现代码:

    // ---------------------------------------------- SocketProcessor Inner Class
    // 这个类相当于一个工作者,但只会在一个外部线程池中简单使用。
    /**
     * This class is the equivalent of the Worker, but will simply use in an
     * external Executor thread pool.
     */
    protected class SocketProcessor implements Runnable {

        // 每个SocketProcessor保存一个NioChannel的引用
        protected NioChannel socket = null;
        protected SocketStatus status = null;

        public SocketProcessor(NioChannel socket, SocketStatus status) {
            reset(socket,status);
        }

        public void reset(NioChannel socket, SocketStatus status) {
            this.socket = socket;
            this.status = status;
        }

        @Override
        public void run() {
            // 从socket中获取SelectionKey
            SelectionKey key = socket.getIOChannel().keyFor(
                    socket.getPoller().getSelector());
            KeyAttachment ka = null;

            if (key != null) {
                ka = (KeyAttachment)key.attachment();
            }

            // Upgraded connections need to allow multiple threads to access the
            // connection at the same time to enable blocking IO to be used when
            // NIO has been configured
            if (ka != null && ka.isUpgraded() &&
                    SocketStatus.OPEN_WRITE == status) {
                synchronized (ka.getWriteThreadLock()) {
                    doRun(key, ka);
                }
            } else {
                synchronized (socket) {
                    doRun(key, ka);
                }
            }
        }

        private void doRun(SelectionKey key, KeyAttachment ka) {
            try {
                int handshake = -1;

                try {
                    if (key != null) {
                        // For STOP there is no point trying to handshake as the
                        // Poller has been stopped.
                        if (socket.isHandshakeComplete() ||
                                status == SocketStatus.STOP) {
                            handshake = 0;
                        } 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.
                            status = SocketStatus.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 (status == null) {
                        // 最关键的代码,这里将KeyAttachment(实际就是socket)交给Handler处理请求
                        state = handler.process(ka, SocketStatus.OPEN_READ);
                    } else {
                        state = handler.process(ka, status);
                    }
                    if (state == SocketState.CLOSED) {
                        // Close socket and pool
                        try {
                            close(ka, socket, key, SocketStatus.ERROR);
                        } catch ( Exception x ) {
                            log.error("",x);
                        }
                    }
                } else if (handshake == -1 ) {
                    close(ka, socket, key, SocketStatus.DISCONNECT);
                } else {
                    ka.getPoller().add(socket, handshake);
                }
            } catch (CancelledKeyException cx) {
                socket.getPoller().cancelledKey(key, null, false);
            } catch (OutOfMemoryError oom) {
                try {
                    oomParachuteData = null;
                    log.error("", oom);
                    if (socket != null) {
                        socket.getPoller().cancelledKey(key,SocketStatus.ERROR, false);
                    }
                    releaseCaches();
                }catch ( Throwable oomt ) {
                    try {
                        System.err.println(oomParachuteMsg);
                        oomt.printStackTrace();
                    }catch (Throwable letsHopeWeDontGetHere){
                        ExceptionUtils.handleThrowable(letsHopeWeDontGetHere);
                    }
                }
            } catch (VirtualMachineError vme) {
                ExceptionUtils.handleThrowable(vme);
            }catch ( Throwable t ) {
                log.error("",t);
                if (socket != null) {
                    socket.getPoller().cancelledKey(key,SocketStatus.ERROR,false);
                }
            } finally {
                socket = null;
                status = null;
                //return to cache
                if (running && !paused) {
                    processorCache.offer(this);
                }
            }
        }

        private void close(KeyAttachment ka, NioChannel socket, SelectionKey key,
                SocketStatus socketStatus) {
        ...
        }
    }

可以看到由 SocketProcessor 寻找合适的 Handler 处理器做最终 socket 转换处理。

可以用下面这幅图总结一下 NioEndpoint 的主要流程:

63_2.png

Acceptor 和 Poller 是线程数组,Worker 是一个线程池( Executor )

文章永久链接:https://tech.souyunku.com/?p=35891

赞(0) 打赏
版权归原创作者所有,任何形式转载请联系作者;码农code之路 博客站点 » Tomcat NIO 处理分析

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏