Tomcat源码学习--servlet请求处理

短命女 2022-05-29 07:18 236阅读 0赞

当前tomcat 8.5.20中已经不再使用bio模型，默认支持NIO实现的org.apache.coyote.http11.Http11NioProtocol，当前tomcat支持的io模型如下：

![Image 1][]

在Tomcat启动时会启用一个端口来监听请求，并且会启动多个线程来处理相关的请求操作，如下图：

![Image 1][]

相关线程介绍：

（1）http-nio-8080-Acceptor为请求接收器，其只接收请求，不会对请求做任务业务处理操作，所以默认为单个线程。

（2）http-nio-8080-ClientPoller-0和http-nio-8080-ClientPoller-1为两个是作为轮询器或者转发器使用的，简单来说就是对获取到的SocketWrapper添加到一个线程池中进行处理，这种类型的线程数与CPU的核数有关。

（3）http-nio-8080-exec-1到10是tomcat的一个线程池产生的默认的10线程，这10个线程是用来执行具体的servlet请求操作，线程的数目可以跟随请求说的变化而变化。

以上3种类型的线程有点类似Reactor模式。

接下来我们看看tomcat是如何处理一个Servlet请求的。

1、tomcat启动时会初始化一个http-nio-8080-Acceptor线程来接收请求

protected class Acceptor extends AbstractEndpoint.Acceptor {
    
            @Override
            public void run() {
    
                int errorDelay = 0;
    
                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 {
                        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();
                            if (running) {
                                // Introduce delay if necessary
                                errorDelay = handleExceptionWithDelay(errorDelay);
                                // re-throw
                                throw ioe;
                            } else {
                                break;
                            }
                        }
                        // Successful accept, reset the error delay
                        errorDelay = 0;
    
                        // Configure the socket
                        if (running && !paused) {
                            // setSocketOptions() will hand the socket off to
                            // an appropriate processor if successful
    			//对获取的请求进行处理操作
                            if (!setSocketOptions(socket)) {
                                closeSocket(socket);
                            }
                        } else {
                            closeSocket(socket);
                        }
                    } catch (Throwable t) {
                        ExceptionUtils.handleThrowable(t);
                        log.error(sm.getString("endpoint.accept.fail"), t);
                    }
                }
                state = AcceptorState.ENDED;
            }
    
    
            private void closeSocket(SocketChannel socket) {
                countDownConnection();
                try {
                    socket.socket().close();
                } catch (IOException ioe)  {
                    if (log.isDebugEnabled()) {
                        log.debug(sm.getString("endpoint.err.close"), ioe);
                    }
                }
                try {
                    socket.close();
                } catch (IOException ioe) {
                    if (log.isDebugEnabled()) {
                        log.debug(sm.getString("endpoint.err.close"), ioe);
                    }
                }
            }
        }

在上面代码中， socket = serverSock.accept();与客户端建立连接，将连接的socket交给processSocket(socket)来处理，在processSocket会将socket进行包装成NioChannel交给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.pop();
                if (channel == null) {
                    SocketBufferHandler bufhandler = new SocketBufferHandler(
                            socketProperties.getAppReadBufSize(),
                            socketProperties.getAppWriteBufSize(),
                            socketProperties.getDirectBuffer());
                    if (isSSLEnabled()) {
                        channel = new SecureNioChannel(socket, bufhandler, selectorPool, this);
                    } else {
                        channel = new NioChannel(socket, bufhandler);
                    }
                } else {
                    channel.setIOChannel(socket);
                    channel.reset();
                }
    			//交给Poller线程进行处理操作
                getPoller0().register(channel);
            } catch (Throwable t) {
                ExceptionUtils.handleThrowable(t);
                try {
                    log.error("",t);
                } catch (Throwable tt) {
                    ExceptionUtils.handleThrowable(tt);
                }
                // Tell to close the socket
                return false;
            }
            return true;
        }

2、tomcat根据处理器核数创建http-nio-8080-ClientPoller-0和http-nio-8080-ClientPoller-1线程

Poller线程会通过while(true)不断从selector.selectdKeys获取是否有连接存在，如果有连接存在则从SelectionKey中获取NioSocketWrapper并通过processKey来进行处理

@Override
            public void run() {
                // Loop until destroy() is called
                while (true) {
    
                    boolean hasEvents = false;
    
                    try {
                        if (!close) {
                            hasEvents = events();
                            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);
                            }
                            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("",x);
                        continue;
                    }
                    //either we timed out or we woke up, process events first
                    if ( keyCount == 0 ) hasEvents = (hasEvents | events());
    
                    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();
    						//获取到的请求NioSocketWrapper
                            processKey(sk, attachment);
                        }
                    }//while
    
                    //process timeouts
                    timeout(keyCount,hasEvents);
                }//while
    
                getStopLatch().countDown();
            }

在processKey中会判断一些SelectionKeyd 的状态，并将NioSocketWrapper提交给processSocket(SocketWrapperBase<S> socketWrapper,

SocketEvent event, boolean dispatch)来进行处理。

public boolean processSocket(SocketWrapperBase<S> socketWrapper,
                SocketEvent event, boolean dispatch) {
            try {
                if (socketWrapper == null) {
                    return false;
                }
                SocketProcessorBase<S> sc = processorCache.pop();
                if (sc == null) {
                    sc = createSocketProcessor(socketWrapper, event);
                } else {
                    sc.reset(socketWrapper, event);
                }
    			//获取线程池，提交请求到线程池中进行处理
                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;
        }

在processSocket函数中会获取请求处理线程，将SocketProcessor提交到线程池中进行请求处理操作。

2、tomcat创建线程池来处理所有的servlet请求，线程池中线程名称为http-nio-8080-exec-1到10，用来对具体的请求进行处理操作，NIO对应的线程操作类为SocketProcessor。

在socketProcessor线程运行时会调用doRun()方法，在doRun()方法中调用getHandler().process(socketWrapper, SocketEvent.OPEN\_READ)来处理socketWrapper请求

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) {
                            if (socket.isHandshakeComplete()) {
                               
                                handshake = 0;
                            } else if (event == SocketEvent.STOP || event == SocketEvent.DISCONNECT ||
                                    event == SocketEvent.ERROR) {
                                handshake = -1;
                            } else {
                                handshake = socket.handshake(key.isReadable(), key.isWritable());
                               
                                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;
                        // Process the request from this socket
    					//获取处理器调用process操作
                        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);
                    } 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("", t);
                    socket.getPoller().cancelledKey(key);
                } finally {
                    socketWrapper = null;
                    event = null;
                    //return to cache
                    if (running && !paused) {
                        processorCache.push(this);
                    }
                }
            }
        }

在父类AbstractProtocol的process方法中会获取处理器Processor，在处理器Processor.process(SocketWrapperBase<?> socketWrapper, SocketEvent status)中调用service方法处理socketWrapper

@Override
        public SocketState process(SocketWrapperBase<?> socketWrapper, SocketEvent status)
                throws IOException {
    
            SocketState state = SocketState.CLOSED;
            Iterator<DispatchType> dispatches = null;
            do {
                if (dispatches != null) {
                    DispatchType nextDispatch = dispatches.next();
                    state = dispatch(nextDispatch.getSocketStatus());
                } else if (status == SocketEvent.DISCONNECT) {
                    // Do nothing here, just wait for it to get recycled
                } else if (isAsync() || isUpgrade() || state == SocketState.ASYNC_END) {
                    state = dispatch(status);
                    if (state == SocketState.OPEN) {
                        // There may be pipe-lined data to read. If the data isn't
                        // processed now, execution will exit this loop and call
                        // release() which will recycle the processor (and input
                        // buffer) deleting any pipe-lined data. To avoid this,
                        // process it now.
    					//调用service方法处理SocketWrapper
                        state = service(socketWrapper);
                    }
                } else if (status == SocketEvent.OPEN_WRITE) {
                    // Extra write event likely after async, ignore
                    state = SocketState.LONG;
                } else if (status == SocketEvent.OPEN_READ){
                    state = service(socketWrapper);
                } else {
                    // Default to closing the socket if the SocketEvent passed in
                    // is not consistent with the current state of the Processor
                    state = SocketState.CLOSED;
                }
    
                if (state != SocketState.CLOSED && isAsync()) {
                    state = asyncPostProcess();
                }
    
                if (getLog().isDebugEnabled()) {
                    getLog().debug("Socket: [" + socketWrapper +
                            "], Status in: [" + status +
                            "], State out: [" + state + "]");
                }
    
                if (dispatches == null || !dispatches.hasNext()) {
                    // Only returns non-null iterator if there are
                    // dispatches to process.
                    dispatches = getIteratorAndClearDispatches();
                }
            } while (state == SocketState.ASYNC_END ||
                    dispatches != null && state != SocketState.CLOSED);
    
            return state;
        }

在http11Processor中调用service(socketWrapper)进行处理操作，调用CoyoteAdapter的service（request,response）方法进行请求操作。

@Override
        public SocketState service(SocketWrapperBase<?> socketWrapper)
            throws IOException {
            RequestInfo rp = request.getRequestProcessor();
            rp.setStage(org.apache.coyote.Constants.STAGE_PARSE);
    
            // Setting up the I/O
            setSocketWrapper(socketWrapper);
            inputBuffer.init(socketWrapper);
            outputBuffer.init(socketWrapper);
    
            // Flags
            keepAlive = true;
            openSocket = false;
            readComplete = true;
            boolean keptAlive = false;
            SendfileState sendfileState = SendfileState.DONE;
    
            while (!getErrorState().isError() && keepAlive && !isAsync() && upgradeToken == null &&
                    sendfileState == SendfileState.DONE && !endpoint.isPaused()) {
    
                // Parsing the request header
                try {
                    if (!inputBuffer.parseRequestLine(keptAlive)) {
                        if (inputBuffer.getParsingRequestLinePhase() == -1) {
                            return SocketState.UPGRADING;
                        } else if (handleIncompleteRequestLineRead()) {
                            break;
                        }
                    }
    
                    if (endpoint.isPaused()) {
                        // 503 - Service unavailable
                        response.setStatus(503);
                        setErrorState(ErrorState.CLOSE_CLEAN, null);
                    } else {
                        keptAlive = true;
                        // Set this every time in case limit has been changed via JMX
                        request.getMimeHeaders().setLimit(endpoint.getMaxHeaderCount());
                        if (!inputBuffer.parseHeaders()) {
                            // We've read part of the request, don't recycle it
                            // instead associate it with the socket
                            openSocket = true;
                            readComplete = false;
                            break;
                        }
                        if (!disableUploadTimeout) {
                            socketWrapper.setReadTimeout(connectionUploadTimeout);
                        }
                    }
                } catch (IOException e) {
                    if (log.isDebugEnabled()) {
                        log.debug(sm.getString("http11processor.header.parse"), e);
                    }
                    setErrorState(ErrorState.CLOSE_CONNECTION_NOW, e);
                    break;
                } catch (Throwable t) {
                    ExceptionUtils.handleThrowable(t);
                    UserDataHelper.Mode logMode = userDataHelper.getNextMode();
                    if (logMode != null) {
                        String message = sm.getString("http11processor.header.parse");
                        switch (logMode) {
                            case INFO_THEN_DEBUG:
                                message += sm.getString("http11processor.fallToDebug");
                                //$FALL-THROUGH$
                            case INFO:
                                log.info(message, t);
                                break;
                            case DEBUG:
                                log.debug(message, t);
                        }
                    }
                    // 400 - Bad Request
                    response.setStatus(400);
                    setErrorState(ErrorState.CLOSE_CLEAN, t);
                    getAdapter().log(request, response, 0);
                }
    
                // Has an upgrade been requested?
                Enumeration<String> connectionValues = request.getMimeHeaders().values("Connection");
                boolean foundUpgrade = false;
                while (connectionValues.hasMoreElements() && !foundUpgrade) {
                    foundUpgrade = connectionValues.nextElement().toLowerCase(
                            Locale.ENGLISH).contains("upgrade");
                }
    
                if (foundUpgrade) {
                    // Check the protocol
                    String requestedProtocol = request.getHeader("Upgrade");
    
                    UpgradeProtocol upgradeProtocol = httpUpgradeProtocols.get(requestedProtocol);
                    if (upgradeProtocol != null) {
                        if (upgradeProtocol.accept(request)) {
                            // TODO Figure out how to handle request bodies at this
                            // point.
                            response.setStatus(HttpServletResponse.SC_SWITCHING_PROTOCOLS);
                            response.setHeader("Connection", "Upgrade");
                            response.setHeader("Upgrade", requestedProtocol);
                            action(ActionCode.CLOSE,  null);
                            getAdapter().log(request, response, 0);
    
                            InternalHttpUpgradeHandler upgradeHandler =
                                    upgradeProtocol.getInternalUpgradeHandler(
                                            getAdapter(), cloneRequest(request));
                            UpgradeToken upgradeToken = new UpgradeToken(upgradeHandler, null, null);
                            action(ActionCode.UPGRADE, upgradeToken);
                            return SocketState.UPGRADING;
                        }
                    }
                }
    
                if (!getErrorState().isError()) {
                    // Setting up filters, and parse some request headers
                    rp.setStage(org.apache.coyote.Constants.STAGE_PREPARE);
                    try {
                        prepareRequest();
                    } catch (Throwable t) {
                        ExceptionUtils.handleThrowable(t);
                        if (log.isDebugEnabled()) {
                            log.debug(sm.getString("http11processor.request.prepare"), t);
                        }
                        // 500 - Internal Server Error
                        response.setStatus(500);
                        setErrorState(ErrorState.CLOSE_CLEAN, t);
                        getAdapter().log(request, response, 0);
                    }
                }
    
                if (maxKeepAliveRequests == 1) {
                    keepAlive = false;
                } else if (maxKeepAliveRequests > 0 &&
                        socketWrapper.decrementKeepAlive() <= 0) {
                    keepAlive = false;
                }
    
                // Process the request in the adapter
                if (!getErrorState().isError()) {
                    try {
                        rp.setStage(org.apache.coyote.Constants.STAGE_SERVICE);
                        getAdapter().service(request, response);
                        // Handle when the response was committed before a serious
                        // error occurred.  Throwing a ServletException should both
                        // set the status to 500 and set the errorException.
                        // If we fail here, then the response is likely already
                        // committed, so we can't try and set headers.
                        if(keepAlive && !getErrorState().isError() && !isAsync() &&
                                statusDropsConnection(response.getStatus())) {
                            setErrorState(ErrorState.CLOSE_CLEAN, null);
                        }
                    } catch (InterruptedIOException e) {
                        setErrorState(ErrorState.CLOSE_CONNECTION_NOW, e);
                    } catch (HeadersTooLargeException e) {
                        log.error(sm.getString("http11processor.request.process"), e);
                        // The response should not have been committed but check it
                        // anyway to be safe
                        if (response.isCommitted()) {
                            setErrorState(ErrorState.CLOSE_NOW, e);
                        } else {
                            response.reset();
                            response.setStatus(500);
                            setErrorState(ErrorState.CLOSE_CLEAN, e);
                            response.setHeader("Connection", "close"); // TODO: Remove
                        }
                    } catch (Throwable t) {
                        ExceptionUtils.handleThrowable(t);
                        log.error(sm.getString("http11processor.request.process"), t);
                        // 500 - Internal Server Error
                        response.setStatus(500);
                        setErrorState(ErrorState.CLOSE_CLEAN, t);
                        getAdapter().log(request, response, 0);
                    }
                }
    
                // Finish the handling of the request
                rp.setStage(org.apache.coyote.Constants.STAGE_ENDINPUT);
                if (!isAsync()) {
                    // If this is an async request then the request ends when it has
                    // been completed. The AsyncContext is responsible for calling
                    // endRequest() in that case.
                    endRequest();
                }
                rp.setStage(org.apache.coyote.Constants.STAGE_ENDOUTPUT);
    
                // If there was an error, make sure the request is counted as
                // and error, and update the statistics counter
                if (getErrorState().isError()) {
                    response.setStatus(500);
                }
    
                if (!isAsync() || getErrorState().isError()) {
                    request.updateCounters();
                    if (getErrorState().isIoAllowed()) {
                        inputBuffer.nextRequest();
                        outputBuffer.nextRequest();
                    }
                }
    
                if (!disableUploadTimeout) {
                    int soTimeout = endpoint.getConnectionTimeout();
                    if(soTimeout > 0) {
                        socketWrapper.setReadTimeout(soTimeout);
                    } else {
                        socketWrapper.setReadTimeout(0);
                    }
                }
    
                rp.setStage(org.apache.coyote.Constants.STAGE_KEEPALIVE);
    
                sendfileState = processSendfile(socketWrapper);
            }
    
            rp.setStage(org.apache.coyote.Constants.STAGE_ENDED);
    
            if (getErrorState().isError() || endpoint.isPaused()) {
                return SocketState.CLOSED;
            } else if (isAsync()) {
                return SocketState.LONG;
            } else if (isUpgrade()) {
                return SocketState.UPGRADING;
            } else {
                if (sendfileState == SendfileState.PENDING) {
                    return SocketState.SENDFILE;
                } else {
                    if (openSocket) {
                        if (readComplete) {
                            return SocketState.OPEN;
                        } else {
                            return SocketState.LONG;
                        }
                    } else {
                        return SocketState.CLOSED;
                    }
                }
            }
        }

在CoyoteAdapter中调用service方法，在方法中调用 connector.getService().getContainer().getPipeline().getFirst().invoke(request, response);查找最终要执行的servlet进行操作。

[Image 1]: