说起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的目的：
//Connector开启
protected void startInternal() throws LifecycleException {
// 校验端口
if (getPort() < 0) {
throw new LifecycleException(sm.getString(
"coyoteConnector.invalidPort", Integer.valueOf(getPort())));
}
//设置Connector的状态为开启
setState(LifecycleState.STARTING);
try {
//开启protocolHandler
protocolHandler.start();
} catch (Exception e) {
String errPrefix = "";
if(this.service != null) {
errPrefix += "service.getName(): \"" + this.service.getName() + "\"; ";
}
throw new LifecycleException
(errPrefix + " " + sm.getString
("coyoteConnector.protocolHandlerStartFailed"), e);
}
}
//protocolHandler开启
public void start() throws Exception {
if (getLog().isInfoEnabled())
getLog().info(sm.getString("abstractProtocolHandler.start",
getName()));
try {
//endpoint开启，初始化Processor缓存、event缓存、exector线程池，开启轮询线程、acceptor线程
endpoint.start();
} catch (Exception ex) {
getLog().error(sm.getString("abstractProtocolHandler.startError",
getName()), ex);
throw ex;
}
// Start async timeout thread
asyncTimeout = new AsyncTimeout();
Thread timeoutThread = new Thread(asyncTimeout, getNameInternal() + "-AsyncTimeout");
int priority = endpoint.getThreadPriority();
if (priority < Thread.MIN_PRIORITY || priority > Thread.MAX_PRIORITY) {
priority = Thread.NORM_PRIORITY;
}
timeoutThread.setPriority(priority);
timeoutThread.setDaemon(true);
timeoutThread.start();
}
至此，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方法
public final void start() throws Exception {
if (bindState == BindState.UNBOUND) {
bind();
bindState = BindState.BOUND_ON_START;
}
startInternal();
}
其中，bind()方法和startInternal()方法，由其子类具体实现。
bind()方法用于初始化endpoint，绑定监听端口等、设置最大线程数、ssl等。
startInternal()方法在EndPoint初始化完毕后，创建pollers轮询线程以及acceptors线程并开启。
/**
* 开启 NIO endpoint, 创建pollers轮询线程以及acceptors线程
*/
@Override
public void startInternal() throws Exception {
if (!running) {
running = true;
paused = false;
//SocketProcessor缓存。若缓存没有，则创建新的SocketProcessor
processorCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
socketProperties.getProcessorCache());
//poller事件缓存
eventCache = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
socketProperties.getEventCache());
//nioChannels缓存。
nioChannels = new SynchronizedStack<>(SynchronizedStack.DEFAULT_SIZE,
socketProperties.getBufferPool());
// Create worker collection 创建线程池
if ( getExecutor() == null ) {
createExecutor();
}
initializeConnectionLatch();
// Start poller threads
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();
}
//start acceptor threads
startAcceptorThreads();
}
}
protected final void startAcceptorThreads() {
int count = getAcceptorThreadCount();
acceptors = new Acceptor[count];
for (int i = 0; i < count; i++) {
acceptors[i] = createAcceptor();
String threadName = getName() + "-Acceptor-" + i;
acceptors[i].setThreadName(threadName);
Thread t = new Thread(acceptors[i], threadName);
t.setPriority(getAcceptorThreadPriority());
t.setDaemon(getDaemon());
t.start();
}
}
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内核设计剖析》
