文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

springboot内置tomcat之NIO处理流程一览

2024-04-02 19:55

关注

前言

springboot内置的tomcat目前默认是基于NIO来实现的,本文介绍下tomcat接受请求的一些组件及组件之间的关联

tomcat组件

本文只介绍NIO中tomcat的组件

我们直接看NIO的核心类NioEndpoint的startInternal方法

Acceptor组件


public void startInternal() throws Exception {
        if (!running) {
            running = true;
            paused = false;
            processorCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
                    socketProperties.getProcessorCache());
            eventCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
                            socketProperties.getEventCache());
            nioChannels = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
                    socketProperties.getBufferPool());
            // Create worker collection
            if ( getExecutor() == null ) {
                createExecutor();
            }
            initializeConnectionLatch();
            // Start poller threads
            // 核心代码1
            pollers = new Poller[getPollerThreadCount()];
            for (int i=0; i<pollers.length; i++) {
                pollers[i] = new Poller();
                Thread pollerThread = new Thread(pollers[i], getName() + "-ClientPoller-"+i);
                pollerThread.setPriority(threadPriority);
                pollerThread.setDaemon(true);
                pollerThread.start();
            }
			// 核心代码2
            startAcceptorThreads();
        }
    }

看核心代码1的位置构造了一个Poller数组,Poller是一个实现了Runnable的类,并且启动了该线程类,

getPollerThreadCount()方法返回了2和当前物理机CPU内核数的最小值,即创建的数组最大值为2

接下来看核心代码2startAcceptorThreads()


protected final void startAcceptorThreads() {
    int count = getAcceptorThreadCount();
    acceptors = new ArrayList<>(count);
    for (int i = 0; i < count; i++) {
        Acceptor<U> acceptor = new Acceptor<>(this);
        String threadName = getName() + "-Acceptor-" + i;
        acceptor.setThreadName(threadName);
        acceptors.add(acceptor);
        Thread t = new Thread(acceptor, threadName);
        t.setPriority(getAcceptorThreadPriority());
        t.setDaemon(getDaemon());
        t.start();
    }
}

创建了多个Acceptor类,Acceptor也是实现了Runnable的线程类,创建个数默认是1

然后我们看下Acceptor启动后做了什么,我们直接看run方法


  public void run() {
        ......
                U socket = null;
                try {
                    // Accept the next incoming connection from the server
                    // socket
                    // 核心代码1
                    socket = endpoint.serverSocketAccept();
                } catch (Exception ioe) {
                    ......
                }
                // 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
                    // 核心代码2
                    if (!endpoint.setSocketOptions(socket)) {
                        endpoint.closeSocket(socket);
                    }
               ......
    }

核心代码1很明显是一个阻塞模型,即接受客户端连接的,当没有客户端连接时会处于阻塞,这里可以看到默认情况下tomcat在nio模式下只有一个Acceptor线程类来接受连接

然后看核心代码2


 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) {
           ......
        }
        return true;
    }

我们看核心代码getPoller0().register(channel)


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);
    }

看addEvent


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

events的定义


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

这里可以看到封装了一个PollerEvent 并且扔到了一个队列里面,然后当前类就结束了

由此可得Acceptor的作用就是接受客户端连接,并且把连接封装起来扔到了一个队列中

Poller

我们前面已经创建并且启动了多个Poller线程类,默认的数量是小于等于2的。

然后我们看下Poller类做了什么,同样我们看run方法


  @Override
    public void run() {
        // Loop until destroy() is called
        while (true) {
            boolean hasEvents = false;
            try {
                if (!close) {
                	// 核心代码1
                    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();
                    // 核心代码2
                    processKey(sk, attachment);
                }
            }//while
            //process timeouts
            timeout(keyCount,hasEvents);
        }//while
        getStopLatch().countDown();
    }

先看核心代码1 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;
            try {
            	// 核心代码
                pe.run();
                pe.reset();
                if (running && !paused) {
                    eventCache.push(pe);
                }
            } catch ( Throwable x ) {
                log.error("",x);
            }
        }
        return result;
    }

核心代码run


@Override
    public void run() {
        if (interestOps == OP_REGISTER) {
            try {
            	// 核心代码,注册到selector轮训器
                socket.getIOChannel().register(
                        socket.getPoller().getSelector(), SelectionKey.OP_READ, socketWrapper);
            } catch (Exception x) {
                log.error(sm.getString("endpoint.nio.registerFail"), x);
            }
        } else {
            ......
        }
    }

可以看出大概的意思就是从刚才我们放进去的队列events里面取数据放到了eventCache里面,eventCache的定义SynchronizedStack eventCache,当取到数据后返回true,这个时候就会进入核心代码2处的processKey(sk, attachment),也就是开始处理请求了


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;
                            }
                        }
                      ......
    }

这里就可以看到我们的NIO的模型了,也就是多路复用的模型,轮询来判断key的状态,当key是可读或者可写时执行processSocket


  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;
    }

这里就是核心代码了,可以看到getExecutor()方法,获取线程池,这个线程池是在初始化tomcat时提前初始化好的,默认情况下核心线程是10,最大线程是200。线程池的配置可以根据我们自己配置来设置大小。

这里拿到线程池然后包装了一个SocketProcessorBase线程类扔到线程池里面取执行

从这里可以看到Poller的功能就是从前面的队列里面获取连接然后包装成SocketProcessorBase之后扔到线程池里面去执行,SocketProcessorBase才是最终真正处理请求的

总结

根据上面的分析我们已经可以看到tomcat的执行流程了,这里盗用网上的一张比较好的图

在这里插入图片描述

大致流程为

1、创建一个Acceptor线程来接收用户连接,接收到之后扔到events queue队列里面,默认情况下只有一个线程来接收

2、创建Poller线程,数量小于等于2,Poller对象是NIO的核心,在Poller中,维护了一个Selector对象;当Poller从队列中取出socket后,注册到该Selector中;然后通过遍历Selector,找出其中可读的socket,然后扔到线程池中处理相应请求,这就是典型的NIO多路复用模型

3、扔到线程池中的SocketProcessorBase处理请求

相较于BIO模型的tomcat,NIO的优势分析

1、BIO中的流程应该是接收到请求之后直接把请求扔给线程池去做处理,在这个情况下一个连接即需要一个线程来处理,线程既需要读取数据还需要处理请求,线程占用时间长,很容易达到最大线程

2、NIO的流程的不同点在于Poller类采用了多路复用模型,即Poller类只有检查到可读或者可写的连接时才把当前连接扔给线程池来处理,这样的好处是大大节省了连接还不能读写时的处理时间(如读取请求数据),也就是说NIO“读取socket并交给Worker中的线程”这个过程是非阻塞的,当socket在等待下一个请求或等待释放时,并不会占用工作线程,因此Tomcat可以同时处理的socket数目远大于最大线程数,并发性能大大提高。

以上就是我对于tomcat中nio处理模型的一些理解。希望能给大家一个参考,也希望大家多多支持编程网

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯