概述
Tomcat涉及的内容较多，本文主要分析其关键部分的高并发设计。
主要内容
- IO模型概述
- Tomcat的IO模型实现
- Tomcat的并发控制
IO模型概述
从操作系统层面看服务端的数据交互流程：
image
如图所示，当服务端收到请求后，用户空间的用户线程会发起read调用，这个read调用依赖两个过程：从网卡拷贝数据到内核空间、从内核空间拷贝数据到用户空间，write调用是反向的两个步骤。
那在处理这两个耗时的Copy操作时，用户线程是占着CPU还是出让CPU？怎样可以更高效的处理数据？不同IO模型的作用就在这里体现出来了。
同步阻塞IO
用户线程发起read调用后就进入阻塞（出让CPU），等待read执行完（数据拷贝到用户空间）后再唤醒。
image
同步非阻塞IO
用户线程发起read调用后不进入阻塞，而是持续不断的发起read轮询，在数据拷贝到内核空间之前read调用都返回false，数据拷贝到内核空间后，read开始阻塞，等待数据从内核空间拷贝的用户空间后再唤醒用户线程。
image
IO多路复用
用户线程读取数据分成两步，先轮询调用select询问数据是否已到内核，如果到了则调用read读取数据，此时read调用会等待数据从内核空间拷贝到用户空间，这个过程是阻塞的。而这里多路复用的意思是：<font color="#cc0000">一次select可以得到多个channel的结果。</font>
image
异步IO
用户线程在调用read时注册一个回调函数，当数据到用户空间后会调用此回调通知用户线程，这个过程用户线程不阻塞。
image
本文介绍的Tomcat9默认是<font color="#cc0000">基于IO多路复用模型</font>做的的高并发设计。
Tomcat的IO模型实现
假设我们把Tomcat当做黑盒子，按照Spring的方式来处理请求，简化的流程是这样的：
image
Tomcat负责读取内核的数据，转换成Servlet对象，然后由Spring框架处理业务后通过Response对象写入返回数据，Tomcat再将返回数据通过内核写入网卡，最后返回到客户端。
接下来我们将Tomcat这部分放大，看看黑盒子里是怎么处理的。
image
如图所示，请求的处理分如下几个步骤：
- Tomcat在启动时会初始化一个ServerSocket用于监听指定端口的IO请求（比如8080）
- 接着启动Acceptor线程，循环调用accept方法接收IO请求（TCP连接建立）
- 将ServerChannel包装成PollerEvent，注册到Poller的event队列中
- Poller线程循环遍历event队列，将Poller关注的ServerChannel的READ操作注册到Selector中
- 在同一个Poller循环中，用Selector查询ServerChannel的状态，这里一次可以查询到多个ServerChannel的状态，即<font color="#cc0000">多路复用</font>
- 将查询到的SelectionKey对应的ServerChannel挨个创建SocketProcessor，SocketProcessor是Runnable的
- 将SocketProcessor扔到工作线程进行处理
- 接下来的协议解析、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多路复用模型实现，在这个实现中每一个环节都有影响并发的关键控制，先看图：
image
内核-Accept List
TCP三次握手建立连接的过程中，内核会为每一个LISTEN状态的Socket维护两个队列：
- SYN队列：这些连接已收到客户端的SYN
- ACCEPT队列：这些连接已经收到客户端的ACK，完成了三次握手，等待被系统accept调用取走
Tomcat的Acceptor负责从ACCEPT队列中取走连接，当Acceptor处理不过来时，连接就堆积在ACCEPT队列中，这个队列长度由<font color="#cc0000">acceptCount（默认100）</font>控制，当我们尝试把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实现等待。即如果服务端请求处理不过来，并发的请求数量超过了<font color="#cc0000">maxConnections（默认8192）</font>，则会进入等待。当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;
}
那怎么知道当前有多少连接呢？可以用<font color="#cc0000">lsof</font>查看占用8080端口的文件
lsof -i :8080
也可以用<font color="#cc0000">netstat</font>，需要注意的是，查看当前正在通信的连接需要筛选ESTABLISHED
netstat -anp | grep 8080 | grep ESTABLISHED
Poller-PollerEvent Queue
是一个Tomcat自己实现的轻量级的同步队列<font color="#cc0000">SynchronizedQueue</font>，默认大小是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扩展了默认线程池策略和线程池队列。
<font color="#cc0000">当线程池达到最大数量时，不会立即执行拒绝，而是再次尝试向任务队列添加任务，如果还是添加不进去才执行拒绝</font>：
// 线程池扩展-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用的线程池队列是<font color="#cc0000">LinkedBlockingQueue</font>，默认是用的无界模式，于是造成一个问题：当线程数达到核心线程数后，任务可以无限往队列添加，就不会再创建新线程了。Tomcat在往队列添加任务的逻辑中增加了<font color="#cc0000">maximumPoolSize</font>的干预，<font color="#cc0000">使得在线程数未达到maximumPoolSize时任务添加不进去</font>，进而可以新建线程。
// 线程池队列扩展-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);
}
总结
- IO模型介绍
- Tomcat实现的IO多路复用
- Acceptor、Poller、Executor
- Accept List、LimitLatch、SynchronizedQueue、线程池扩展
网友评论