由 dawei
对NIO的理解
个人单方面认为，NIO与BIO的最大区别在于主动和被动，使用BIO的方式需要等待被调用方返回数据，很明显此时调用者是被动的。
举个例子
阻塞IO
ASP站长网假设你是一个胆小又害羞的男孩子，你约了隔壁测试的妹子，但你并不敢主动约会，所以你把自己的手机号码给她，并暗示她想要约会的时候打电话给你。很明显此时你陷入了被动，约不约会的结果需要妹子主动告知你，如果她忘了，那么你要陷入长时间的等待中以及无尽的猜测和自我怀疑中（太惨了)。[如果你是一个胆小害羞又好色的男孩子，那就惨了]
非阻塞IO 我们知道，渣男通常有很多的备胎，我管这个叫做备胎池(SpareTirePool), 那么当他想要约会的时候，只要群发问妹子要不要约会，如果要约会的话就和妹子约会，约会结束之后，处理其他约会事件，如果没有继续下一次询问。在这个例子中约会可以视为IO事件，问妹子的过程可以视为备胎池的轮询。
Tomcat 如何使用NIO
既然是网络通信的I/O那必然有以下两个步骤
SeverSocket的启动
I/O事件的处理
关键代码在 package org.apache.tomcat.util.net.NioEndpoint 中
P.S. 文章太长，如果不想看可以直接阅读结论
ServerSocket的启动
在最开始看代码，是震惊的，真的，如果你看Reactor模型的话
以下bind方法代码是启动ServerSocket的流程，主要流程如下
绑定地址
设置接收新连接的方式为阻塞方式(关键点)
设置Acceptor和Poller的数量以及初始化SelectorPool
@Override
public void bind() throws Exception {
if (!getUseInheritedChannel()) {
serverSock = ServerSocketChannel.open();
socketProperties.setProperties(serverSock.socket());
InetSocketAddress addr = (getAddress()!=null?new InetSocketAddress(getAddress(),getPort()):new InetSocketAddress(getPort()));
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和Poller的数量
if (acceptorThreadCount == 0) {
// FIXME: Doesn't seem to work that well with multiple accept threads
// 顾名思义,Acceptor是用来处理新连接的
acceptorThreadCount = 1;
}
if (pollerThreadCount <= 0) {
// Poller 用来处理I/O事件
pollerThreadCount = 1;
}
setStopLatch(new CountDownLatch(pollerThreadCount));
// Initialize SSL if needed
initialiseSsl();
// 从此处可以看出tomcat池化了selector
selectorPool.open();
}
Tomcat NIO 如何处理I/O事件
先说结论，Tomcat NIO模型中有以下关键角色
Acceptor 用于接收新连接，每个Acceptor一个线程，以阻塞的方式接收新连接
Poller 当Acceptor接收到新连接，进行处理之后选择一个Poller处理该连接上的I/O事件。
LimitLatch 一个用来限制连接数的锁
Acceptor
Acceptor的主要工作就是不断接收来自客户端的连接，在简单处理之后将该连接交给Poller处理
接收来自客户端连接, 如果你不想看代码，以下是其主要流程
接收来自客户端的连接，并将其交给Poller处理
@Override
public void run() {
int errorDelay = 0;
// running的检测贯穿了Accpetor的处理流程，在每次关键操作的时候都会执行检测
while (running) {
// 如果进入暂停状态则每隔一段时间检测一下
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 {
//接收新连接
socket = serverSock.accept();
} catch (IOException ioe) {
// 发生异常,则减少连接数
countDownConnection();
if (running) {
handleExceptionWithDelay(errorDelay);
// re-throw
throw ioe;
} else {
break;
}
}
// Successful accept, reset the error delay
errorDelay = 0;
// Configure the socket
if (running && !paused) {
//setSocketOptions会导致将该连接交给Poller处理
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;
}
再来看看setSocketOptions做了什么,不想看代码的话，总结如下
将客户端socket设置为非阻塞模式
将客户端的socket封装为NioChannel或SecureNioChannel(使用了对象池技术)
从Poller池中获取一个Poller，将NioChannel注册到Poller上
protected boolean setSocketOptions(SocketChannel socket) {
// Process the connection
try {
//设置为非阻塞模式，以便通过selector进行查询
socket.configureBlocking(false);
Socket sock = socket.socket();
socketProperties.setProperties(sock);
//从对象池中获取一个NioChannel，tomcat会复用一切可以复用的对象以减少创建新对象所带来的消耗
NioChannel channel = nioChannels.pop();
if (channel == null) {
// 没有获取到，那就新建一个呗
SocketBufferHandler bufhandler = new SocketBufferHandler(
socketProperties.getAppReadBufSize(),
socketProperties.getAppWriteBufSize(),
socketProperties.getDirectBuffer());
// SSL这一块还没研究
if (isSSLEnabled()) {
channel = new SecureNioChannel(socket, bufhandler, selectorPool, this);
} else {
channel = new NioChannel(socket, bufhandler);
}
} else {
channel.setIOChannel(socket);
//重新设置SocketBufferHandler，将其设置为可写和可读
channel.reset();
}
//从Poller池中获取一个Poller(按照次序获取,可以理解为一个圆环)，并将Channel注册到上面
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;
}
Poller
从连接注册到Poller说起
不加锁的获取一个Poller
具体说明见代码
关键点：对一个数A取余会将余数的结果限制在A的范围内
/**
* Return an available poller in true round robin fashion.
* 很明显，取余的方式揭示了获取Poller的方法。你可以理解为
* Poller会组成一个圆环，这样我们就可以通过不断递增获取
* 下一个Poller,但是数据会溢出所以我们要取绝对值
* @return The next poller in sequence
*/
public Poller getPoller0() {
int idx = Math.abs(pollerRotater.incrementAndGet()) % pollers.length;
return pollers[idx];
}