Tomcat是如何处理请求的(上)

Tomcat是如何处理请求的(上)

技术杂谈小彩虹2021-07-10 19:30:2760A+A-

概述

Tomcat涉及的内容较多,本文主要分析其关键部分的高并发设计。

主要内容

  1. IO模型概述
  2. Tomcat的IO模型实现
  3. Tomcat的并发控制

IO模型概述

从操作系统层面看服务端的数据交互流程:

如图所示,当服务端收到请求后,用户空间的用户线程会发起read调用,这个read调用依赖两个过程:从网卡拷贝数据到内核空间、从内核空间拷贝数据到用户空间,write调用是反向的两个步骤。

那在处理这两个耗时的Copy操作时,用户线程是占着CPU还是出让CPU?怎样可以更高效的处理数据?不同IO模型的作用就在这里体现出来了。

同步阻塞IO

用户线程发起read调用后就进入阻塞(出让CPU),等待read执行完(数据拷贝到用户空间)后再唤醒。

同步非阻塞IO

用户线程发起read调用后不进入阻塞,而是持续不断的发起read轮询,在数据拷贝到内核空间之前read调用都返回false,数据拷贝到内核空间后,read开始阻塞,等待数据从内核空间拷贝的用户空间后再唤醒用户线程。

IO多路复用

用户线程读取数据分成两步,先轮询调用select询问数据是否已到内核,如果到了则调用read读取数据,此时read调用会等待数据从内核空间拷贝到用户空间,这个过程是阻塞的。而这里多路复用的意思是:一次select可以得到多个channel的结果。

异步IO

用户线程在调用read时注册一个回调函数,当数据到用户空间后会调用此回调通知用户线程,这个过程用户线程不阻塞。

本文介绍的Tomcat9默认是基于IO多路复用模型做的的高并发设计。

Tomcat的IO模型实现

假设我们把Tomcat当做黑盒子,按照Spring的方式来处理请求,简化的流程是这样的:

Tomcat负责读取内核的数据,转换成Servlet对象,然后由Spring框架处理业务后通过Response对象写入返回数据,Tomcat再将返回数据通过内核写入网卡,最后返回到客户端。

接下来我们将Tomcat这部分放大,看看黑盒子里是怎么处理的。

如图所示,请求的处理分如下几个步骤:

  1. Tomcat在启动时会初始化一个ServerSocket用于监听指定端口的IO请求(比如8080)
  2. 接着启动Acceptor线程,循环调用accept方法接收IO请求(TCP连接建立)
  3. 将ServerChannel包装成PollerEvent,注册到Poller的event队列中
  4. Poller线程循环遍历event队列,将Poller关注的ServerChannel的READ操作注册到Selector中
  5. 在同一个Poller循环中,用Selector查询ServerChannel的状态,这里一次可以查询到多个ServerChannel的状态,即多路复用
  6. 将查询到的SelectionKey对应的ServerChannel挨个创建SocketProcessor,SocketProcessor是Runnable的
  7. 将SocketProcessor扔到工作线程进行处理
  8. 接下来的协议解析、Request和Response适配、Servlet处理、业务处理等都在SocketProcessor的流程中

涉及的主要代码:

  • Acceptor线程:Acceptor.run->NioEndpoint.setSocketOptions->NioEndpoint.register->Poller.addEvent
  • Poller线程:Poller.run->Poller.events->PollerEvent.run->Poller.processKey->AbstractEndpoint.processSocket

下一节我们再对部分源码进行分析。

Tomcat的并发控制

上面分析了Tomcat的IO多路复用模型实现,在这个实现中每一个环节都有影响并发的关键控制,先看图:

内核-Accept List

TCP三次握手建立连接的过程中,内核会为每一个LISTEN状态的Socket维护两个队列:

  • SYN队列:这些连接已收到客户端的SYN
  • ACCEPT队列:这些连接已经收到客户端的ACK,完成了三次握手,等待被系统accept调用取走

Tomcat的Acceptor负责从ACCEPT队列中取走连接,当Acceptor处理不过来时,连接就堆积在ACCEPT队列中,这个队列长度由acceptCount(默认100)控制,当我们尝试把acceptCount设置的很小,发起并发请求时,会收到一个Socket Error。

// 初始化服务端端口监听-NioEndpoint.initServerSocket
protected void initServerSocket() throws Exception {
    if (!getUseInheritedChannel()) {
        serverSock = ServerSocketChannel.open();
        socketProperties.setProperties(serverSock.socket());
        InetSocketAddress addr = new InetSocketAddress(getAddress(), getPortWithOffset());
        // 绑定acceptCount,可以通过配置server.tomcat.accept-count调整大小
        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
}

Acceptor-LimitLatch

并发处理请求的最大连接控制器,内部通过AQS实现等待。即如果服务端请求处理不过来,并发的请求数量超过了maxConnections(默认8192),则会进入等待。当Acceptor线程接收出现异常,Socket被异常关闭或者SocketProessor处理完后会回收连接数。

// Acceptor线程执行逻辑-Acceptor.run
public void run() {
        // Loop until we receive a shutdown command
        while (endpoint.isRunning()) {
            state = AcceptorState.RUNNING;

            try {
                //if we have reached max connections, wait
                // LimitLatch控制最大并发连接数,如果达到最大连接数则进入等待,直到有空余连接数
                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
                    // 调用accept方法从内核提取IO请求
                    socket = endpoint.serverSocketAccept();
                } catch (Exception ioe) {
                    // We didn't get a socket
                    // 发生错误则回收连接数
                    endpoint.countDownConnection();
                }
                // 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) {
            }
        }
        state = AcceptorState.ENDED;
    }

那怎么知道当前有多少连接呢?可以用lsof查看占用8080端口的文件

lsof -i :8080

也可以用netstat,需要注意的是,查看当前正在通信的连接需要筛选ESTABLISHED

netstat -anp | grep 8080 | grep ESTABLISHED

Poller-PollerEvent Queue

是一个Tomcat自己实现的轻量级的同步队列SynchronizedQueue,默认大小是128,超过后会自动扩容到当前的两倍。主要特点是很轻量,内部用了System.copy提升性能,对GC很友好。

// 注册PollerEvent到SynchronizedQueue-NioEndpoint.register
public void register(final NioChannel socket, final NioSocketWrapper socketWrapper) {
    socketWrapper.interestOps(SelectionKey.OP_READ);//this is what OP_REGISTER turns into.
    PollerEvent r = null;
    if (eventCache != null) {
        r = eventCache.pop();
    }
    if (r == null) {
        // 包装PollerEvent
        r = new PollerEvent(socket, OP_REGISTER);
    } else {
        r.reset(socket, OP_REGISTER);
    }
    // 往Poller.events(SynchronizedQueue)里加入PollerEvent
    addEvent(r);
}
//往Selector注册感兴趣的事件-PollerEvent.run
public void run() {
    if (interestOps == OP_REGISTER) {
        try {
            // 往Selector注册感兴趣的数据可读事件
            socket.getIOChannel().register(socket.getSocketWrapper().getPoller().getSelector(), SelectionKey.OP_READ, socket.getSocketWrapper());
        } catch (Exception x) {
            log.error(sm.getString("endpoint.nio.registerFail"), x);
        }
    }
}
// Poller线程执行逻辑-Poller.run
public void run() {
    while (true) {
        boolean hasEvents = false;
        try {
            if (!close) {
                // events()方法内部会遍历PollerEvent队列,并往Selector注册事件
                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 {
                    // 当数据可读时,selector会从readylist中返回可读的个数,一次可以查询到多个Channel的数据状态
                    keyCount = selector.select(selectorTimeout);
                }
                wakeupCounter.set(0);
            }
        } catch (Throwable x) {
        }

        Iterator<SelectionKey> iterator =
            keyCount > 0 ? selector.selectedKeys().iterator() : null;
        while (iterator != null && iterator.hasNext()) {
            // 轮询数据准备好的Channel
            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);
            }
        }
    }
}

Execotor-ThreadPoolExecutor、TaskQueue

为了性能最大化,Tomcat扩展了默认线程池策略和线程池队列。 当线程池达到最大数量时,不会立即执行拒绝,而是再次尝试向任务队列添加任务,如果还是添加不进去才执行拒绝

// 线程池扩展-org.apache.tomcat.util.threads.ThreadPoolExecutor.execute
public void execute(Runnable command, long timeout, TimeUnit unit) {
        submittedCount.incrementAndGet();
        try {
            super.execute(command);
        } catch (RejectedExecutionException rx) {
            // 当达到最大线程数时,会尝试把任务放到队列,如果还是放不进去才会拒绝
            if (super.getQueue() instanceof TaskQueue) {
                final TaskQueue queue = (TaskQueue)super.getQueue();
                try {
                    if (!queue.force(command, timeout, unit)) {
                        submittedCount.decrementAndGet();
                        throw new RejectedExecutionException(sm.getString("threadPoolExecutor.queueFull"));
                    }
                } catch (InterruptedException x) {
                    submittedCount.decrementAndGet();
                    throw new RejectedExecutionException(x);
                }
            } else {
                submittedCount.decrementAndGet();
                throw rx;
            }
        }
    }

Tomcat用的线程池队列是LinkedBlockingQueue,默认是用的无界模式,于是造成一个问题:当线程数达到核心线程数后,任务可以无限往队列添加,就不会再创建新线程了。Tomcat在往队列添加任务的逻辑中增加了maximumPoolSize的干预,使得在线程数未达到maximumPoolSize时任务添加不进去,进而可以新建线程。

// 线程池队列扩展-TaskQueue.offer
public boolean offer(Runnable o) {
    // 此方法被调用说明:当前线程数已经大于核心线程数 
    if (parent==null) return super.offer(o);

    // 线程数等于最大线程数时,不能再创建新线程,将任务放入队列
    if (parent.getPoolSize() == parent.getMaximumPoolSize()) return super.offer(o);

    // 当前线程数大于核心线程数,并且小于最大线程数

    // 如果任务数小于线程数,表名有空闲线程,不需要新建,放入队列
    if (parent.getSubmittedCount()<=(parent.getPoolSize())) return super.offer(o);

    // 如果任务数大于线程数,且线程数小于最大线程数,此时应该创建新线程
    if (parent.getPoolSize()<parent.getMaximumPoolSize()) return false;
        
    // 其他情况放入队列
    return super.offer(o);
}

总结

  1. IO模型介绍
  2. Tomcat实现的IO多路复用
  3. Acceptor、Poller、Executor
  4. Accept List、LimitLatch、SynchronizedQueue、线程池扩展

作者:Johnny

点击这里复制本文地址 以上内容由权冠洲的博客整理呈现,请务必在转载分享时注明本文地址!如对内容有疑问,请联系我们,谢谢!

支持Ctrl+Enter提交

联系我们| 本站介绍| 留言建议 | 交换友链 | 域名展示
本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除

权冠洲的博客 © All Rights Reserved.  Copyright quanguanzhou.top All Rights Reserved
苏公网安备 32030302000848号   苏ICP备20033101号-1

联系我们