zoukankan      html  css  js  c++  java
  • Tomcat NIO

    说起Tomcat的NIO,不得不提的就是Connector这个Tomcat组件。Connector是Tomcat的连接器,其主要任务是负责处理收到的请求,并创建一个Request和Response的对象,然后用一个线程用于处理请求,Connector会把Request和Response对象传递给该线程,该线程的具体的处理过程是Container容器的事了。

    在tomcat启动过程中,会初始化Connector,并调用Connector的startInternal()方法开启Connector,开始监听、处理请求。

    想了解Tomcat NIO的工作方式,就得先了解一下Connector的实现原理。下面从三个方面来了解一下Connector组件:Connector的数据结构、Connector初始化以及Connector开启。

    Connector

    Connector的数据结构

    先了解一下Connector的数据结构。Connector的一个主要的属性:ProtocolHandler protocolHandler(协议)

    protocolHandler(协议)

    • 维护服务器使用的协议,如http1.1等。ProtocolHandler是接口,实现类有Http11Nio2Protocol 、Http11Nio2Protocol等
    • 维护服务提供的IO方式,负责EndPoint的初始化、启动。目前有BIO、NIO、AIO等IO方式,来实现监听端口、读写socket数据的功能。通过EndPoint封装实现不同的IO方式
    • EndPoint监听到IO读写,交给Tomcat线程池中的一个线程来处理,SocketProcessor会根据protocolHandler采用的协议,调用协议的process方法处理请求。
    • 维护adapter(适配器),可以将请求/响应数据进行适配

    protocolHandler会找到socket对应的处理器(如Http11Processor),然后进行数据读写、适配,处理。请求由adapter最终会交给servlet处理

    常说的BIO、NIO,主要的应用就在protocolHandler中。protocolHandler负责维护Connector使用的协议以及IO方式。在protocolHandler中,不同的IO方式,会使用不同的EndPoint,具体采用哪种IO方式,取决于采用哪个EndPoint,每一个EndPoint的实现类,都封装了一种IO策略。若采用NIO,则为NioEndpoint。

    Connector初始化

    创建Connector时,会拿到Tomcat目录下conf/server.xml中Connector的协议配置,利用反射创建ProtocolHandler:

    /**
     * Coyote Protocol handler class name.
     * Defaults to the Coyote HTTP/1.1 protocolHandler.
     */
    protected String protocolHandlerClassName = "org.apache.coyote.http11.Http11NioProtocol";
    
    
    public Connector(String protocol) {
    	//设置protocolHandlerClassName类名
        setProtocol(protocol);
        // Instantiate protocol handler
        ProtocolHandler p = null;
        try {
    		//根据server.xml中<connector/>标签的protocol属性值,获取到对应的http协议类
            Class<?> clazz = Class.forName(protocolHandlerClassName);
            p = (ProtocolHandler) clazz.getConstructor().newInstance();
        } catch (Exception e) {
            log.error(sm.getString(
                    "coyoteConnector.protocolHandlerInstantiationFailed"), e);
        } finally {
            this.protocolHandler = p;
        }
    
        if (Globals.STRICT_SERVLET_COMPLIANCE) {
            uriCharset = StandardCharsets.ISO_8859_1;
        } else {
            uriCharset = StandardCharsets.UTF_8;
        }
    }
    
    //设置protocolHandlerClassName类名
    public void setProtocol(String protocol) {
    
        boolean aprConnector = AprLifecycleListener.isAprAvailable() &&
                AprLifecycleListener.getUseAprConnector();
    
    	//若配置了protocol="HTTP/1.1"或者没配,则默认是Http11NioProtocol或者Http11AprProtocol
        if ("HTTP/1.1".equals(protocol) || protocol == null) {
            if (aprConnector) {
                setProtocolHandlerClassName("org.apache.coyote.http11.Http11AprProtocol");
            } else {
                setProtocolHandlerClassName("org.apache.coyote.http11.Http11NioProtocol");
            }
        } else if ("AJP/1.3".equals(protocol)) {
            if (aprConnector) {
                setProtocolHandlerClassName("org.apache.coyote.ajp.AjpAprProtocol");
            } else {
                setProtocolHandlerClassName("org.apache.coyote.ajp.AjpNioProtocol");
            }
        } else {
    		//直接取配置的类名
            setProtocolHandlerClassName(protocol);
        }
    }
    

     

    以Tomcat8.5.20为例,这里默认是http1.1的NIO。

    Connector.start()开启

    Connector初始化后,调用start方法开启。主要涉及一下几个方法:

    Connector的startInternal()方法,会调用protocolHandler.start();

    protocolHandler中会调用endpoint.start(),从而达到开启endpoint、监听端口、读写Socket的目的:

    以Tomcat8.5.20为例,这里默认是http1.1的NIO。
    
    
    Connector.start()开启
    
    Connector初始化后,调用start方法开启。主要涉及一下几个方法:
    
    Connector的startInternal()方法,会调用protocolHandler.start();
    protocolHandler中会调用endpoint.start(),从而达到开启endpoint、监听端口、读写Socket的目的:
    

      

    至此,Connector完成了开启的过程,开启监听端口、可以读写Socket了。

    总结一下,关于Connector:

    创建Connector时,会拿到Tomcat目录下conf/server.xml中Connector的协议配置,利用反射创建ProtocolHandler。

    ProtocolHandler负责维护Connector使用的协议以及IO方式,不同的IO方式如BIO、NIO、AIO封装在EndPoint中

    开启Connector时,会开启protocolHandler,从而达到EndPoint的开启,开始监听端口、读写socket数据了

    protocolHandler中将请求拿到的数据进行适配,通过adapter适配成Request和Response对象,最终交给Container去处理

    下面重点就来了,NIO。

    Tomcat NIO

    Tomcat在处理客户端请求时,读写socket数据是一种网络IO操作。目前Tomcat有几种IO方式,分别是BIO(同步阻塞),NIO(同步非阻塞)和AIO(异步非阻塞)。不同IO方式的读写机制,被封装在了Endpoint中。BIO、AIO不再赘述。这里主要看NIO。

    Tomcat NIO模型

    当然要了解一下Tomcat NIO的模型了。Tomcat NIO是基于Java NIO实现的,其基本原理如下:

    Tomcat NIO是对Java NIO的一种典型的应用方式:通过JDK提供的同步非阻塞的IO方式,实现了IO多路复用,即一个线程管理多个客户端的连接。了解Java NIO,可以看一下Java NIO

    Tomcat在NIO模式下,所有客户端的请求先由一个接收线程接收,然后由若干个(一般为CPU的个数)线程轮询读写事件,最后将具体的读写操作交由线程池处理。

    NioEndpoint

    要了解Tomcat的NIO实现,其实就是了解NioEndpoint的实现原理。

    数据结构

    它一共包含LimitLatch、Acceptor、Poller、SocketProcessor、Excutor5个部分

    • LimitLatch是连接控制器,它负责维护连接数的计算,nio模式下默认是10000,达到这个阈值后,就会拒绝连接请求。
    • Acceptor负责接收连接,默认是1个线程来执行,将请求的事件注册到事件列表
    • Poller来负责轮询上述产生的事件。Poller线程数量是cpu的核数Math.min(2,Runtime.getRuntime().availableProcessors())。由Poller将就绪的事件生成SocketProcessor,然后交给Excutor去执行。
    • SocketProcessor继承了SocketProcessorBase,实现了Runnable接口,可以提交给线程池Excutor来执行。它里面的doRun()方法,封装了读写Socket、完成Container调用的逻辑
    • Excutor线程池是一个Tomcat线程池。用来执行Poller创建的SocketProcessor。Excutor线程池的大小就是我们在Connector节点配置的maxThreads的值。

    SocketProcessor被一个线程执行的时候,会完成从socket中读取http request,解析成HttpServletRequest对象,分派到相应的servlet并完成逻辑,然后将response通过socket发回client。在从socket中读数据和往socket中写数据的过程,并没有像典型的非阻塞的NIO的那样,注册OP_READ或OP_WRITE事件到主Selector,而是直接通过socket完成读写,这时是阻塞完成的,但是在timeout控制上,使用了NIO的Selector机制,但是这个Selector并不是Poller线程维护的主Selector,而是BlockPoller线程中维护的Selector,称之为辅Selector,实现可见org.apache.coyote.http11.Http11InputBuffer#fill。

    了解了NioEndPoint的数据结构之后,可以看一下它们的关系图

    NioEndpoint组件关系图

    以上过程就以同步非阻塞的方式完成了网络IO。

    其实是一个Reactor模型:

    • 一个Acceptor(当然多个也行,不过一般场景一个够了)负责accept事件,把接收到SocketChannel注册到按某种算法从Reactor池中取出的一个Reactor上,注册的事件为读,写等,之后这个Socket Channel的所有IO事件都和Acceptor没关系,都由被注册到的那个Reactor来负责。
    • 每个Acceptor和每个Reactor都各自持有一个Selector
    • 当然每个Acceptor和Reactor都是一个线程

    这里的Poller池其实就是一个Reactor池,可以是多个线程。

    NioEndPoint实现

    工作原理简单了解了一下,接下来看一下具体的代码实现吧。先上一个NioEndpoint的UML图:

    NioEndPoint启动

    AbstractEndpoint里实现了一些EndPoint的抽象的通用的方法,其中主要的一个入口方法是org.apache.tomcat.util.net.AbstractEndpoint#start方法

     
    NioEndPoint启动
    
    AbstractEndpoint里实现了一些EndPoint的抽象的通用的方法,其中主要的一个入口方法是org.apache.tomcat.util.net.AbstractEndpoint#start方法
    

      

    其中,bind()方法和startInternal()方法,由其子类具体实现。

    bind()方法用于初始化endpoint,绑定监听端口等、设置最大线程数、ssl等。

    startInternal()方法在EndPoint初始化完毕后,创建pollers轮询线程以及acceptors线程并开启。

    其中,bind()方法和startInternal()方法,由其子类具体实现。
    bind()方法用于初始化endpoint,绑定监听端口等、设置最大线程数、ssl等。
    startInternal()方法在EndPoint初始化完毕后,创建pollers轮询线程以及acceptors线程并开启。
    

      

    NioEndPoint时序图

     

    看完了开启EndPoint的过程,再来详细看一下NioEndpoint处理的的时序图:

    通过上面的时序图,结合代码来详细了解一下Acceptor和Poller的工作方式。

    Acceptor接收请求

    NioEndPoint中的Acceptor方法实现了Runnable接口,主要干的活就是上述图中的3,4,5,6,7

    @Override  
    public void run() {  
      
        int errorDelay = 0;  
      
        // 循环,直到收到一个关闭的命令  
        while (running) {  
      
            // 如果EndPoint被暂停,则循环sleep  
            while (paused && running) {  
                state = AcceptorState.PAUSED;  
                try {  
                    Thread.sleep(50);  
                } catch (InterruptedException e) {  
                    // Ignore  
                }  
            }  
      
            if (!running) {  
                break;  
            }  
            state = AcceptorState.RUNNING;  
      
            try {  
                //如果达到了最大连接数,则等待  
                countUpOrAwaitConnection();  
      
                SocketChannel socket = null;  
                try {  
                    // 创建一个socketChannel,接收下一个从服务器进来的连接  
                    socket = serverSock.accept();  
                } catch (IOException ioe) {  
                    // We didn't get a socket  
                    countDownConnection();  
                    if (running) {  
                        // Introduce delay if necessary  
                        errorDelay = handleExceptionWithDelay(errorDelay);  
                        // re-throw  
                        throw ioe;  
                    } else {  
                        break;  
                    }  
                }  
                // 成功接收,重置error delay  
                errorDelay = 0;  
      
                // 如果处于EndPoint处于running状态并且没有没暂停,Configure the socket  
                if (running && !paused) {  
                    // setSocketOptions()将把socket传递给适当的处理器。如果成功,会关闭socket。  
                    // 否则,在这里关闭socket  
                    if (!setSocketOptions(socket)) {  
                        closeSocket(socket);  
                    }  
                } else {  
                    closeSocket(socket);  
                }  
            } catch (Throwable t) {  
                ExceptionUtils.handleThrowable(t);  
                log.error(sm.getString("endpoint.accept.fail"), t);  
            }  
        }  
        state = AcceptorState.ENDED;  
    }  
    

      

    看的出来,Acceptor使用serverSock.accept()阻塞的监听端口,如果有连接进来,拿到了socket,并且EndPoint处于正常运行状态,则调用NioEndPoint的setSocketOptions方法,一顿操作。

    至于setSocketOptions做了什么,概括来说就是根据socket构建一个NioChannel,然后把这个的NioChannel注册到Poller的事件列表里面,等待poller轮询。

    看下setSocketOptions的代码:

     
    /**
     * 处理指定的连接
     * @param socket The socket channel
     * @return  
     *  如果socket配置正确,并且可能会继续处理,返回true 
     *  如果socket需要立即关闭,则返回false
     */
    protected boolean setSocketOptions(SocketChannel socket) {
        // Process the connection
        try {
            //非阻塞模式
            socket.configureBlocking(false);
            Socket sock = socket.socket();
            socketProperties.setProperties(sock);
    		
    		//从缓存中拿一个nioChannel  若没有,则创建一个。将socket传进去
            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();
            }
    		//从pollers数组中获取一个Poller对象,注册这个nioChannel
            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;
    }
    

      

    显然,下面的重点就是register这个方法了。这个方法是NioEndPoint中的Poller实现的,主要干的事就是在Poller注册新创建的套接字。

    /**
     * 使用轮询器注册新创建的socket
     *
     * @param socket    新创建的socket
     */
    public void register(final NioChannel socket) {
        socket.setPoller(this);
    	//创建一个NioSocketWrapper,包装一下socket。然后一顿设置。
        NioSocketWrapper ka = new NioSocketWrapper(socket, NioEndpoint.this);
        socket.setSocketWrapper(ka);
        ka.setPoller(this);
        ka.setReadTimeout(getSocketProperties().getSoTimeout());
        ka.setWriteTimeout(getSocketProperties().getSoTimeout());
        ka.setKeepAliveLeft(NioEndpoint.this.getMaxKeepAliveRequests());
        ka.setSecure(isSSLEnabled());
        ka.setReadTimeout(getConnectionTimeout());
        ka.setWriteTimeout(getConnectionTimeout());
    
    
    	//从缓存中取出一个PollerEvent对象,若没有则创建一个。将socket和NioSocketWrapper设置进去
        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);
    
    
    	//添到到该Poller的事件列表
        addEvent(r);
    }
    

      

    总结一下,从Acceptor接收到请求,它做了这么些工作:

    • 如果达到了最大连接数,则等待。否则,阻塞监听端口。
    • 监听到有连接,则创建一个socketChannel。若服务正常运行,则把socket传递给适当的处理器。如果成功,会关闭socket。

    在这里,适当的处理是指调用NioEndPoint的setSocketOptions方法,处理指定的连接:

    • 将socket设置为非阻塞
    • 从缓存中拿一个nioChannel  若没有,则创建一个。将socket传进去。
    • 从pollers数组中获取一个Poller对象,把nioChannel注册到该Poller中。

    其中最后一步注册的过程,是调用Poller的register()方法:

    • 创建一个NioSocketWrapper,包装socket。然后配置相关属性,设置感兴趣的操作为SelectionKey.OP_READ
    • PollerEvent。PollerEvent可以是从缓存中取出来的,若没有则创建一个。初始化或者重置此Event对象,设置感兴趣的操作为OP_REGISTER (Poller轮询时会用到)
    • 将新的PollerEvent添加到这个Poller的事件列表events,等待Poller线程轮询。

    Poller轮询

    其实上面已经提到了Poller将一个事件注册到事件队列的过程。接下来便是Poller线程如何处理这些事件了,这就是Poller线程的工作机制。

    Poller作为一个线程,实现了Runnable接口的run方法,在run方法中会轮询事件队列events,将每个PollerEvent中的SocketChannel感兴趣的事件注册到Selector中,然后将PollerEvent从队列里移除。之后就是SocketChanel通过Selector调度来进行非阻塞的读写数据了。

    看下Poller.run()代码:

    /**
     * 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() {
        // 循环直到 destroy() 被调用
        while (true) {
    
            boolean hasEvents = false;
    
            try {
                if (!close) {
    				//将events队列,将每个事件中的通道感兴趣的事件注册到Selector中
                    hasEvents = events();
                    if (wakeupCounter.getAndSet(-1) > 0) {
                        //如果走到了这里,代表已经有就绪的IO通道
                        //调用非阻塞的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;
            }
            //如果上面select方法超时,或者被唤醒,先将events队列中的通道注册到Selector上。
            if ( keyCount == 0 ) hasEvents = (hasEvents | events());
    
            Iterator<SelectionKey> iterator =
                keyCount > 0 ? selector.selectedKeys().iterator() : null;
            // 遍历已就绪的通道,并调用processKey来处理该Socket的IO。
            while (iterator != null && iterator.hasNext()) {
                SelectionKey sk = iterator.next();
                NioSocketWrapper attachment = (NioSocketWrapper)sk.attachment();
                // 如果其它线程已调用,则Attachment可能为空
                if (attachment == null) {
                    iterator.remove();
                } else {
                    iterator.remove();
    				//创建一个SocketProcessor,放入Tomcat线程池去执行
                    processKey(sk, attachment);
                }
            }//while
    
            //process timeouts
            timeout(keyCount,hasEvents);
        }//while
    
        getStopLatch().countDown();
    }
    

      

    读取已就绪通道的部分,是常见的Java NIO的用法,Selector调用selectedKeys(),获取IO数据已经就绪的通道,遍历并调用processKey方法来处理每一个通道就绪的事件。而processKey方法会创建一个SocketProcessor,然后丢到Tomcat线程池中去执行。

    其中需要注意的一个点是,events()方法,用来处理PollerEvent事件,执行PollerEvent.run(),然后将PollerEvent重置再次放入缓存中,以便对象复用。

    /**
     * Processes events in the event queue of the Poller.
     *
     * @return <code>true</code> if some events were processed,
     *   <code>false</code> if queue was empty
     */
    public boolean events() {
        boolean result = false;
    
        PollerEvent pe = null;
        while ( (pe = events.poll()) != null ) {
            result = true;
            try {
    			//把SocketChannel感兴趣的事件注册到Selector中
                pe.run();
                pe.reset();
                if (running && !paused) {
                    eventCache.push(pe);
                }
            } catch ( Throwable x ) {
                log.error("",x);
            }
        }
    
        return result;
    }
    

    可以看出,PollerEvent.run()方法才是重点:

    public void run() {
    	//Acceptor调用Poller.register()方法时,创建的PollerEvent感兴趣的事件为OP_REGISTER,因此走这个分支
        if (interestOps == OP_REGISTER) {
            try {
    			//将SocketChannel的读事件注册到Poller线程的Selector中,使用Selector来调度IO。
                socket.getIOChannel().register(
                        socket.getPoller().getSelector(), SelectionKey.OP_READ, socketWrapper);
            } catch (Exception x) {
                log.error(sm.getString("endpoint.nio.registerFail"), x);
            }
        } else {
            final SelectionKey key = socket.getIOChannel().keyFor(socket.getPoller().getSelector());
            try {
                if (key == null) {
                    // The key was cancelled (e.g. due to socket closure)
                    // and removed from the selector while it was being
                    // processed. Count down the connections at this point
                    // since it won't have been counted down when the socket
                    // closed.
                    socket.socketWrapper.getEndpoint().countDownConnection();
                } else {
                    final NioSocketWrapper socketWrapper = (NioSocketWrapper) key.attachment();
                    if (socketWrapper != null) {
                        //we are registering the key to start with, reset the fairness counter.
                        int ops = key.interestOps() | interestOps;
                        socketWrapper.interestOps(ops);
                        key.interestOps(ops);
                    } else {
                        socket.getPoller().cancelledKey(key);
                    }
                }
            } catch (CancelledKeyException ckx) {
                try {
                    socket.getPoller().cancelledKey(key);
                } catch (Exception ignore) {}
            }
        }
    }
    

      

    至此,可以看出Poller线程的作用

    • 将Acceptor接收到的请求注册到Poller的事件队列中
    • Poller轮询事件队列中,处理到达的事件,将PollerEvent中的通道注册到Poller的Selector中
    • 轮询已就绪的通道,对每个就绪通道创建一个SocketProcessor,交个Tomcat线程池去处理

    剩下的事情,就是SocketProcessor怎么适配客户端发来请求的数据、然后怎样交给Tomcat容器去处理了。

    SocketProcessor处理请求

    简单提一下SocketProcessor的处理过程,不是这篇文章的重点。通过上面可以知道,具体处理一个请求,是在SocketProcessor通过线程池去执行的。执行一次请求的时序图

    SocketProcessor中通过Http11ConnectionHandler,取到Htpp11Processor,Htpp11Processor调用prepareRequest方法,准备好请求数据。然后调用CoyoteAdapter的service方法进行request和response的适配,之后交给容器进行处理。

    在CoyoteAdapter的service方法中,主要干了2件事:

    • org.apache.coyote.Request -> org.apache.catalina.connector.Request extends HttpServletRequest,org.apache.coyote.Response -> org.apache.catalina.connector. Response extends HttpServletResponse
    • 将请求交给StandardEngineValue处理

    将请求交给Tomcat容器处理后,后将请求一层一层传递到Engin、Host、Context、Wrapper,最终经过一系列Filter,来到了Servlet,执行我们自己具体的代码逻辑。其中,容器之间数据的传递用到了管道流的机制。这里就不在赘述,以后有时间专门写一篇Tomcat容器的工作原理。

    参考文章:

    《Tomcat内核设计剖析》

    深度解读Tomcat中的NIO模型

  • 相关阅读:
    Oracle备份 还原命令
    maven错误解决一:Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:2.5.1:compile (default-compile)
    maven File encoding has not been set
    maven 错误: 程序包org.junit不存在
    <转>JDBC获取DB元数据
    <转>SQL语句执行顺序说明
    Oracle 创建/删除 表空间、用户、授权
    lucene/solr 修改评分规则方法总结
    Solr入门之(8)中文分词器配置
    Solr入门之(7)Solr客户端界面简介
  • 原文地址:https://www.cnblogs.com/z941030/p/8796479.html
Copyright © 2011-2022 走看看