Wormhole
码农翻身

Tomcat源码篇之HTTP请求处理流程

- by MRyan, 2023-01-15



本系列针对于 Tomcat 版本为 8.5X

文章已收录至精进Tomcat系列 系列其它文章 https://www.wormholestack.com/tag/Tomcat/

源码阅读环境：https://gitee.com/M-Analysis/source_tomcat8 已填充关键注释


当请求从客户端发起之后，Tomcat 的处理流程是怎么样的呢？

1. Tomcat 线程模型结构

不知道你有没有注意到，上文《Tomcat源码篇之启动流程》中，Tomcat 成功启动了之后，会开启 6 个线程核心线程。

uTools_1673270800355

其中 1 个用户线程，5 个守护线程。

守护线程

所谓守护线程，是指在程序运行的时候在后台提供一种通用服务的线程，比如垃圾回收线程。这种线程并不属于程序中不可或缺的部分，用户线程和守护线程两者几乎没有区别，唯一的不同之处就在于虚拟机的离开：如果用户线程已经全部退出运行了，只剩下守护线程存在了，虚拟机也就退出了。因为没有了被守护者，守护线程也就没有工作可做了，也就没有继续运行程序的必要了。反过来说，只要任何非守护线程还在运行，程序就不会终止。

将线程转换为守护线程可以通过调用 Thread 对象的 setDaemon(true) 方法来实现。

在《Tomcat源码篇之启动流程》中我们简要介绍了以上线程的创建过程

Engine 启动时会开启 ContainerBackgroundProcessor 守护线程作用于调用子容器的backgroundProcess 方法。

Connector 启动时会开启 AsyncTimeout 守护线程作用于检测超时的请求，并将该请求再转发到工作线程池处理。

Endpoint 启动时会创建 Work-exec 工作者线程池， 默认 10 个等待处理，作用于真正的任务处理工作。

同时也会创建 Poller 守护线程，用于对接受者线程生产的消息（或事件）进行处理。

还会创建 Acceptor 守护线程，端口 8080，用于监听套接字，将已连接套接字转给Poller线程。

Connector 组件是 Tomcat 最核心的两个组件之一，主要的职责就是负责接收客户端连接和客户端请求的处理加工，上述核心线程大部分都作用于 Connector 组件上。

在《Tomcat架构的秘密》中我们简单对 Connector 进行了架构解析，了解了 Connector 组件中的几个部件，如下图所示：

image-20230109220401350


下面我们首先通过 Connector 源码的详细分析，来真正的了解上述核心线程的用途。

1.1 Acceptor线程

Acceptor 作为监听接受请求的线程，也是 Tomcat 处理一次请求的入口点。

// 已删去非关键代码
public class Acceptor<U> implements Runnable {

    private static final Log log = LogFactory.getLog(Acceptor.class);
    private static final StringManager sm = StringManager.getManager(Acceptor.class);

    private static final int INITIAL_ERROR_DELAY = 50;
    private static final int MAX_ERROR_DELAY = 1600;

    private final AbstractEndpoint<?， U> endpoint;

    public Acceptor(AbstractEndpoint<?， U> endpoint) {
        this.endpoint = endpoint;
    }


    /**
     * 接受端口来的数据
     */
    @SuppressWarnings("deprecation")
    @Override
    public void run() {

        int errorDelay = 0;
        long pauseStart = 0;

        try {
            // Loop until we receive a shutdown command
            while (!stopCalled) {

                 // 运行过程中，如果`Endpoint`暂停了，则`Acceptor`进行自旋`
                while (endpoint.isPaused() && !stopCalled) {
                    if (state != AcceptorState.PAUSED) {
                        pauseStart = System.nanoTime();
                        // Entered pause state
                        state = AcceptorState.PAUSED;
                    }
                    if ((System.nanoTime() - pauseStart) > 1_000_000) {
                        // Paused for more than 1ms
                        try {
                            if ((System.nanoTime() - pauseStart) > 10_000_000) {
                                Thread.sleep(10);
                            } else {
                                Thread.sleep(1);
                            }
                        } catch (InterruptedException e) {
                            // Ignore
                        }
                    }
                }

                // 如果`Endpoint`终止运行了，则`Acceptor`也会终止
                if (stopCalled) {
                    break;
                }
                state = AcceptorState.RUNNING;

                try {
                    //if we have reached max connections， wait
                    // 如果请求达到了最大连接数，则wait直到连接数降下来
                    endpoint.countUpOrAwaitConnection();

                    // Endpoint might have been paused while waiting for latch
                    // If that is the case， don't accept new connections
                    if (endpoint.isPaused()) {
                        continue;
                    }

                    U socket = null;
                    try {
                        // Accept the next incoming connection from the server
                        // socket
                        // 接受下一次连接的socket 接受8080端口数据
                        socket = endpoint.serverSocketAccept();
                    } catch (Exception ioe) {
                        // We didn't get a socket
                        endpoint.countDownConnection();
                        if (endpoint.isRunning()) {
                            // 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 (!stopCalled && !endpoint.isPaused()) {
 
                        // 会将socket以事件的方式传递给poller队列
                        if (!endpoint.setSocketOptions(socket)) {
                            // 设置socket失败 关闭socket
                            endpoint.closeSocket(socket);
                        }
                    } else {
                        endpoint.destroySocket(socket);
                    }
                } catch (Throwable t) {
                    ExceptionUtils.handleThrowable(t);
                    String msg = sm.getString("endpoint.accept.fail");
                    // APR specific.
                    // Could push this down but not sure it is worth the trouble.
                    if (t instanceof org.apache.tomcat.jni.Error) {
                        org.apache.tomcat.jni.Error e = (org.apache.tomcat.jni.Error) t;
                        if (e.getError() == 233) {
                            // Not an error on HP-UX so log as a warning
                            // so it can be filtered out on that platform
                            // See bug 50273
                            log.warn(msg， t);
                        } else {
                            log.error(msg， t);
                        }
                    } else {
                        log.error(msg， t);
                    }
                }
            }
        } finally {
            stopLatch.countDown();
        }
        state = AcceptorState.ENDED;
    }


}

当 Acceptor 监听到请求时，会触发以下代码

其关键代码

 endpoint.countUpOrAwaitConnection();

这一行代码的作用是检查当前最大连接数，若未达到 maxConnections 则 +1，如果请求达到了最大连接数，则等待直到连接数降下来。

socket = endpoint.serverSocketAccept();

这一行中的 serverSocket 正是 NioEndpoint 的 bind 函数中打开的 ServerSocketChannel，用于接受下一次连接的 socket 接受 8080 端口数据。

endpoint.setSocketOptions(socket)

进入 Endpoint 的 setSocketOptions 方法

@Override
    protected boolean setSocketOptions(SocketChannel socket) {
        NioSocketWrapper socketWrapper = null;
        try {
            // Allocate channel and wrapper
            NioChannel channel = null;
            if (nioChannels != null) {
                channel = nioChannels.pop();
            }
            if (channel == null) {
                SocketBufferHandler bufhandler = new SocketBufferHandler(
                    socketProperties.getAppReadBufSize()，
                    socketProperties.getAppWriteBufSize()，
                    socketProperties.getDirectBuffer());
                if (isSSLEnabled()) {
                    channel = new SecureNioChannel(bufhandler， this);
                } else {
                    channel = new NioChannel(bufhandler);
                }
            }
            NioSocketWrapper newWrapper = new NioSocketWrapper(channel， this);
            channel.reset(socket， newWrapper);
            // 缓存
            connections.put(socket， newWrapper);
            socketWrapper = newWrapper;

            // Set socket properties
            // Disable blocking， polling will be used
            socket.configureBlocking(false);
            socketProperties.setProperties(socket.socket());

            socketWrapper.setReadTimeout(getConnectionTimeout());
            socketWrapper.setWriteTimeout(getConnectionTimeout());
            socketWrapper.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
            // 将Socket注册到poller队列
            poller.register(socketWrapper);
            return true;
        } catch (Throwable t) {
            ExceptionUtils.handleThrowable(t);
            try {
                log.error(sm.getString("endpoint.socketOptionsError")， t);
            } catch (Throwable tt) {
                ExceptionUtils.handleThrowable(tt);
            }
            if (socketWrapper == null) {
                destroySocket(socket);
            }
        }
        // Tell to close the socket if needed
        return false;
    }

可以看到 poller.register(socketWrapper); 将 Socket 包装成了 socketWrapper ，并注册到了 poller 队列中。

注册的方式如下：


public void register(final NioSocketWrapper socketWrapper) {  
            socketWrapper.interestOps(SelectionKey.OP_READ);//this is what OP_REGISTER turns into.
            PollerEvent pollerEvent = createPollerEvent(socketWrapper， OP_REGISTER);
                      // 向events对了中间添加事件
            addEvent(pollerEvent);
        }
    // poller事件队列
private final SynchronizedQueue<PollerEvent> events = new SynchronizedQueue<>();   

private void addEvent(PollerEvent event) {
            events.offer(event);
            if (wakeupCounter.incrementAndGet() == 0) {
                selector.wakeup();
            }
        }

将 Socket 的包装类 socketWrapper 再次封装成 poller 事件，向 events 事件队列中添加了该事件（此时并未读取数据）。

总结一下其实 Acceptor 线程主要用于监听套接字，将已连接套接字转给 Poller 线程。

1.2 Poller线程

Poller 线程在 NioEndpoint 中被定义

  public class Poller implements Runnable {
     // 省略部分代码

       private Selector selector;
  
        // poller事件队列
        private final SynchronizedQueue<PollerEvent> events =
            new SynchronizedQueue<>();

        public Poller() throws IOException {
            // 为每个Poller打开了一个新的Selector
            this.selector = Selector.open();
        }      
  }

Poller 线程构造函数为每个 Poller 打开了一个新的 Selector

public class Poller implements Runnable {
/**
         * The background thread that adds sockets to the Poller， checks the
         * poller for triggered events and hands the associated socket off to an
         * appropriate processor as events occur.
         * <p>
         * Poller线程执行方法 事件处理
         */
        @Override
        public void run() {
            // Loop until destroy() is called
            // 循环
            while (true) {

                boolean hasEvents = false;

                try {
                    if (!close) {
                        hasEvents = events();
                       // 将通道注册到Poller的Selector上
                        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;
                    }
                    // Either we timed out or we woke up， process events first
                    if (keyCount == 0) {
                        hasEvents = (hasEvents | events());
                    }
                } catch (Throwable x) {
                    ExceptionUtils.handleThrowable(x);
                    log.error(sm.getString("endpoint.nio.selectorLoopError")， x);
                    continue;
                }

                // 获取当前选择器中所有注册的“选择键(已就绪的监听事件)”
                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();
                    iterator.remove();
                    NioSocketWrapper socketWrapper = (NioSocketWrapper) sk.attachment();
                    // Attachment may be null if another thread has called
                    // cancelledKey()
                    if (socketWrapper != null) {
                        // 真正处理事件的地方
                        processKey(sk， socketWrapper);
                    }
                }

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

            getStopLatch().countDown();
        }
}

上文我们分析了 Poller#register() 方法。Poller 维持了一个 events同步队列，所以 Acceptor 接收到 Poller事件（Socket 监听事件）会放在这个队列里面，通过 events() 方法获取队列中的 Poller事件（Socket 监听事件）

hasEvents = events();
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;
                NioSocketWrapper socketWrapper = pe.getSocketWrapper();
                SocketChannel sc = socketWrapper.getSocket().getIOChannel();
                int interestOps = pe.getInterestOps();
                if (sc == null) {
                    log.warn(sm.getString("endpoint.nio.nullSocketChannel"));
                    socketWrapper.close();
                } else if (interestOps == OP_REGISTER) {
                    try {
                        sc.register(getSelector()， SelectionKey.OP_READ， socketWrapper);
                    } catch (Exception x) {
                        log.error(sm.getString("endpoint.nio.registerFail")， x);
                    }
                } else {
                    final SelectionKey key = sc.keyFor(getSelector());
                    if (key == null) {
                         socketWrapper.close();
                    } else {
                        final NioSocketWrapper attachment = (NioSocketWrapper) key.attachment();
                        if (attachment != null) {
                            try {
                                int ops = key.interestOps() | interestOps;
                                attachment.interestOps(ops);
                                key.interestOps(ops);
                            } catch (CancelledKeyException ckx) {
                                cancelledKey(key， socketWrapper);
                            }
                        } else {
                            cancelledKey(key， socketWrapper);
                        }
                    }
                }
                if (running && eventCache != null) {
                    pe.reset();
                    eventCache.push(pe);
                }
            }

            return result;
        }

线程获取到了 event 队列的就绪事件，来继续看下 Poller 线程处理

 // 获取当前选择器中所有注册的(已就绪的监听事件)
                Iterator<SelectionKey> iterator =
                    keyCount > 0 ? selector.selectedKeys().iterator() : null;
                  while (iterator != null && iterator.hasNext()) {
                    SelectionKey sk = iterator.next();
                    iterator.remove();
                    NioSocketWrapper socketWrapper = (NioSocketWrapper) sk.attachment();
                    if (socketWrapper != null) {
                        // 真正处理事件的地方
                        processKey(sk， socketWrapper);
                    }
                }

获取已连接套接字通道注册到 Poller 线程的 Selector 上的 SelectionKey，进行事件处理


接着分析下 processKey()，该方法会根据 key 的类型，来分别处理读操作和写操作

  protected void processKey(SelectionKey sk， NioSocketWrapper socketWrapper) {
            try {
                if (close) {
                    cancelledKey(sk， socketWrapper);
                } else if (sk.isValid()) {
                    if (sk.isReadable() || sk.isWritable()) {
                        if (socketWrapper.getSendfileData() != null) {
                            processSendfile(sk， socketWrapper， false);
                        } else {
                            unreg(sk， socketWrapper， sk.readyOps());
                            boolean closeSocket = false;
                            // Read goes before write
                            // 处理读事件，生成Request对象
                            if (sk.isReadable()) {
                                if (socketWrapper.readOperation != null) {
                                    if (!socketWrapper.readOperation.process()) {
                                        closeSocket = true;
                                    }
                                } else if (socketWrapper.readBlocking) {
                                    synchronized (socketWrapper.readLock) {
                                        socketWrapper.readBlocking = false;
                                        socketWrapper.readLock.notify();
                                    }
                                } else if (!processSocket(socketWrapper， SocketEvent.OPEN_READ， true)) {
                                    closeSocket = true;
                                }
                            }
                            // 处理写事件，将生成的Response对象通过socket写回客户端
                            if (!closeSocket && sk.isWritable()) {
                                if (socketWrapper.writeOperation != null) {
                                    if (!socketWrapper.writeOperation.process()) {
                                        closeSocket = true;
                                    }
                                } else if (socketWrapper.writeBlocking) {
                                    synchronized (socketWrapper.writeLock) {
                                        socketWrapper.writeBlocking = false;
                                        socketWrapper.writeLock.notify();
                                    }
                                } else if (!processSocket(socketWrapper， SocketEvent.OPEN_WRITE， true)) {
                                    closeSocket = true;
                                }
                            }
                            if (closeSocket) {
                                cancelledKey(sk， socketWrapper);
                            }
                        }
                    }
                } else {
                    // Invalid key
                    cancelledKey(sk， socketWrapper);
                }
            } catch (CancelledKeyException ckx) {
                cancelledKey(sk， socketWrapper);
            } catch (Throwable t) {
                ExceptionUtils.handleThrowable(t);
                log.error(sm.getString("endpoint.nio.keyProcessingError")， t);
            }
        }

来看下 processSocket() 方法

public boolean processSocket(SocketWrapperBase<S> socketWrapper， SocketEvent event， boolean dispatch) {
        try {
            if (socketWrapper == null) {
                return false;
            }
            SocketProcessorBase<S> sc = null;
            // 为提升性能，针对每个有效的链接都会缓存其Processor对象，不仅如此，当前链接关闭时，其Processor对象还会被释放到一个回收队列（升级协议不会回收）这样后续链接可以重置冰重复利用，以减少对象构造
            // 缓存中有则从缓存中获取一个Processor
            if (processorCache != null) {
                // 从processorCache里面拿一个Processor来处理socket，Processor的实现为SocketProcessor
                sc = processorCache.pop();
            }
            // 缓存中没有则根据协商协议创建一个SocketProcessor
            if (sc == null) {
                sc = createSocketProcessor(socketWrapper， event);
            } else {
                sc.reset(socketWrapper， event);
            }
            //  将Processor放到Work工作线程池中执行
            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);
             getLog().error(sm.getString("endpoint.process.fail")， t);
            return false;
        }
        return true;
    }

实际上它只做了两件事

  1. 从 processorCache 里面拿一个 Processor 来处理socket，缓存中没有则根据协商协议创建爱一个Processor，Processor 的实现为 SocketProcessor
  2. 将 Processor 放到 Work 工作线程池中执行

dispatch 参数表示是否要在工作线程中处理，上文 processKey 各处传递的参数默认都是true。

  • dispatch 为 true 且工作线程池存在时会执行 executor.execute(sc)，之后是由工作线程池处理已连接套接字；
  • 否则继续由 Poller 线程自己处理已连接套接字。

AbstractEndPoint 的 createSocketProcessor 是抽象方法，NioEndPoint 实现了它：

  @Override
    protected SocketProcessorBase<NioChannel> createSocketProcessor(
        SocketWrapperBase<NioChannel> socketWrapper， SocketEvent event) {
        return new SocketProcessor(socketWrapper， event);
    }

SocketProcessor 继承了 SocketProcessorBase，而 SocketProcessorBase 实现了 Runnable。


public abstract class SocketProcessorBase<S> implements Runnable {

    protected SocketWrapperBase<S> socketWrapper;
    protected SocketEvent event;

    @Override
    public final void run() {
        synchronized (socketWrapper) {
             if (socketWrapper.isClosed()) {
                return;
            }
            doRun();
        }
    }


    protected abstract void doRun();
}

SocketProcessorBase 线程执行中会调用 抽象方法 doRun，由子类实现，也就是 SocketProcessor 的 doRun 方法，也就是说 Work 工作线程最终会执行 SocketProcessor 的 doRun方法。

   protected class SocketProcessor extends SocketProcessorBase<NioChannel> {

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

        @Override
        protected void doRun() {
          
            Poller poller = NioEndpoint.this.poller;
            if (poller == null) {
                socketWrapper.close();
                return;
            }

            try {
                // https 握手环节
                int handshake = -1;
                try {
                    if (socketWrapper.getSocket().isHandshakeComplete()) {
                        handshake = 0;
                    } else if (event == SocketEvent.STOP || event == SocketEvent.DISCONNECT ||
                        event == SocketEvent.ERROR) {    
                        handshake = -1;
                    } else {
                        handshake = socketWrapper.getSocket().handshake(event == SocketEvent.OPEN_READ， event == SocketEvent.OPEN_WRITE);
                        event = SocketEvent.OPEN_READ;
                    }
                } catch (IOException x) {
                    handshake = -1;
                    if (log.isDebugEnabled()) {
                        log.debug(sm.getString("endpoint.err.handshake")， x);
                    }
                } catch (CancelledKeyException ckx) {
                    handshake = -1;
                }
                if (handshake == 0) {
                    SocketState state = SocketState.OPEN;        
                    // 将处理逻辑交给`Handler`处理，当event为null时，则表明是一个`OPEN_READ`事件
                    if (event == null) {
                        state = getHandler().process(socketWrapper， SocketEvent.OPEN_READ);
                    } else {
                        state = getHandler().process(socketWrapper， event);
                    }
                    if (state == SocketState.CLOSED) {
                        poller.cancelledKey(getSelectionKey()， socketWrapper);
                    }
                } else if (handshake == -1) {
                    getHandler().process(socketWrapper， SocketEvent.CONNECT_FAIL);
                    poller.cancelledKey(getSelectionKey()， socketWrapper);
                } else if (handshake == SelectionKey.OP_READ) {
                    socketWrapper.registerReadInterest();
                } else if (handshake == SelectionKey.OP_WRITE) {
                    socketWrapper.registerWriteInterest();
                }
            } catch (CancelledKeyException cx) {
                poller.cancelledKey(getSelectionKey()， socketWrapper);
            } catch (VirtualMachineError vme) {
                ExceptionUtils.handleThrowable(vme);
            } catch (Throwable t) {
                log.error(sm.getString("endpoint.processing.fail")， t);
                poller.cancelledKey(getSelectionKey()， socketWrapper);
            } finally {
                socketWrapper = null;
                event = null;
                //return to cache
                if (running && processorCache != null) {
                    processorCache.push(this);
                }
            }
        }
    }

Handler 的关键方法是 process()

state = getHandler().process(socketWrapper， event);

主要是调用 Processor.process()方法。

 protected static class ConnectionHandler<S> implements AbstractEndpoint.Handler<S> {

        @Override
        public SocketState process(SocketWrapperBase<S> wrapper, SocketEvent status) {
          // 省略部分代码
          
             // 拿到Socket nio包装wrapper 监听8080端口的socket
            S socket = wrapper.getSocket();

            Processor processor = (Processor) wrapper.takeCurrentProcessor();

            if (processor == null) {
                // 当前协议创建一个Processor
                processor = getProtocol().createProcessor();
                register(processor);
            }

            SocketState state = SocketState.CLOSED
            // 关键代码 处理器处理socket
            state = processor.process(wrapper, status);

            return state;
        }
 }

public abstract class AbstractHttp11Protocol<S> extends AbstractProtocol<S> {
    @Override
    protected Processor createProcessor() {
        Http11Processor processor = new Http11Processor(this, getEndpoint());
        processor.setAdapter(getAdapter());
        // 默认的 KeepAlive 情况下, 每个 Socket 处理的最多的 请求次数
        processor.setMaxKeepAliveRequests(getMaxKeepAliveRequests());
        // http 当遇到文件上传时的 默认超时时间 (300 * 1000)
        processor.setConnectionUploadTimeout(getConnectionUploadTimeout());
        processor.setDisableUploadTimeout(getDisableUploadTimeout());
        processor.setRestrictedUserAgents(getRestrictedUserAgents());
        // 最大的 Post 处理尺寸的大小 4 * 1000
        processor.setMaxSavePostSize(getMaxSavePostSize());
        return processor;
    }
}

代码比较长，我们直接看关键代码

  state = processor.process(wrapper, status);
public abstract class AbstractProcessorLight implements Processor {

    private Set<DispatchType> dispatches = new CopyOnWriteArraySet<>();


    @Override
    public SocketState process(SocketWrapperBase<?> socketWrapper, SocketEvent status)
        throws IOException {

        // 处理中 设置关闭socket状态
        SocketState state = SocketState.CLOSED;
        Iterator<DispatchType> dispatches = null;
        do {
            if (dispatches != null) {
                DispatchType nextDispatch = dispatches.next();
                if (getLog().isDebugEnabled()) {
                    getLog().debug("Processing dispatch type: [" + nextDispatch + "]");
                }
                state = dispatch(nextDispatch.getSocketStatus());
                if (!dispatches.hasNext()) {
                    state = checkForPipelinedData(state, socketWrapper);
                }
            } 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);
                state = checkForPipelinedData(state, socketWrapper);
            } else if (status == SocketEvent.OPEN_WRITE) {
                // Extra write event likely after async, ignore
                state = SocketState.LONG;
            } else if (status == SocketEvent.OPEN_READ) {
                // 调用`service()`方法
                state = service(socketWrapper);
            } else if (status == SocketEvent.CONNECT_FAIL) {
                logAccess(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;
            }
            
        } while (state == SocketState.ASYNC_END ||
            dispatches != null && state != SocketState.CLOSED);

        return state;
    }

对于读操作，会调用service()方法

 public class Http11Processor extends AbstractProcessor {
  
        @Override
    public SocketState service(SocketWrapperBase<?> socketWrapper)
        throws IOException {
       // 省略部分代码，感兴趣可自行翻看源码
      
       // 关键代码，CoyoteAdapter请求映射
        getAdapter().service(request, response);
    }
 }

上述代码实际只完成了 2 件事

  1. 读取消息报文，解析请求行、请求体、请求头，封装成 Request 对象
  2. 生成 Response 对象
  3. 调用 Adapter.service() 方法，将生成的 Request 和 Response 对象传进去，由连接器分发请求各组件执行处理流程

1.3 小结

回顾下我们刚刚分析的 Tomcat 接受请求的流程

image-20230115001028034


分析到目前为止 Tomcat 线程模型结构应该是非常清晰了，简单总结下：

Acceptor守护线程，Poller守护线程，Worker批量工作线程

  1. Tomcat启动结束后,Accepor线程 一直接受 8080 的请求，会尝试 endpoint.setSocketOptions(socket)，socket 封装为 socketWrapper,创建一个 PollerEvent 然后把请求注册到 poller 事件队列 SynchronizedQueue 上。
  2. Poller线程 一直判断是否有事件 events.poll()，事件就拿到读取 socket 的内容并处理 processSocket (socketWrapper)，processSocket 的时候 poller 会进行select()，接收到 key 后提交给 worker 进行processKey，也就是把线程形成一个 processor 进行后提交给 Worker 线程池,
  3. 由 Worker 线程池控制执行工作事件，在里面执行 processor.process()，在 process 里调用service()，交由连接器组件分发处理请求,调用连接器适配器 CoyoteAdapter.service()，最终由连接器分发到各个组件处理。

image-20230109220435952

2. 处理一次 Http 请求的流程

2.1 连接器处理请求的流程

CoyoteAdaptor 组件负责将 Connector 组件和 Engine 容器关联起来，把生成的 Request对象和响应对象 Response 传递到 Engine 容器中，调用 Pipeline。

上文我们分析过 Processor 会调用 Adapter.service() 方法，请求会经过 CoyoteAdaptor 组件，如下代码所示。

   @Override
    public void service(org.apache.coyote.Request req, org.apache.coyote.Response res)
        throws Exception {
        // 根据coyote框架的request和response对象，生成connector的request和response对象（是HttpServletRequest和HttpServletResponse的封装）
        Request request = (Request) req.getNote(ADAPTER_NOTES);
        Response response = (Response) res.getNote(ADAPTER_NOTES);

        if (request == null) {
            // Create objects
            request = connector.createRequest();
            request.setCoyoteRequest(req);
            response = connector.createResponse();
            response.setCoyoteResponse(res);

            // Link objects
            request.setResponse(response);
            response.setRequest(request);

            // Set as notes
            req.setNote(ADAPTER_NOTES, request);
            res.setNote(ADAPTER_NOTES, response);

            // Set query string encoding
            req.getParameters().setQueryStringCharset(connector.getURICharset());
        }

        // 补充header
        if (connector.getXpoweredBy()) {
            response.addHeader("X-Powered-By", POWERED_BY);
        }

        boolean async = false;
        boolean postParseSuccess = false;

        req.getRequestProcessor().setWorkerThreadName(THREAD_NAME.get());
        req.setRequestThread();

        try {
            // Parse and set Catalina and configuration specific
            // request parameters
            // 用来处理请求映射 (获取 host, context, wrapper, URI 后面的参数的解析, sessionId )
            postParseSuccess = postParseRequest(req, request, res, response);
            if (postParseSuccess) {
                //check valves if we support async
                // 检查Value是否支持异步
                request.setAsyncSupported(
                    connector.getService().getContainer().getPipeline().isAsyncSupported());
                // Calling the container
                // 得到Engine首个Value()
                // 每一级Container的基础Value在完成自身处理情况下，同时要保证启动下一级Container的Value链的执行，由于请求映射已经将映射结果保存到请求对象中，因此Value直接从请求中获取下一级Container即可
                // 真正进入容器的地方，调用Engine容器下pipeline的阀门
                connector.getService().getContainer().getPipeline().getFirst().invoke(
                    request, response);
            }
            if (request.isAsync()) {
                async = true;
                ReadListener readListener = req.getReadListener();
                if (readListener != null && request.isFinished()) {
                    // Possible the all data may have been read during service()
                    // method so this needs to be checked here
                    ClassLoader oldCL = null;
                    try {
                        oldCL = request.getContext().bind(false, null);
                        if (req.sendAllDataReadEvent()) {
                            req.getReadListener().onAllDataRead();
                        }
                    } finally {
                        request.getContext().unbind(false, oldCL);
                    }
                }

                Throwable throwable =
                    (Throwable) request.getAttribute(RequestDispatcher.ERROR_EXCEPTION);

                if (!request.isAsyncCompleting() && throwable != null) {
                    request.getAsyncContextInternal().setErrorState(throwable, true);
                }
            } else {
                //  // 通过request.finishRequest 与 response.finishResponse(将OutputBuffer中的数据写到浏览器) 来完成整个请求
                request.finishRequest();
                response.finishResponse();
            }

        } catch (IOException e) {
            // Ignore
        } finally {
            AtomicBoolean error = new AtomicBoolean(false);
            res.action(ActionCode.IS_ERROR, error);

            if (request.isAsyncCompleting() && error.get()) {
                // Connection will be forcibly closed which will prevent
                // completion happening at the usual point. Need to trigger
                // call to onComplete() here.
                res.action(ActionCode.ASYNC_POST_PROCESS, null);
                async = false;
            }

            req.getRequestProcessor().setWorkerThreadName(null);
            req.clearRequestThread();

            // Recycle the wrapper request and response
            if (!async) {
                updateWrapperErrorCount(request, response);
                request.recycle();
                response.recycle();
            }
        }

上述代码只要做了这么几件事:

  1. 处理请求映射 (获取 host, context, wrapper, URI 后面的参数的解析, sessionId )
  2. 解析请求，该方法会出现代理服务器、设置必要的 header 等操作
  3. 调用 Engine 容器下 pipeline 的阀门，请求处理交由容器
  4. 通过request.finishRequest 与 response.finishResponse(将OutputBuffer中的数据写到浏览器) 来完成整个请求

2.2 各容器处理请求的流程

在《Tomcat源码篇之启动流程中》我们说过 StandardEngine 的构造函数为自己的 Pipeline 添加了基本阀 StandardEngineValve

Engine请求处理

所以我们直接来看 StandardEngineValve 的 invoke 方法

inal class StandardEngineValve extends ValveBase {

    //------------------------------------------------------ Constructor
    public StandardEngineValve() {
        super(true);
    }


    // --------------------------------------------------------- Public Methods
    @Override
    public final void invoke(Request request, Response response)
        throws IOException, ServletException {

        // 从请求中获取Host
        Host host = request.getHost();
        // 如果不存在，则返回404
        if (host == null) {
            if (!response.isError()) {
                response.sendError(404);
            }
            return;
        }
        if (request.isAsyncSupported()) {
            request.setAsyncSupported(host.getPipeline().isAsyncSupported());
        }

        // 调用Host容器下pipeline的阀门
        host.getPipeline().getFirst().invoke(request, response);
    }
}

该方法主要是选择合适的 Host，然后调用 Host 中 pipeline 的第一个 Valve 的 invoke() 方法。


Host请求处理

host.getPipeline().getFirst().invoke(request, response)，可以看到 Host 容器先获取自己的管道，再获取第一个阀门。

StandardHost 的构造函数为自己的 Pipeline 添加了基本阀 StandardHostValve,同时 Host 管道中还存在 ErrorReportValve 与 StandardHostValve 两个 Valve 处于管道前面，基本阀门在管道最后。

ErrorReportValve 主要是检测 Http 请求过程中是否出现过什么异常, 有异常的话, 直接拼装 html 页面, 输出到客户端。

public class ErrorReportValve extends ValveBase {
    public ErrorReportValve() {
        super(true);
    }


    // --------------------------------------------------------- Public Methods

      @Override
    public void invoke(Request request, Response response) throws IOException, ServletException {

        // 将请求转发给下一个 Valve
        getNext().invoke(request, response);
    
             // 正常处理结束   
        if (response.isCommitted()) {
            if (response.setErrorReported()) {
                // Error wasn't previously reported but we can't write an error
                // page because the response has already been committed.

                // See if IO is allowed
                AtomicBoolean ioAllowed = new AtomicBoolean(true);
                response.getCoyoteResponse().action(ActionCode.IS_IO_ALLOWED, ioAllowed);

                if (ioAllowed.get()) {
                    // I/O is currently still allowed. Flush any data that is
                    // still to be written to the client.
                    try {
                        response.flushBuffer();
                    } catch (Throwable t) {
                        ExceptionUtils.handleThrowable(t);
                    }
                    // Now close immediately to signal to the client that
                    // something went wrong
                    response.getCoyoteResponse().action(ActionCode.CLOSE_NOW,
                            request.getAttribute(RequestDispatcher.ERROR_EXCEPTION));
                }
            }
            return;
        }

        // 判断请求过程中是否有异常发生
        Throwable throwable = (Throwable) request.getAttribute(RequestDispatcher.ERROR_EXCEPTION);

         if (request.isAsync() && !request.isAsyncCompleting()) {
            return;
        }

        if (throwable != null && !response.isError()) {
           // 重置 response 里面的数据
            response.reset();
            // 500 错误码
            response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
        }

         response.setSuspended(false);

        try {
            // 这里就是将 异常的堆栈信息组合成 html 页面, 输出到前台 
            report(request, response, throwable);
        } catch (Throwable tt) {
            ExceptionUtils.handleThrowable(tt);
        }
    }

}

来看下 StandardHostValve 的 invoke 方法的实现。

final class StandardHostValve extends ValveBase {
  
    public StandardHostValve() {
        super(true);
    }

    @Override
    public final void invoke(Request request, Response response)
        throws IOException, ServletException {

        // 从请求中获取Context
        Context context = request.getContext();
        if (context == null) {
            // Don't overwrite an existing error
            if (!response.isError()) {
                response.sendError(404);
            }
            return;
        }

        if (request.isAsyncSupported()) {
            request.setAsyncSupported(context.getPipeline().isAsyncSupported());
        }

        boolean asyncAtStart = request.isAsync();

        try {
            context.bind(Globals.IS_SECURITY_ENABLED, MY_CLASSLOADER);

            if (!asyncAtStart && !context.fireRequestInitEvent(request.getRequest())) {
                return;
            }

            try {
                if (!response.isErrorReportRequired()) {
                    // 调用Context容器下pipeline的阀门
                    context.getPipeline().getFirst().invoke(request, response);
                }
            } catch (Throwable t) {
                ExceptionUtils.handleThrowable(t);
                container.getLogger().error("Exception Processing " + request.getRequestURI(), t);
                if (!response.isErrorReportRequired()) {
                    request.setAttribute(RequestDispatcher.ERROR_EXCEPTION, t);
                    throwable(request, response, t);
                }
            }

             response.setSuspended(false);

            Throwable t = (Throwable) request.getAttribute(RequestDispatcher.ERROR_EXCEPTION);

            if (!context.getState().isAvailable()) {
                return;
            }

            // Look for (and render if found) an application level error page
            if (response.isErrorReportRequired()) {
                // If an error has occurred that prevents further I/O, don't waste time
                // producing an error report that will never be read
                AtomicBoolean result = new AtomicBoolean(false);
                response.getCoyoteResponse().action(ActionCode.IS_IO_ALLOWED, result);
                if (result.get()) {
                    if (t != null) {
                        throwable(request, response, t);
                    } else {
                        status(request, response);
                    }
                }
            }

            if (!request.isAsync() && !asyncAtStart) {
                context.fireRequestDestroyEvent(request.getRequest());
            }
        } finally {
            // Access a session (if present) to update last accessed time, based
            // on a strict interpretation of the specification
            if (ACCESS_SESSION) {
                request.getSession(false);
            }

            context.unbind(Globals.IS_SECURITY_ENABLED, MY_CLASSLOADER);
        }
    
  
}

StandardHostValve 首先校验了 Request 是否存在 Context（在执行 CoyoteAdapter.postParseRequest 方法的时候设置进去），如果 Context 不存在，就返回 500 的错误码，然后调用了 Context 容器下 pipeline 的阀门

context.getPipeline().getFirst().invoke(request, response);，可以看到 Context 容器先获取自己的管道，再获取第一个阀门。


Context请求处理

同理，我们来看 StandardContextValve 的 invoke 实现。

final class StandardContextValve extends ValveBase {

    private static final StringManager sm = StringManager.getManager(StandardContextValve.class);

    public StandardContextValve() {
        super(true);
    }


    @Override
    public final void invoke(Request request, Response response)
        throws IOException, ServletException {

        // Disallow any direct access to resources under WEB-INF or META-INF
        MessageBytes requestPathMB = request.getRequestPathMB();
        if ((requestPathMB.startsWithIgnoreCase("/META-INF/", 0))
            || (requestPathMB.equalsIgnoreCase("/META-INF"))
            || (requestPathMB.startsWithIgnoreCase("/WEB-INF/", 0))
            || (requestPathMB.equalsIgnoreCase("/WEB-INF"))) {
            response.sendError(HttpServletResponse.SC_NOT_FOUND);
            return;
        }

        // Select the Wrapper to be used for this Request
        // 获取请求中的Wrapper
        Wrapper wrapper = request.getWrapper();
        if (wrapper == null || wrapper.isUnavailable()) {
            response.sendError(HttpServletResponse.SC_NOT_FOUND);
            return;
        }

        // Acknowledge the request
        try {
            response.sendAcknowledgement(ContinueResponseTiming.IMMEDIATELY);
        } catch (IOException ioe) {
            container.getLogger().error(sm.getString(
                "standardContextValve.acknowledgeException"), ioe);
            request.setAttribute(RequestDispatcher.ERROR_EXCEPTION, ioe);
            response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
            return;
        }

        if (request.isAsyncSupported()) {
            request.setAsyncSupported(wrapper.getPipeline().isAsyncSupported());
        }
        // 调用Wrapper容器下pipeline的阀门
        wrapper.getPipeline().getFirst().invoke(request, response);
    }
}

同理调用了 Wrapper 容器下的 pipeline 阀门


Wrapper请求处理

Wrapper 基础阀门为 StandardWrapperValve，来看下 StandardWrapperValve 的 invoke 实现

final class StandardWrapperValve extends ValveBase {
   @Override
    public final void invoke(Request request, Response response)
        throws IOException, ServletException {

        // Initialize local variables we may need
        boolean unavailable = false;
        Throwable throwable = null;
        // This should be a Request attribute...
        long t1 = System.currentTimeMillis();
        requestCount.incrementAndGet();
        StandardWrapper wrapper = (StandardWrapper) getContainer();
        Servlet servlet = null;
        Context context = (Context) wrapper.getParent();

        // Check for the application being marked unavailable
        if (!context.getState().isAvailable()) {
            response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE,
                sm.getString("standardContext.isUnavailable"));
            unavailable = true;
        }

        // Check for the servlet being marked unavailable
        if (!unavailable && wrapper.isUnavailable()) {
            container.getLogger().info(sm.getString("standardWrapper.isUnavailable",
                wrapper.getName()));
            long available = wrapper.getAvailable();
            if ((available > 0L) && (available < Long.MAX_VALUE)) {
                response.setDateHeader("Retry-After", available);
                response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE,
                    sm.getString("standardWrapper.isUnavailable",
                        wrapper.getName()));
            } else if (available == Long.MAX_VALUE) {
                response.sendError(HttpServletResponse.SC_NOT_FOUND,
                    sm.getString("standardWrapper.notFound",
                        wrapper.getName()));
            }
            unavailable = true;
        }

        // Allocate a servlet instance to process this request
        try {
            if (!unavailable) {
                // 这儿调用Wrapper的allocate()方法分配一个Servlet实例
                servlet = wrapper.allocate();
            }
        } catch (UnavailableException e) {
            container.getLogger().error(
                sm.getString("standardWrapper.allocateException",
                    wrapper.getName()), e);
            long available = wrapper.getAvailable();
            if ((available > 0L) && (available < Long.MAX_VALUE)) {
                response.setDateHeader("Retry-After", available);
                response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE,
                    sm.getString("standardWrapper.isUnavailable",
                        wrapper.getName()));
            } else if (available == Long.MAX_VALUE) {
                response.sendError(HttpServletResponse.SC_NOT_FOUND,
                    sm.getString("standardWrapper.notFound",
                        wrapper.getName()));
            }
        } catch (ServletException e) {
            container.getLogger().error(sm.getString("standardWrapper.allocateException",
                wrapper.getName()), StandardWrapper.getRootCause(e));
            throwable = e;
            exception(request, response, e);
        } catch (Throwable e) {
            ExceptionUtils.handleThrowable(e);
            container.getLogger().error(sm.getString("standardWrapper.allocateException",
                wrapper.getName()), e);
            throwable = e;
            exception(request, response, e);
            servlet = null;
        }

        MessageBytes requestPathMB = request.getRequestPathMB();
        DispatcherType dispatcherType = DispatcherType.REQUEST;
        if (request.getDispatcherType() == DispatcherType.ASYNC) {
            dispatcherType = DispatcherType.ASYNC;
        }
        request.setAttribute(Globals.DISPATCHER_TYPE_ATTR, dispatcherType);
        request.setAttribute(Globals.DISPATCHER_REQUEST_PATH_ATTR,
            requestPathMB);
        // Create the filter chain for this request
        // 创建过滤器链，类似于Pipeline的功能
        ApplicationFilterChain filterChain =
            ApplicationFilterFactory.createFilterChain(request, wrapper, servlet);

        // Call the filter chain for this request
        // NOTE: This also calls the servlet's service() method
        Container container = this.container;
        try {
            if ((servlet != null) && (filterChain != null)) {
                // Swallow output if needed
                if (context.getSwallowOutput()) {
                    try {
                        SystemLogHandler.startCapture();
                        if (request.isAsyncDispatching()) {
                            request.getAsyncContextInternal().doInternalDispatch();
                        } else {
                            // 调用过滤器链的doFilter，最终会调用到Servlet的service方法
                            filterChain.doFilter(request.getRequest(),
                                response.getResponse());
                        }
                    } finally {
                        String log = SystemLogHandler.stopCapture();
                        if (log != null && log.length() > 0) {
                            context.getLogger().info(log);
                        }
                    }
                } else {
                    if (request.isAsyncDispatching()) {
                        request.getAsyncContextInternal().doInternalDispatch();
                    } else {
                        // 调用过滤器链的doFilter，最终会调用到Servlet的service方法
                        filterChain.doFilter
                            (request.getRequest(), response.getResponse());
                    }
                }

            }
        } catch (ClientAbortException | CloseNowException e) {
            if (container.getLogger().isDebugEnabled()) {
                container.getLogger().debug(sm.getString(
                    "standardWrapper.serviceException", wrapper.getName(),
                    context.getName()), e);
            }
            throwable = e;
            exception(request, response, e);
        } catch (IOException e) {
            container.getLogger().error(sm.getString(
                "standardWrapper.serviceException", wrapper.getName(),
                context.getName()), e);
            throwable = e;
            exception(request, response, e);
        } catch (UnavailableException e) {
            container.getLogger().error(sm.getString(
                "standardWrapper.serviceException", wrapper.getName(),
                context.getName()), e);
            //            throwable = e;
            //            exception(request, response, e);
            wrapper.unavailable(e);
            long available = wrapper.getAvailable();
            if ((available > 0L) && (available < Long.MAX_VALUE)) {
                response.setDateHeader("Retry-After", available);
                response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE,
                    sm.getString("standardWrapper.isUnavailable",
                        wrapper.getName()));
            } else if (available == Long.MAX_VALUE) {
                response.sendError(HttpServletResponse.SC_NOT_FOUND,
                    sm.getString("standardWrapper.notFound",
                        wrapper.getName()));
            }
            // Do not save exception in 'throwable', because we
            // do not want to do exception(request, response, e) processing
        } catch (ServletException e) {
            Throwable rootCause = StandardWrapper.getRootCause(e);
            if (!(rootCause instanceof ClientAbortException)) {
                container.getLogger().error(sm.getString(
                        "standardWrapper.serviceExceptionRoot",
                        wrapper.getName(), context.getName(), e.getMessage()),
                    rootCause);
            }
            throwable = e;
            exception(request, response, e);
        } catch (Throwable e) {
            ExceptionUtils.handleThrowable(e);
            container.getLogger().error(sm.getString(
                "standardWrapper.serviceException", wrapper.getName(),
                context.getName()), e);
            throwable = e;
            exception(request, response, e);
        } finally {
            // Release the filter chain (if any) for this request
            if (filterChain != null) {
                filterChain.release();
            }

            // Deallocate the allocated servlet instance
            try {
                if (servlet != null) {
                    // 释放掉Servlet及相关资源
                    wrapper.deallocate(servlet);
                }
            } catch (Throwable e) {
                ExceptionUtils.handleThrowable(e);
                container.getLogger().error(sm.getString("standardWrapper.deallocateException",
                    wrapper.getName()), e);
                if (throwable == null) {
                    throwable = e;
                    exception(request, response, e);
                }
            }

            // If this servlet has been marked permanently unavailable,
            // unload it and release this instance
            try {
                if ((servlet != null) &&
                    (wrapper.getAvailable() == Long.MAX_VALUE)) {
                    // 如果servlet被标记为永远不可达，则需要卸载掉它，并释放这个servlet实例
                    wrapper.unload();
                }
            } catch (Throwable e) {
                ExceptionUtils.handleThrowable(e);
                container.getLogger().error(sm.getString("standardWrapper.unloadException",
                    wrapper.getName()), e);
                if (throwable == null) {
                    exception(request, response, e);
                }
            }
            long t2 = System.currentTimeMillis();

            long time = t2 - t1;
            processingTime += time;
            if (time > maxTime) {
                maxTime = time;
            }
            if (time < minTime) {
                minTime = time;
            }
        }
    }
 
}

我们都知道 Wrapper 实际上是 servlet 的上层包装类，与 servlet 打交道。

上述代码实现了：

  1. Servlet 实例的分配 servlet = wrapper.allocate();

  2. 创建过滤器链 `ApplicationFilterChain filterChain =

    ApplicationFilterFactory.createFilterChain(request, wrapper, servlet);`

  3. 调用过滤器链的 doFilter，最终会调用到 Servlet 的 service 方法 `filterChain.doFilter(request.getRequest(),

    response.getResponse());`
  4. 释放掉过滤器链及其相关资源，释放掉Servlet及相关资源，整个请求链路结束。


首先我们详细来看下 Servlet 实例是如何分配的

    @Override
    public Servlet allocate() throws ServletException {

        // If we are currently unloading this servlet, throw an exception
        // 卸载过程中，不能分配Servlet
        if (unloading) {
            throw new ServletException(sm.getString("standardWrapper.unloading", getName()));
        }

        boolean newInstance = false;

        // If not SingleThreadedModel, return the same instance every time
        // 如果Wrapper没有实现SingleThreadedModel，则每次都会返回同一个Servlet
        if (!singleThreadModel) {
            // Load and initialize our instance if necessary
            // 实例为null或者实例还未初始化，使用synchronized来保证并发时的原子性
            if (instance == null || !instanceInitialized) {
                synchronized (this) {
                    if (instance == null) {
                        try {
                            if (log.isDebugEnabled()) {
                                log.debug("Allocating non-STM instance");
                            }

                            // Note: We don't know if the Servlet implements
                            // SingleThreadModel until we have loaded it.
                            // 加载Servlet
                            instance = loadServlet();
                            newInstance = true;
                            if (!singleThreadModel) {
                                // For non-STM, increment here to prevent a race
                                // condition with unload. Bug 43683, test case
                                // #3
                                countAllocated.incrementAndGet();
                            }
                        } catch (ServletException e) {
                            throw e;
                        } catch (Throwable e) {
                            ExceptionUtils.handleThrowable(e);
                            throw new ServletException(sm.getString("standardWrapper.allocate"), e);
                        }
                    }
                    // 初始化Servlet
                    if (!instanceInitialized) {
                        initServlet(instance);
                    }
                }
            }
            // 非单线程模型，直接返回已经创建的Servlet，也就是说，这种情况下只会创建一个Servlet
            if (singleThreadModel) {
                if (newInstance) {
                    // Have to do this outside of the sync above to prevent a
                    // possible deadlock
                    synchronized (instancePool) {
                        instancePool.push(instance);
                        nInstances++;
                    }
                }
            } else {
                if (log.isTraceEnabled()) {
                    log.trace("  Returning non-STM instance");
                }
                // For new instances, count will have been incremented at the
                // time of creation
                if (!newInstance) {
                    countAllocated.incrementAndGet();
                }
                return instance;
            }
        }

        // 如果是单线程模式，则使用servlet对象池技术来加载多个Servlet
        synchronized (instancePool) {
            while (countAllocated.get() >= nInstances) {
                // Allocate a new instance if possible, or else wait
                if (nInstances < maxInstances) {
                    try {
                        instancePool.push(loadServlet());
                        nInstances++;
                    } catch (ServletException e) {
                        throw e;
                    } catch (Throwable e) {
                        ExceptionUtils.handleThrowable(e);
                        throw new ServletException(sm.getString("standardWrapper.allocate"), e);
                    }
                } else {
                    try {
                        instancePool.wait();
                    } catch (InterruptedException e) {
                        // Ignore
                    }
                }
            }
            if (log.isTraceEnabled()) {
                log.trace("  Returning allocated STM instance");
            }
            countAllocated.incrementAndGet();
            return instancePool.pop();
        }
    }

   public synchronized Servlet loadServlet() throws ServletException {

        // Nothing to do if we already have an instance or an instance pool
        if (!singleThreadModel && (instance != null)) {
            return instance;
        }

        PrintStream out = System.out;
        if (swallowOutput) {
            SystemLogHandler.startCapture();
        }

        Servlet servlet;
        try {
            long t1 = System.currentTimeMillis();
            // Complain if no servlet class has been specified
            if (servletClass == null) {
                unavailable(null);
                throw new ServletException
                    (sm.getString("standardWrapper.notClass", getName()));
            }

            // 关键的地方，就是通过实例管理器，创建Servlet实例，而实例管理器是通过特殊的类加载器来加载给定的类
            InstanceManager instanceManager = ((StandardContext) getParent()).getInstanceManager();
            try {
                servlet = (Servlet) instanceManager.newInstance(servletClass);
            } catch (ClassCastException e) {
                unavailable(null);
                // Restore the context ClassLoader
                throw new ServletException
                    (sm.getString("standardWrapper.notServlet", servletClass), e);
            } catch (Throwable e) {
                e = ExceptionUtils.unwrapInvocationTargetException(e);
                ExceptionUtils.handleThrowable(e);
                unavailable(null);

                // Added extra log statement for Bugzilla 36630:
                // https://bz.apache.org/bugzilla/show_bug.cgi?id=36630
                if (log.isDebugEnabled()) {
                    log.debug(sm.getString("standardWrapper.instantiate", servletClass), e);
                }

                // Restore the context ClassLoader
                throw new ServletException
                    (sm.getString("standardWrapper.instantiate", servletClass), e);
            }

            if (multipartConfigElement == null) {
                MultipartConfig annotation =
                    servlet.getClass().getAnnotation(MultipartConfig.class);
                if (annotation != null) {
                    multipartConfigElement =
                        new MultipartConfigElement(annotation);
                }
            }

            // Special handling for ContainerServlet instances
            // Note: The InstanceManager checks if the application is permitted
            //       to load ContainerServlets
            if (servlet instanceof ContainerServlet) {
                ((ContainerServlet) servlet).setWrapper(this);
            }

            classLoadTime = (int) (System.currentTimeMillis() - t1);

            if (servlet instanceof SingleThreadModel) {
                if (instancePool == null) {
                    instancePool = new Stack<>();
                }
                singleThreadModel = true;
            }

            // 调用Servlet的init方法
            initServlet(servlet);

            fireContainerEvent("load", this);

            loadTime = System.currentTimeMillis() - t1;
        } finally {
            if (swallowOutput) {
                String log = SystemLogHandler.stopCapture();
                if (log != null && log.length() > 0) {
                    if (getServletContext() != null) {
                        getServletContext().log(log);
                    } else {
                        out.println(log);
                    }
                }
            }
        }
        return servlet;

    }

   private synchronized void initServlet(Servlet servlet)
        throws ServletException {

        if (instanceInitialized && !singleThreadModel) {
            return;
        }

        // Call the initialization method of this servlet
        try {
            if (Globals.IS_SECURITY_ENABLED) {
                boolean success = false;
                try {
                    Object[] args = new Object[]{facade};
                    SecurityUtil.doAsPrivilege("init",
                        servlet,
                        classType,
                        args);
                    success = true;
                } finally {
                    if (!success) {
                        // destroy() will not be called, thus clear the reference now
                        SecurityUtil.remove(servlet);
                    }
                }
            } else {
                servlet.init(facade);
            }

            instanceInitialized = true;
        } catch (UnavailableException f) {
            unavailable(f);
            throw f;
        } catch (ServletException f) {
            // If the servlet wanted to be unavailable it would have
            // said so, so do not call unavailable(null).
            throw f;
        } catch (Throwable f) {
            ExceptionUtils.handleThrowable(f);
            getServletContext().log(sm.getString("standardWrapper.initException", getName()), f);
            // If the servlet wanted to be unavailable it would have
            // said so, so do not call unavailable(null).
            throw new ServletException
                (sm.getString("standardWrapper.initException", getName()), f);
        }
    }


接着来看下过滤器链的执行过程

public final class ApplicationFilterChain implements FilterChain {
   @Override
    public void doFilter(ServletRequest request, ServletResponse response)
        throws IOException, ServletException {

        if (Globals.IS_SECURITY_ENABLED) {
            final ServletRequest req = request;
            final ServletResponse res = response;
            try {
                java.security.AccessController.doPrivileged(
                    new java.security.PrivilegedExceptionAction<Void>() {
                        @Override
                        public Void run()
                            throws ServletException, IOException {
                            // 执行Filter链
                            internalDoFilter(req, res);
                            return null;
                        }
                    }
                );
            } catch (PrivilegedActionException pe) {
                Exception e = pe.getException();
                if (e instanceof ServletException) {
                    throw (ServletException) e;
                } else if (e instanceof IOException) {
                    throw (IOException) e;
                } else if (e instanceof RuntimeException) {
                    throw (RuntimeException) e;
                } else {
                    throw new ServletException(e.getMessage(), e);
                }
            }
        } else {
            // 执行Filter链
            internalDoFilter(request, response);
        }
     
}
    /**
     * `internalDoFilter`方法通过pos和n来调用过滤器链里面的每个过滤器。pos表示当前的过滤器下标，n表示总的过滤器数量
     * `internalDoFilter`方法最终会调用servlet.service()方法
     *
     * @param request
     * @param response
     * @throws IOException
     * @throws ServletException
     */
    private void internalDoFilter(ServletRequest request,
                                  ServletResponse response)
        throws IOException, ServletException {

        // Call the next filter if there is one
        // 当pos小于n时, 则执行Filter
        if (pos < n) {
            // 得到 过滤器 Filter，执行一次post++
            ApplicationFilterConfig filterConfig = filters[pos++];
            try {
                Filter filter = filterConfig.getFilter();

                if (request.isAsyncSupported() && "false".equalsIgnoreCase(
                    filterConfig.getFilterDef().getAsyncSupported())) {
                    request.setAttribute(Globals.ASYNC_SUPPORTED_ATTR, Boolean.FALSE);
                }
                if (Globals.IS_SECURITY_ENABLED) {
                    final ServletRequest req = request;
                    final ServletResponse res = response;
                    Principal principal =
                        ((HttpServletRequest) req).getUserPrincipal();

                    Object[] args = new Object[]{req, res, this};
                    SecurityUtil.doAsPrivilege("doFilter", filter, classType, args, principal);
                } else {
                    // 执行Filter
                    filter.doFilter(request, response, this);
                }
            } catch (IOException | ServletException | RuntimeException e) {
                throw e;
            } catch (Throwable e) {
                e = ExceptionUtils.unwrapInvocationTargetException(e);
                ExceptionUtils.handleThrowable(e);
                throw new ServletException(sm.getString("filterChain.filter"), e);
            }
            return;
        }

        // 当pos等于n时，过滤器都执行完毕执行下面的代码
        // We fell off the end of the chain -- call the servlet instance
        try {
            if (ApplicationDispatcher.WRAP_SAME_OBJECT) {
                lastServicedRequest.set(request);
                lastServicedResponse.set(response);
            }

            if (request.isAsyncSupported() && !servletSupportsAsync) {
                request.setAttribute(Globals.ASYNC_SUPPORTED_ATTR,
                    Boolean.FALSE);
            }
            // Use potentially wrapped request from this point
            if ((request instanceof HttpServletRequest) &&
                (response instanceof HttpServletResponse) &&
                Globals.IS_SECURITY_ENABLED) {
                final ServletRequest req = request;
                final ServletResponse res = response;
                Principal principal =
                    ((HttpServletRequest) req).getUserPrincipal();
                Object[] args = new Object[]{req, res};
                SecurityUtil.doAsPrivilege("service",
                    servlet,
                    classTypeUsedInService,
                    args,
                    principal);
            } else {
                // service执行请求
                servlet.service(request, response);
            }
        } catch (IOException | ServletException | RuntimeException e) {
            throw e;
        } catch (Throwable e) {
            e = ExceptionUtils.unwrapInvocationTargetException(e);
            ExceptionUtils.handleThrowable(e);
            throw new ServletException(sm.getString("filterChain.servlet"), e);
        } finally {
            if (ApplicationDispatcher.WRAP_SAME_OBJECT) {
                lastServicedRequest.set(null);
                lastServicedResponse.set(null);
            }
        }

最终调用servlet.service(request, response); 方法，

真正将请求交给 Servlet 处理

protected void service(HttpServletRequest req, HttpServletResponse resp)
        throws ServletException, IOException {

        String method = req.getMethod();

        if (method.equals(METHOD_GET)) {
            long lastModified = getLastModified(req);
            if (lastModified == -1) {
                 doGet(req, resp);
            } else {
                long ifModifiedSince;
                try {
                    ifModifiedSince = req.getDateHeader(HEADER_IFMODSINCE);
                } catch (IllegalArgumentException iae) {
                    ifModifiedSince = -1;
                }
                if (ifModifiedSince < (lastModified / 1000 * 1000)) {
                    maybeSetLastModified(resp, lastModified);
                    doGet(req, resp);
                } else {
                    resp.setStatus(HttpServletResponse.SC_NOT_MODIFIED);
                }
            }

        } else if (method.equals(METHOD_HEAD)) {
            long lastModified = getLastModified(req);
            maybeSetLastModified(resp, lastModified);
            doHead(req, resp);

        } else if (method.equals(METHOD_POST)) {
            doPost(req, resp);

        } else if (method.equals(METHOD_PUT)) {
            doPut(req, resp);

        } else if (method.equals(METHOD_DELETE)) {
            doDelete(req, resp);

        } else if (method.equals(METHOD_OPTIONS)) {
            doOptions(req,resp);

        } else if (method.equals(METHOD_TRACE)) {
            doTrace(req,resp);

        } else {
        
            String errMsg = lStrings.getString("http.method_not_implemented");
            Object[] errArgs = new Object[1];
            errArgs[0] = method;
            errMsg = MessageFormat.format(errMsg, errArgs);

            resp.sendError(HttpServletResponse.SC_NOT_IMPLEMENTED, errMsg);
        }
    }

至此整个请求链路结束，一次HTTP请求处理完成。

3. 小节

本文通过以 Connector 接受请求为起点，对框架源码进行了深度分析，并解答了 Tomcat 线程模型的结构的意义，以及一个 HTTP 请求经过 Tomcat 服务器的完整流程。

下图是框架接受请求各组件的时序图，以助于更好的理解。

https://www.processon.com/view/link/63c2e873e8d4677705c87b56

作者：MRyan


本文采用 知识共享署名-相同方式共享 4.0 国际许可协议 进行许可。
转载时请注明本文出处及文章链接。本文链接：https://wormholestack.com/archives/649/
吉ICP备19007414号
2024 © MRyan 107 ms