死磕Tomcat系列(2)——EndPoint源码解析

待我称王封你为后i 2021-09-01 06:30 443阅读 0赞

# 死磕Tomcat系列(2)——EndPoint源码解析 #

## 前言 ##

上一节中我们描述了Tomcat的整体架构，我们知道Tomcat分为两大组件，一个连接器和一个容器。而我们这次要讲的 Endpoint的组件就是属于连接器里面的。它是一个通信的端点，就是负责对外实现TCP/IP协议。Endpoint是个接口，它的具体实现类就是AbstractEndpoint，而AbstractEndpoint具体的实现就是AprEndpoint、Nio2Endpoint、NioEndpoint。

*  AprEndpoint：对应的是APR模式，简单理解就是从操作系统级别解决异步IO的问题，大幅度的提高服务器的处理和响应性能。但是启动这种模式需要安装一些其他依赖库。
 *  Nio2Endpoint：利用代码来实现异步IO
 *  NioEndpoint：利用java的NIO实现了非阻塞IO，Tomcat默认启动是以这个来启动的。

## NioEndpoint中重要的组件 ##

我们知道NioEndpoint的原理还是对于Linux的多路复用器的使用，而多路复用器简单来说就两个步骤；

1.  创建一个Selector，在它身上注册各种Channel，然后select方法，等待通道中有感兴趣的事情发生
2.  如果有感兴趣的事情发生了，例如是读事件，那么就想事件从通道中读出来  
    而NioEndpoint为了实现上面这两步，用了五个组件。这五个组件是：LimitLatch、Acceptor、Poller、SocketProcessor、Executor

//AbstractEndpoint.java
    
    /**
      * Thread used to accept new connections and pass them to worker threads.
    */
      protected Acceptor<U> acceptor;
    /**
     * counter for nr of connections handled by an endpoint
    */
    private volatile LimitLatch connectionLimitLatch = null;
    /**
    * External Executor based thread pool.
    */
    private Executor executor = null;
    //内部类
    SocketProcessor
    
    // NioEndpoint.java
    /**
    * The socket poller.poller内部类
    */
     private Poller poller = null;

我们可以看到在代码中定义了五个组件，这五个组件的功能为：

*  LimitLatch:连接控制器，负责控制最大的连接数
 *  Acceptor:负责接收新的连接，返回一个Channel对象给Poller
 *  Poller:可以将其看成是NIO中Selector，负责监控Channel的状态
 *  SocketProcessor:可以看成是一个被封装的任务类
 *  Executor:Tomcat自己扩展的线程池，用来执行任务类

用图简单的表示为以下关系：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70]

## LimitLatch ##

我们在上面说了：LimitLatch的主要是用来控制Tomcat所能接受的最大线程数量，如果超过此连接，那么Tomcat就会将此连接线程阻塞等待，等里面有其他连接释放了再消费此连接。那么LimitLatch是如何做到的呢？我们可以看LimitLatch这个类？

//LimitLatch.java部分代码
    
    public class LimitLatch {
        private static final Log log = LogFactory.getLog(LimitLatch.class);
        private class Sync extends  AbstractQueuedSynchronizer {
            private static final long serialVersionUID = 1L;
            public Sync() {
            }
            @Override
            protected int tryAcquireShared(int ignored) {
                long newCount = count.incrementAndGet();
                if (!released && newCount > limit) {
                    // Limit exceeded
                    count.decrementAndGet();
                    return -1;
                } else {
                    return 1;
                }
            }
            @Override
            protected boolean tryReleaseShared(int arg) {
                count.decrementAndGet();
                return true;
            }
        }
        private final Sync sync;
    	//当前连接数
        private final AtomicLong count;
    	//最大连接数
        private volatile long limit;
        private volatile boolean released = false;

我们可以看到他的内部实现了AbstractQueuedSynchronizer，AQS其实是一个框架，实现他的类可以自定义控制线程什么时候挂起什么时候释放。Limit参数就是控制的最大连接数。我们可以看到AbstractEndpoint调用LimitLatch的countUpOrAwait方法来判断是否能获取连接。

/**
    * Acquires a shared latch if one is available or waits for one if no shared
    * latch is current available.
    * @throws InterruptedException If the current thread is interrupted
    */
    public void countUpOrAwait() throws InterruptedException {
            if (log.isDebugEnabled()) {
                log.debug("Counting up["+Thread.currentThread().getName()+"] latch="+getCount());
            }
            sync.acquireSharedInterruptibly(1);
        }

ASQ是如何知道什么时候阻塞线程呢？即不能获取连接呢？这些就要靠用户自己实现AbstractQueuedSynchronizer自己定义什么时候获取连接，什么时候释放连接了。可以看到Sync类重写了tryAcquireShared和tryReleaseShared方法。在tryReleaseShared方法中定义了一旦当前连接数大于设置的最大连接数，那么就返回-1表示将此线程放入到ASQ队列中等待。

## Acceptor ##

Acceptor是接受连接的，我们可以看到Acceptor实现了Runnable接口，那么在哪会新开启线程来执行Acceptor的run方法呢？在AbstractEndpoint的startAcceptorThreads方法中。

protected void startAcceptorThread() {
            acceptor = new Acceptor<>(this);
            String threadName = getName() + "-Acceptor";
            acceptor.setThreadName(threadName);
            Thread t = new Thread(acceptor, threadName);
            t.setPriority(getAcceptorThreadPriority());
            t.setDaemon(getDaemon());
            t.start();
        }

可以看到这里可以设置开启几个Acceptor,默认是一个。而端口只能对应一个ServerSocketChannel，那么SerSocketChannel在哪初始化呢？我们可以看到在acceptor = new Acceptor<>(this);这句话中传入了this进去，那么应该是由Endpoint组件初始化的连接。在NioEndpoint的initSeverSocket方法中初始化连接。

// Separated out to make it easier for folks that extend NioEndpoint to
        // implement custom [server]sockets
        protected void initServerSocket() throws Exception {
            if (!getUseInheritedChannel()) {
                serverSock = ServerSocketChannel.open();
                socketProperties.setProperties(serverSock.socket());
                InetSocketAddress addr = new InetSocketAddress(getAddress(), getPortWithOffset());
                serverSock.socket().bind(addr,getAcceptCount());
            } else {
                // Retrieve the channel provided by the OS
                Channel ic = System.inheritedChannel();
                if (ic instanceof ServerSocketChannel) {
                    serverSock = (ServerSocketChannel) ic;
                }
                if (serverSock == null) {
                    throw new IllegalArgumentException(sm.getString("endpoint.init.bind.inherited"));
                }
            }
            serverSock.configureBlocking(true); //mimic APR behavior
        }

在这里我们可以看到两点：

1.  在bind方法中的第二个参数表示操作系统的等待队列长度，即Tomcat不在接受连接时（达到了设置的最大连接数），但是在操作系统层面还是能够接受连接的，对此就将此连接信息放入等待队列，那么这个队列的大小就是此参数设置的。
2.  ServerSocketChannel被设置成了阻塞的模式，也就是说以阻塞方式接受连接的。或许会有疑问。在平时的NIO编程中Channel不是都要设置成非阻塞模式的吗？这里解释一下，如果是设置成了非阻塞模式那么就必须设置一个Selector不断的轮询，但是接受连接只需要阻塞一个通道即可。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 1]  
    这里需要注意一点，每个Acceptor在生成PollerEvent对象放入Poller队列中时都是随机取出Poller对象的，具体代码如下，所以Poller中的Queue对象设置成了SynchronizedQueue,因为可能有多个Acceptor同时向此Poller的队列中放入PollerEvent对象。

public Poller getPoller0() { 
    	if (pollerThreadCount == 1) { 
    		return pollers[0]; 
    	} else { 
    	int idx = Math.abs(pollerRotater.incrementAndGet()) % pollers.length; return pollers[idx]; 
    	} 
    }

**什么是操作系统级别的连接呢？在TCP的三次握手中，系统通常会每一个LISTEN状态的Socket维护两个队列，一个是半连接队列（SYN）：这些连接已经收到客户端SYN；另一个是全连接队列（ACCEPT）：这些连接已经收到客户端（ACK），完成三次握手，等待被应用调用accept方法取走使用**。

所有的Acceptor共用这一个连接，在Acceptor的run方法中，放一些重要代码。

@Override
        public void run() {
            int errorDelay = 0;
            // Loop until we receive a shutdown command
            while (endpoint.isRunning()) {
                // Loop if endpoint is paused
                while (endpoint.isPaused() && endpoint.isRunning()) {
                    state = AcceptorState.PAUSED;
                    try {
                        Thread.sleep(50);
                    } catch (InterruptedException e) {
                        // Ignore
                    }
                }
                if (!endpoint.isRunning()) {
                    break;
                }
                state = AcceptorState.RUNNING;
                try {
                    //if we have reached max connections, 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 = 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 (endpoint.isRunning() && !endpoint.isPaused()) {
                        // setSocketOptions() will hand the socket off to
                        // an appropriate processor if successful
                        if (!endpoint.setSocketOptions(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 Error) {
                        Error e = (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);
                    }
                }
            }
            state = AcceptorState.ENDED;
        }

里面我们可以得到两点：

1.  运行时会判断是否达到最大连接数，如果到达了那么就阻塞线程等待，里面调用的是LimitLatch组件来判断的。
2.  最重要的是socket到了这一步了，是endpoint.setSocketOptions(socket)这段代码

//NioEndpoint.java
    
    protected boolean setSocketOptions(SocketChannel socket) {
            NioChannel channel = null;
            boolean success = false;
            // Process the connection
            try {
                // Disable blocking, polling will be used
                socket.configureBlocking(false);
                Socket sock = socket.socket();
                socketProperties.setProperties(sock);
    
                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, selectorPool, this);
                    } else {
                        channel = new NioChannel(bufhandler);
                    }
                }
                NioSocketWrapper socketWrapper = new NioSocketWrapper(channel, this);
                connections.put(channel, socketWrapper);
                channel.reset(socket, socketWrapper);
                socketWrapper.setReadTimeout(getConnectionTimeout());
                socketWrapper.setWriteTimeout(getConnectionTimeout());
                socketWrapper.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
                socketWrapper.setSecure(isSSLEnabled());
                poller.register(channel, socketWrapper);
                success = true;
            } catch (Throwable t) {
                ExceptionUtils.handleThrowable(t);
                try {
                    log.error(sm.getString("endpoint.socketOptionsError"), t);
                } catch (Throwable tt) {
                    ExceptionUtils.handleThrowable(tt);
                }
            } finally {
                if (!success && channel != null) {
                    connections.remove(channel);
                    channel.free();
                }
            }
            // Tell to close the socket if needed
            return success;
        }

其实里面最终要的就是将Acceptor与一个Poller绑定起来，然后两个组件通过队列通信，每个Poller都维护着一个SynchronizedQueue队列，ChannelEvent放入到队列中，然后Poller从队列中取出事件进行消费。

## Poller ##

我们可以看到Poller是NioException的内部类，而他也是实现了Runnable接口，可以看到在其类中维护了一个Queue和Selector，定义如下。所以本质上Poller就是Sellctor。

private Selector selector;
    private final SynchronizedQueue<PollerEvent> events =new SynchronizedQueue<>();

其重点在run方法里面：

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(sm.getString("endpoint.nio.selectorLoopError"), 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 socketWrapper = (NioSocketWrapper) sk.attachment();
                        // Attachment may be null if another thread has called
                        // cancelledKey()
                        if (socketWrapper == null) {
                            iterator.remove();
                        } else {
                            iterator.remove();
                            processKey(sk, socketWrapper);
                        }
                    }
                    // Process timeouts
                    timeout(keyCount,hasEvents);
                }
                getStopLatch().countDown();
            }

其中主要是调用了event()方法，就是不断的查看队列中是否有PollerEvent事件，如果有的话就将其取出然后把里面的Channel取出来注册到该Selector中然后不断的轮询注册过的Channel查看是否有事件发生。

## SocketProcessor ##

我们知道Poller在轮询Channel有事件发生时，就会调用将此事件封装起来，然后交给线程池去执行。那么这个包装类就是SocketProcessor。而我们打开此类，能够看到它也实现了Runnable接口，用来定义线程池Executor中线程所执行的任务。那么这里是如何将Channel中的字节流转换为Tomcat需要的ServletRequest对象呢？其实就是调用了Http11Processor来进行字节流与对象的转换的

## Executor ##

Executor其实是Tomcat定制版的线程池。我们可以看它的类的定义，可以发现它其实是扩展了Java的线程池。

public interface Executor extends java.util.concurrent.Executor, Lifecycle

在线程池中最重要的两个参数就是核心线程数和最大线程数，正常的Java线程池的执行流程是这样的

1.  如果当前线程小于核心线程数，那么来一个任务就创建一个线程。
2.  如果当前线程大于核心线程数，那么就再来任务就将任务放入到任务队列中。所有线程抢任务。
3.  如果队列满了，那么就开始创建临时线程。
4.  如果总线程数到了最大的线程数并且队列也满了，那么就抛出异常  
    但是在Tomcat自定义的线程池中是不一样的，通过重写了execute方法实现了自己的任务处理逻辑

*  . 如果当前线程小于核心线程数，那么来一个任务就创建一个线程。
 *  . 如果当前线程大于核心线程数，那么就再来任务就将任务放入到任务队列中。所有线程抢任务。
 *  . 如果队列满了，那么就开始创建临时线程。
 *  . 如果总线程数到了最大的线程数，再次获得任务队列，再尝试一次将任务加入队列中。
 *  . 如果此时还是满的，就抛异常。  
    差别就在于第四步的差别，原生线程池的处理策略是只要当前线程数大于最大线程数，那么就抛异常，而Tomcat的则是如果当前线程数大于最大线程数，就再尝试一次，如果还是满的才会抛异常。下面是定制化线程池execute的执行逻辑。

//ExecutorFactory.java
    public void execute(Runnable command) {
                try {
                    super.execute(command);
                } catch (RejectedExecutionException rx) {
                    if (super.getQueue() instanceof TaskQueue) {
                        TaskQueue queue = (TaskQueue)super.getQueue();
                        if (!queue.force(command)) {
                            throw new RejectedExecutionException(sm.getString("executorFactory.queue.full"));
                        }
                    }
                }
            }

那么如何规定任务队列是否已满呢？如果设置了队列的最大长度当然好了，但是Tomcat默认情况下是没有设置，所以默认是无限的。所以Tomcat的TaskQueue继承了LinkedBlockingQueue，重写了offer方法，在里面定义了什么时候返回false。

public boolean offer(Runnable o) {
                //we can't do any checks
                if (parent==null) return super.offer(o);
                //we are maxed out on threads, simply queue the object
                if (parent.getPoolSize() == parent.getMaximumPoolSize()) return super.offer(o);
                //we have idle threads, just add it to the queue
                //this is an approximation, so it could use some tuning
                if (parent.getActiveCount()<(parent.getPoolSize())) return super.offer(o);
                //if we have less threads than maximum force creation of a new thread
                if (parent.getPoolSize()<parent.getMaximumPoolSize()) return false;
                //if we reached here, we need to add it to the queue
                return super.offer(o);
            }

这就是submittedCount的意义，目的就是为了在任务队列长度无限的情况下，让线程池有机会创建新的线程。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70]: /images/20210901/e652779472674cd6be37c85a9ecac978.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNDE5MTA1_size_16_color_FFFFFF_t_70 1]: /images/20210901/5286706f8e3a4fdbb6dd5afcb9338676.png