阅读 103

Tomcat9的Connector组件(二)

Acceptor控制与tomcat建立连接的数量,但只负责建立连接。Acceptor实现了Runnable接口,主要用于接收网络请求,建立连接。

Acceptor是Endpoint的内部类,主要的职责就是监听是否有客户端套接字连接并接收socket,再将socket交由任务执行者(Executor)执行,不断从系统底层读取socket,接着做尽可能少的处理(最好就是接收到后不做任何处理),最后扔进线程池。为什么强调要做尽可能少的处理?这里关系到系统性能问题,过多的处理会严重影响吞吐量。因为tomcat默认只有一个接收器(一条线程负责套接字接收工作),所以它对每次接收处理的时间长短将很可能对整体性能产生影响。于是接收器所干的活都是非常少且简单的,仅仅维护了几个状态变量、流量控制闸门的累加操作、serverSocket的接收操作、设置接收到的socket的一些属性、将接收到的socket放入线程池以及一些异常处理。其他需要较长时间处理的逻辑就交给了线程池,例如对socket底层数据的读取,对http协议报文的解析及响应客户端的一些操作等等,很精妙地将事务剥离,远远提升了系统处理响应性能。Tomcat必然面对用户并发请求,因此这里Socket的处理通过新的线程池来处理。


Acceptor监听到网络请求,SocketChannel——》SocketProcessor线程,Executor执行线程。SocketProcessor线程调用ConnectionHandler进行业务处理。


org.apache.tomcat.util.net.Acceptor:

@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;
}复制代码

org.apache.tomcat.util.net.NioEndpoint:

连接建立之后,将一个SocketChannel对象包装成一个NioChannel,并注册到Poller中,由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();
        }
        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;
}复制代码

org.apache.tomcat.util.net.NioEndpoint.Poller:

Acceptor封装后的NioChannel放入Poller线程内部维护的一个PollerEvent队列中,然后在Poller线程运行时处理队列,将NioChannel注册到这个Poller的Selector上。

private Selector selector;
private final SynchronizedQueue<PollerEvent> events =
        new SynchronizedQueue<>();
/**
 * Registers a newly created socket with the poller.
 *
 * @param socket    The newly created socket
 */
public void register(final NioChannel socket) {
    socket.setPoller(this);
    NioSocketWrapper ka = new NioSocketWrapper(socket, NioEndpoint.this);
    socket.setSocketWrapper(ka);
    ka.setPoller(this);
    ka.setReadTimeout(getConnectionTimeout());
    ka.setWriteTimeout(getConnectionTimeout());
    ka.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
    ka.setSecure(isSSLEnabled());
    PollerEvent r = eventCache.pop();
    ka.interestOps(SelectionKey.OP_READ);//this is what OP_REGISTER turns into.
    if ( r==null) r = new PollerEvent(socket,ka,OP_REGISTER);
    else r.reset(socket,ka,OP_REGISTER);
    addEvent(r);
}

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

当事件到来的时候,Selector发现要处理的事件,通过selector.select系列方法来获取数据,然后经由processKey()到processSocket(),封装成一个SocketProcessorBase对象后,放在EndPoint的线程池中执行。

/**
 * 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.
 */
@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();
                processKey(sk, attachment);
            }
        }//while

        //process timeouts
        timeout(keyCount,hasEvents);
    }//while

    getStopLatch().countDown();
}

protected void processKey(SelectionKey sk, NioSocketWrapper attachment) {
    try {
        if ( close ) {
            cancelledKey(sk);
        } else if ( sk.isValid() && attachment != null ) {
            if (sk.isReadable() || sk.isWritable() ) {
                if ( attachment.getSendfileData() != null ) {
                    processSendfile(sk,attachment, false);
                } else {
                    unreg(sk, attachment, sk.readyOps());
                    boolean closeSocket = false;
                    // Read goes before write
                    if (sk.isReadable()) {
                        if (!processSocket(attachment, SocketEvent.OPEN_READ, true)) {
                            closeSocket = true;
                        }
                    }
                    if (!closeSocket && sk.isWritable()) {
                        if (!processSocket(attachment, SocketEvent.OPEN_WRITE, true)) {
                            closeSocket = true;
                        }
                    }
                    if (closeSocket) {
                        cancelledKey(sk);
                    }
                }
            }
        } else {
            //invalid key
            cancelledKey(sk);
        }
    } catch ( CancelledKeyException ckx ) {
        cancelledKey(sk);
    } catch (Throwable t) {
        ExceptionUtils.handleThrowable(t);
        log.error("",t);
    }
}
/**
 * Process the given SocketWrapper with the given status. Used to trigger
 * processing as if the Poller (for those endpoints that have one)
 * selected the socket.
 *
 * @param socketWrapper The socket wrapper to process
 * @param event         The socket event to be processed
 * @param dispatch      Should the processing be performed on a new
 *                          container thread
 *
 * @return if processing was triggered successfully
 */
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;
}
复制代码

org.apache.tomcat.util.net.SocketProcessorBase(任务定义器):

请求封装为SocketProcessorBase之后,首先执行run(),再执行SocketProcessor的doRun(),请求再被转移到ConnectHandler.process()。

public abstract class SocketProcessorBase<S> implements Runnable {

    protected SocketWrapperBase<S> socketWrapper;
    protected SocketEvent event;

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


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


    @Override
    public final void run() {
        synchronized (socketWrapper) {
            // It is possible that processing may be triggered for read and
            // write at the same time. The sync above makes sure that processing
            // does not occur in parallel. The test below ensures that if the
            // first event to be processed results in the socket being closed,
            // the subsequent events are not processed.
            if (socketWrapper.isClosed()) {
                return;
            }
            doRun();
        }
    }


    protected abstract void doRun();
}复制代码

org.apache.tomcat.util.net.NioEndpoint.SocketProcessor(extends SocketProcessorBase<NioChannel>):

  1. 处理socket并相应客户端。
  2. 连接数计数器减一。
  3. 关闭socket。

其中对socket的处理是最重要也是最复杂的,它包括对底层socket字节流的读取、http协议请求报文的解析(请求行、请求头、请求体等信息的解析)根据请求行解析得到路径去寻找相应主机上web项目的资源、根据处理的结果组装好http协议响应报文输出到客户端。

-> org.apache.tomcat.util.net.SocketProcessor#run

-->org.apache.coyote.AbstractProtocol.ConnectionHandler#process

--->org.apache.coyote.http11.Http11Processor#process

---->org.apache.coyote.http11.Http11InputBuffer#parseRequestLine

---->org.apache.coyote.http11.Http11InputBuffer#parseHeaders

---->org.apache.catalina.connector.CoyoteAdapter#service


Http11Processor组件提供了对Http协议通信的处理,包括对套接字的读取过滤、对http协议的解析并封装成请求对象、http响应对象的生成、套接字的过滤写入等等操作。


Socket的写入并非意味着真正写入,存在操作系统底层缓冲区。


  1. 创建ServerSocket实例,传入端口号等待阻塞。
  2. 初始化底层socket并将进行监听。
  3. 创建socket底层数据结构,socket初始状态为关闭。
  4. 填入应用层传入的端口号并且设置socket状态为监听状态。
  5. 服务端开始监听客户端的访问。
  6. 客户端访问时经过三次握手完成连接,准备接收socket连接。
  7. 为该连接创建一个新的套接字数据结构,根据到来的分组报文设置远程端口跟远程IP。由于是完成了三次握手了,所以把状态设置为连接建立
  8. 建立好连接的底层套接字数据结构会被放到一个缓冲区队列里,供应用层读取。
  9. Serversocket实例调用accept()方法后,即开始轮询队列缓冲区,一旦队列中有新的连接,则创建并返回一个应用层的socket实例。
  10. 如此进行工作,等待客户端的访问,直到Serversocket实例关闭。


  1. 确定通信目标,包括目标IP和目标端口。
  2. 根据目标IP跟端口,在Java应用层创建一个Socket实例。
  3. 阻塞等待,准备进行系统底层相关工作。
  4. 创建socket底层数据结构,socket初始状态为关闭。
  5. 向这个socket填入本地、远程的地址跟端口,并向远程服务器发送请求,此时socket的状态为正在连接。
  6. 与远程服务器完成3次握手后,就完成了连接的建立,此时的socket状态为连接建立完成。
  7. 完成应用层上的socket实例化,对这个socket进行操作,实现通信。





请求处理:www.processon.com/view/link/5…

类图:www.processon.com/view/link/5…

启动停止:www.processon.com/view/link/5…