Tomcat 源码分析 (二) : Connector

傷城~ 2022-05-15 02:44 326阅读 0赞

# NIOEndPoint #

`NIOEndPoint`的`bind()`方法开启一个`SocketServer`

@Override
        public void bind() throws Exception {
             //开启一个server socket
            serverSock = ServerSocketChannel.open();
            //根据配置文件设置server socket的属性
            socketProperties.setProperties(serverSock.socket());
            InetSocketAddress addr = (getAddress()!=null?new InetSocketAddress(getAddress(),getPort()):new InetSocketAddress(getPort()));
            serverSock.socket().bind(addr,getAcceptCount());
            serverSock.configureBlocking(true); //mimic APR behavior
    
            // Initialize thread count defaults for acceptor, poller
            if (acceptorThreadCount == 0) {
                // FIXME: Doesn't seem to work that well with multiple accept threads
                // 这个东西和下面的配置有关
                acceptorThreadCount = 1;
            }
            if (pollerThreadCount <= 0) {
                //minimum one poller thread
                pollerThreadCount = 1;
            }
            stopLatch = new CountDownLatch(pollerThreadCount);
    
            // 如果需要的话，初始化SSL
            initialiseSsl();
            //开启Selector，坚挺NIO的 IO的事件
            selectorPool.open();
        }

# Acceptor线程接收客户请求 #

在Tomcat启动的时候会启动一个Endpoint,并会调用它的startInternal方法，在这里开启了一个Acceptor的子线程。利用这个Acceptor子线程来接收Client端的Socket连接

@Override
        public void run() {
                ........................省略代码................................................
               // 这里调用上面NIOEndPoint 的serverSocketChannel 来接收一个客户端发送来的socket
               socket = endpoint.serverSocketAccept();
               ........................省略代码................................................
              // setSocketOptions() will hand the socket off to
             // an appropriate processor if successful
             if (!endpoint.setSocketOptions(socket)) {
                    //及时关闭Socket连接
                   endpoint.closeSocket(socket);  
              }
        }

从Tomcat8 以后，Tomcat的默认连接为NIO。所以在这里的EndPoint的具体实现是NioEndPoint

@Override
        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 {
                        // NIOChannel 实质上对ByteChannel 的一个封装实现
                        channel = new NioChannel(socket, bufhandler);
                    }
                } else {
                    // 根据Socket，对设置当前的NioChannel
                    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;
        }

通过阅读代码可以看出其处理过程如下：

1.  设置非阻塞，以及其他的一些参数如SoTimeout、ReceiveBufferSize、SendBufferSize
2.  然后将SocketChannel封装成一个NioChannel，封装过程使用了缓存，即避免了重复创建NioChannel，封装过程中使用了缓存，既避免了重复创建NioChannel对象，直接利用原有的NIOChannel，并将NioChannel中的数据全部清空。
3.  选择一个Poller进行注册

再来看一下Poller

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); //将event 加入到事件处理队列中
            }

这里又是进行一些参数包装，将socket和Poller的关系绑定，再次从缓存中取出或者重新构建一个PollerEvent，然后将该event放到Poller的事件队列中等待被异步处理

再来看看对于被加入到队列中的events的处理

public boolean events() {
                boolean result = false;
    
                PollerEvent pe = null;
                while ( (pe = events.poll()) != null ) {
                    result = true;
                    try {
                        pe.run(); //开始处理这个pollerEvent
                        pe.reset(); //
                        if (running && !paused) {
                            eventCache.push(pe);
                        }
                    } catch ( Throwable x ) {
                        log.error("",x);
                    }
                }
    
                return result;
            }

Poller的run 方法会对

@Override
            public void run() {
    .........................省略代码................................
    
                    // 遍历并处理 已经准备好的NIO 事件
                    while (iterator != null && iterator.hasNext()) {
                        SelectionKey sk = iterator.next();
                        NioSocketWrapper attachment = (NioSocketWrapper)sk.attachment();
    
                        if (attachment == null) {
                            iterator.remove();
                        } else {
                            iterator.remove();
                            //将已经准备好的IO事件和 绑定的Socket进行分类处理
                            processKey(sk, attachment);
                        }
                    }//while
      ............................省略代码.............
    }

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 {
                                //为了避免多线程对于attach的socketChannel处理的冲突，在这里将socketChannel的
                                // ready 操作位取反
                                unreg(sk, attachment, sk.readyOps());
                                boolean closeSocket = false;
                                // 如果SelectorKey 为Read，那就去处理Read
                                if (sk.isReadable()) {
                                    if (!processSocket(attachment, SocketEvent.OPEN_READ, true)) {
                                        closeSocket = true;
                                    }
                                }
                                // 如果SelectorKey 为Read，那就去处理Write
                                if (!closeSocket && sk.isWritable()) {
                                    if (!processSocket(attachment, SocketEvent.OPEN_WRITE, true)) {
                                        closeSocket = true;
                                    }
                                }
                                if (closeSocket) {
                                    // 将attach的socketChannel从key中解绑
                                    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) {
                  //将SocketProcessorBase交给线程池中的线程来处理
                    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;
        }

线程池中的线程即为 `NioEndPonint$SocketProcessor` 内部类来看一下其内部的`doRun()`方法

doRun{
      .....................
        //如果握手已经完成....
        if (handshake == 0) {
                        SocketState state = SocketState.OPEN;
                        // Process the request from this socket
                        if (event == null) {
                            state = getHandler().process(socketWrapper, SocketEvent.OPEN_READ);
                        } else {
                            // 这里是关键
                            state = getHandler().process(socketWrapper, event);
                        }
                        if (state == SocketState.CLOSED) {
                            close(socket, key);
                        }
                    }
        ....................
    }

从上面代码中可以看出，注意它调用了hanler.process(socket)来生成响应数据。并且根据处理之后的返回状态来决定是否关闭Socket连接

对于handler.process(socket)的处理。

NioEndpoint类中的Handler接口的具体实现是静态类 `AbstractProtocol$ConnectionHandler<S>`通过查看其`process(socket)`方法.这个方法逻辑很长并且很复杂。说多了也没用。  
它在这个方法里做的动作

1.  通过协议类型得到相应的Socket的 Processor,并cache起来,在这里这个**Processor就是`Http11Processor`**
2.  通过一个ThreadLocal 类标识现在的这个线程正在处理一个请求
3.  调用Processor的process方法。

上面的Processor的process方法通过抽象类间接的调用了`Http11Processor`的`service()`方法。这个`service()`方法也是相当复杂\`  
它主要完成的动作有

*  填充Request，Reponse属性
 *  调用`CoyoteAdapter`的service()方法

通过以上不走，一个请求连接就从Connector走到了Container

# 小结： #

实现一个tomcat连接器Connector就是实现ProtocolHander接口的过程。Connector用来接收Socket Client端的请求，通过内置的线程池去调用Servlet Container生成响应结果，并将响应结果同步或异步的返回给Socket Client。在第三方应用集成tomcat作为Web容器时，一般不会动Servlet Container端的代码，那么connector的性能将是整个Web容器性能的关键。

![Image 1][]

Tomcat 对于Socket的处理.png

作者：walker\_liu\_fei  
链接：https://www.jianshu.com/p/f007b6427d60  
來源：简书  
简书著作权归作者所有，任何形式的转载都请联系作者获得授权并注明出处。

[Image 1]: