前言

前面已經(jīng)初步分析請求流程，下面我們繼續(xù)。

Poller流程處理

從上一篇直到Acceptor接受到請求并注冊到Poller中的events緩存棧中，下面來想起看一下Poller的處理流程。

        public void run() {
            // Loop until destroy() is called
            while (true) {

                boolean hasEvents = false;

                try {
                    if (!close) {
                        //1. 更新PollerEvent隊(duì)列，主要執(zhí)行PollerEvent的run方法來更新selector感興趣的事件
                        hasEvents = events();
                        //2. 將wakeupCounter設(shè)置為-1，如果oldvalue>0，做非阻塞select;否則做超時(shí)的阻塞select。其中wakeupCounter在#addEvent()時(shí)會(huì)加1
                        if (wakeupCounter.getAndSet(-1) > 0) {
                            // If we are here, means we have other stuff to do
                            // Do a non blocking select
                            keyCount = selector.selectNow();
                        } else {
                            keyCount = selector.select(selectorTimeout);
                        }
                        //3. wakeupCounter設(shè)置為0
                        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 (Throwable x) {
                    ExceptionUtils.handleThrowable(x);
                    log.error(sm.getString("endpoint.nio.selectorLoopError"), x);
                    continue;
                }
                //4. keyCount為0，說明沒有事件到來，再執(zhí)行一次#events()
                if (keyCount == 0) hasEvents = (hasEvents | events());
                //5. 遍歷SelectionKey(事件已到來，進(jìn)行后續(xù)處理)，進(jìn)行讀寫處理
                Iterator<SelectionKey> iterator =
                        keyCount > 0 ? selector.selectedKeys().iterator() : null;
                // Walk through the collection of ready keys and dispatch
                // any active event.
                while (iterator != null && iterator.hasNext()) {
                    SelectionKey sk = iterator.next();
                    NioSocketWrapper attachment = (NioSocketWrapper) sk.attachment();
                    // Attachment may be null if another thread has called
                    // cancelledKey()
                    if (attachment == null) {
                        iterator.remove();
                    } else {
                        iterator.remove();
                        //6. 提交到線程池中進(jìn)行處理處理
                        processKey(sk, attachment);
                    }
                }

                // Process timeouts
                timeout(keyCount, hasEvents);
            }

            getStopLatch().countDown();
        }
       public boolean events() {
            boolean result = false;

            PollerEvent pe = null;
            for (int i = 0, size = events.size(); i < size && (pe = events.poll()) != null; i++) {
                result = true;
                try {
                    pe.run();
                    pe.reset();
                    if (running && !paused) {
                        eventCache.push(pe);
                    }
                } catch (Throwable x) {
                    log.error(sm.getString("endpoint.nio.pollerEventError"), x);
                }
            }

            return result;
        }

主要流程如下：

• 調(diào)用#event()方法，更新selector感興趣的事件；
• 執(zhí)行#selector.select()方法，檢測是否有事件到來；
• 將到來的事件提交至線程池進(jìn)行下一步處理。

        protected void processKey(SelectionKey sk, NioSocketWrapper socketWrapper) {
            try {
                if (close) {
                    cancelledKey(sk);
                } else if (sk.isValid() && socketWrapper != null) {
                    if (sk.isReadable() || sk.isWritable()) {
                        if (socketWrapper.getSendfileData() != null) {
                            processSendfile(sk, socketWrapper, false);
                        } else {
                            //1. 在通道上注銷對已經(jīng)發(fā)生事件的關(guān)注
                            unreg(sk, socketWrapper, sk.readyOps());
                            boolean closeSocket = false;
                            // Read goes before write
                            if (sk.isReadable()) {
                                //2. 進(jìn)行異步IO的處理或者交給SocketProcessor處理讀操作
                                if (socketWrapper.readOperation != null) {
                                    getExecutor().execute(socketWrapper.readOperation);
                                } else if (!processSocket(socketWrapper, SocketEvent.OPEN_READ, true)) {
                                    closeSocket = true;
                                }
                            }
                            if (!closeSocket && sk.isWritable()) {
                                //3. 進(jìn)行異步IO的處理或者交給SocketProcessor處理寫操作
                                if (socketWrapper.writeOperation != null) {
                                    getExecutor().execute(socketWrapper.writeOperation);
                                } else if (!processSocket(socketWrapper, SocketEvent.OPEN_WRITE, true)) {
                                    closeSocket = true;
                                }
                            }
                            if (closeSocket) {
                                cancelledKey(sk);
                            }
                        }
                    }
                } else {
                    // Invalid key
                    cancelledKey(sk);
                }
            } catch (CancelledKeyException ckx) {
                cancelledKey(sk);
            } catch (Throwable t) {
                ExceptionUtils.handleThrowable(t);
                log.error(sm.getString("endpoint.nio.keyProcessingError"), t);
            }
        }
        //防止了通道對同一個(gè)事件不斷select的問題
        protected void unreg(SelectionKey sk, NioSocketWrapper socketWrapper, int readyOps) {
            // This is a must, so that we don't have multiple threads messing with the socket
            reg(sk, socketWrapper, sk.interestOps() & (~readyOps));
        }
        
        protected void reg(SelectionKey sk, NioSocketWrapper socketWrapper, int intops) {
            sk.interestOps(intops);
            socketWrapper.interestOps(intops);
        }

這里主要注銷了對已經(jīng)發(fā)生事件的關(guān)注，然后將具體的處理邏輯交給SocketProcessor來處理，后面會(huì)介紹。

工作線程流程處理

從上面了解到，最后會(huì)調(diào)用#Poller.processSocket()方法，將處理邏輯交給SocketProcessor類，我們來看下：

public boolean processSocket(SocketWrapperBase<S> socketWrapper,
            SocketEvent event, boolean dispatch) {
        try {
            if (socketWrapper == null) {
                return false;
            }
            //1. 從緩存棧中獲取SocketProcessor，無則創(chuàng)建否則重置SocketProcessor對象
            SocketProcessorBase<S> sc = processorCache.pop();
            if (sc == null) {
                sc = createSocketProcessor(socketWrapper, event);
            } else {
                sc.reset(socketWrapper, event);
            }
            //2. 獲取線程池，如果配置了線程池則將SocketProcessor提交到線程池中執(zhí)行，否則直接執(zhí)行SocketProcessor的run方法。
            Executor executor = getExecutor();
            if (dispatch && executor != null) {
                executor.execute(sc);
            } else {
                sc.run();
            }
        } catch (RejectedExecutionException ree) {
            getLog().warn(sm.getString("endpoint.executor.fail", socketWrapper) , ree);
            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
            getLog().error(sm.getString("endpoint.process.fail"), t);
            return false;
        }
        return true;
    }

如果配置了線程池，則提交SocketProcessor線程至線程池中執(zhí)行，否則直接執(zhí)行#SocketProcessor.run()。下面來看下SocketProcessor類：

public abstract class SocketProcessorBase<S> implements Runnable {

    protected SocketWrapperBase<S> socketWrapper;
    //socket事件狀態(tài)
    protected SocketEvent event;

    public SocketProcessorBase(SocketWrapperBase<S> socketWrapper, SocketEvent event) {
        reset(socketWrapper, event);
    }


    public void reset(SocketWrapperBase<S> socketWrapper, SocketEvent event) {
        Objects.requireNonNull(event);
        this.socketWrapper = socketWrapper;
        this.event = event;
    }


    @Override
    public final void run() {
        synchronized (socketWrapper) {
            // It is possible that processing may be triggered for read and
            // write at the same time. The sync above makes sure that processing
            // does not occur in parallel. The test below ensures that if the
            // first event to be processed results in the socket being closed,
            // the subsequent events are not processed.
            if (socketWrapper.isClosed()) {
                return;
            }
            doRun();
        }
    }
    //子類實(shí)現(xiàn)
    protected abstract void doRun();
}
protected class SocketProcessor extends SocketProcessorBase<NioChannel> {

        public SocketProcessor(SocketWrapperBase<NioChannel> socketWrapper, SocketEvent event) {
            super(socketWrapper, event);
        }

        @Override
        protected void doRun() {
            NioChannel socket = socketWrapper.getSocket();
            SelectionKey key = socket.getIOChannel().keyFor(socket.getPoller().getSelector());

            try {
                int handshake = -1;

                try {
                    if (key != null) {
                        //NioChannel默認(rèn)返回true，SecureNioChannel這里才需要處理
                        if (socket.isHandshakeComplete()) {
                            // No TLS handshaking required. Let the handler
                            // process this socket / event combination.
                            handshake = 0;
                        } else if (event == SocketEvent.STOP || event == SocketEvent.DISCONNECT ||
                                event == SocketEvent.ERROR) {
                            // Unable to complete the TLS handshake. Treat it as
                            // if the handshake failed.
                            handshake = -1;
                        } 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.
                            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;
                    //最關(guān)鍵的代碼，交給handler處理socket
                    if (event == null) {
                        state = getHandler().process(socketWrapper, SocketEvent.OPEN_READ);
                    } else {
                        state = getHandler().process(socketWrapper, event);
                    }
                    if (state == SocketState.CLOSED) {
                        close(socket, key);
                    }
                } else if (handshake == -1) {
                    close(socket, key);
                //如果handshake返回的是SelectionKey.OP_READ，注冊讀事件到Poller；如果返回的是SelectionKey.OP_WRITE，注冊寫事件到Poller，進(jìn)行后續(xù)處理
                } else if (handshake == SelectionKey.OP_READ) {
                    socketWrapper.registerReadInterest();
                } else if (handshake == SelectionKey.OP_WRITE) {
                    socketWrapper.registerWriteInterest();
                }
            } catch (CancelledKeyException cx) {
                socket.getPoller().cancelledKey(key);
            } catch (VirtualMachineError vme) {
                ExceptionUtils.handleThrowable(vme);
            } catch (Throwable t) {
                log.error(sm.getString("endpoint.processing.fail"), t);
                socket.getPoller().cancelledKey(key);
            } finally {
                socketWrapper = null;
                event = null;
                //return to cache
                if (running && !paused) {
                    //處理結(jié)束后，將SocketProcessor重新放入緩存棧中
                    processorCache.push(this);
                }
            }
        }
    }

這里SocketEvent有OPEN_READ、OPEN_WRITE、STOP、TIMEOUT、DISCONNECT、ERROR六中狀態(tài)。SocketProcessor對象主要將socket交給Handler來處理請求。

總結(jié)

到這里NIO的整個(gè)處理流程就大致清楚了，整體流程如下：

NIO整體處理流程

下面深入Handler來看一下Socket請求是如何轉(zhuǎn)換為Request對象，以及如何調(diào)用Servlet中的方法。