前言
前面已经初步分析请求流程，下面我们继续。
Poller流程处理
从上一篇直到Acceptor接受到请求并注册到Poller中的events缓存栈中，下面来想起看一下Poller的处理流程。
public void run() {
// Loop until destroy() is called
while (true) {
boolean hasEvents = false;
try {
if (!close) {
//1. 更新PollerEvent队列，主要执行PollerEvent的run方法来更新selector感兴趣的事件
hasEvents = events();
//2. 将wakeupCounter设置为-1，如果oldvalue>0，做非阻塞select;否则做超时的阻塞select。其中wakeupCounter在#addEvent()时会加1
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 {
keyCount = selector.select(selectorTimeout);
}
//3. wakeupCounter设置为0
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(sm.getString("endpoint.nio.selectorLoopError"), x);
continue;
}
//4. keyCount为0，说明没有事件到来，再执行一次#events()
if (keyCount == 0) hasEvents = (hasEvents | events());
//5. 遍历SelectionKey(事件已到来，进行后续处理)，进行读写处理
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();
//6. 提交到线程池中进行处理处理
processKey(sk, attachment);
}
}
// Process timeouts
timeout(keyCount, hasEvents);
}
getStopLatch().countDown();
}
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(sm.getString("endpoint.nio.pollerEventError"), x);
}
}
return result;
}
主要流程如下：
- 调用#event()方法，更新selector感兴趣的事件；
- 执行#selector.select()方法，检测是否有事件到来；
- 将到来的事件提交至线程池进行下一步处理。
protected void processKey(SelectionKey sk, NioSocketWrapper socketWrapper) {
try {
if (close) {
cancelledKey(sk);
} else if (sk.isValid() && socketWrapper != null) {
if (sk.isReadable() || sk.isWritable()) {
if (socketWrapper.getSendfileData() != null) {
processSendfile(sk, socketWrapper, false);
} else {
//1. 在通道上注销对已经发生事件的关注
unreg(sk, socketWrapper, sk.readyOps());
boolean closeSocket = false;
// Read goes before write
if (sk.isReadable()) {
//2. 进行异步IO的处理或者交给SocketProcessor处理读操作
if (socketWrapper.readOperation != null) {
getExecutor().execute(socketWrapper.readOperation);
} else if (!processSocket(socketWrapper, SocketEvent.OPEN_READ, true)) {
closeSocket = true;
}
}
if (!closeSocket && sk.isWritable()) {
//3. 进行异步IO的处理或者交给SocketProcessor处理写操作
if (socketWrapper.writeOperation != null) {
getExecutor().execute(socketWrapper.writeOperation);
} else if (!processSocket(socketWrapper, SocketEvent.OPEN_WRITE, true)) {
closeSocket = true;
}
}
if (closeSocket) {
cancelledKey(sk);
}
}
}
} else {
// Invalid key
cancelledKey(sk);
}
} catch (CancelledKeyException ckx) {
cancelledKey(sk);
} catch (Throwable t) {
ExceptionUtils.handleThrowable(t);
log.error(sm.getString("endpoint.nio.keyProcessingError"), t);
}
}
//防止了通道对同一个事件不断select的问题
protected void unreg(SelectionKey sk, NioSocketWrapper socketWrapper, int readyOps) {
// This is a must, so that we don't have multiple threads messing with the socket
reg(sk, socketWrapper, sk.interestOps() & (~readyOps));
}
protected void reg(SelectionKey sk, NioSocketWrapper socketWrapper, int intops) {
sk.interestOps(intops);
socketWrapper.interestOps(intops);
}
这里主要注销了对已经发生事件的关注，然后将具体的处理逻辑交给SocketProcessor来处理，后面会介绍。
工作线程流程处理
从上面了解到，最后会调用#Poller.processSocket()方法，将处理逻辑交给SocketProcessor类，我们来看下：
public boolean processSocket(SocketWrapperBase<S> socketWrapper,
SocketEvent event, boolean dispatch) {
try {
if (socketWrapper == null) {
return false;
}
//1. 从缓存栈中获取SocketProcessor，无则创建否则重置SocketProcessor对象
SocketProcessorBase<S> sc = processorCache.pop();
if (sc == null) {
sc = createSocketProcessor(socketWrapper, event);
} else {
sc.reset(socketWrapper, event);
}
//2. 获取线程池，如果配置了线程池则将SocketProcessor提交到线程池中执行，否则直接执行SocketProcessor的run方法。
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;
}
如果配置了线程池，则提交SocketProcessor线程至线程池中执行，否则直接执行#SocketProcessor.run()。下面来看下SocketProcessor类：
public abstract class SocketProcessorBase<S> implements Runnable {
protected SocketWrapperBase<S> socketWrapper;
//socket事件状态
protected SocketEvent event;
public SocketProcessorBase(SocketWrapperBase<S> socketWrapper, SocketEvent event) {
reset(socketWrapper, event);
}
public void reset(SocketWrapperBase<S> socketWrapper, SocketEvent event) {
Objects.requireNonNull(event);
this.socketWrapper = socketWrapper;
this.event = event;
}
@Override
public final void run() {
synchronized (socketWrapper) {
// It is possible that processing may be triggered for read and
// write at the same time. The sync above makes sure that processing
// does not occur in parallel. The test below ensures that if the
// first event to be processed results in the socket being closed,
// the subsequent events are not processed.
if (socketWrapper.isClosed()) {
return;
}
doRun();
}
}
//子类实现
protected abstract void doRun();
}
protected class SocketProcessor extends SocketProcessorBase<NioChannel> {
public SocketProcessor(SocketWrapperBase<NioChannel> socketWrapper, SocketEvent event) {
super(socketWrapper, event);
}
@Override
protected void doRun() {
NioChannel socket = socketWrapper.getSocket();
SelectionKey key = socket.getIOChannel().keyFor(socket.getPoller().getSelector());
try {
int handshake = -1;
try {
if (key != null) {
//NioChannel默认返回true，SecureNioChannel这里才需要处理
if (socket.isHandshakeComplete()) {
// No TLS handshaking required. Let the handler
// process this socket / event combination.
handshake = 0;
} else if (event == SocketEvent.STOP || event == SocketEvent.DISCONNECT ||
event == SocketEvent.ERROR) {
// Unable to complete the TLS handshake. Treat it as
// if the handshake failed.
handshake = -1;
} else {
//具体处理
handshake = socket.handshake(key.isReadable(), key.isWritable());
// The handshake process reads/writes from/to the
// socket. status may therefore be OPEN_WRITE once
// the handshake completes. However, the handshake
// happens when the socket is opened so the status
// must always be OPEN_READ after it completes. It
// is OK to always set this as it is only used if
// the handshake completes.
event = SocketEvent.OPEN_READ;
}
}
} catch (IOException x) {
handshake = -1;
if (log.isDebugEnabled()) log.debug("Error during SSL handshake", x);
} catch (CancelledKeyException ckx) {
handshake = -1;
}
if (handshake == 0) {
SocketState state = SocketState.OPEN;
//最关键的代码，交给handler处理socket
if (event == null) {
state = getHandler().process(socketWrapper, SocketEvent.OPEN_READ);
} else {
state = getHandler().process(socketWrapper, event);
}
if (state == SocketState.CLOSED) {
close(socket, key);
}
} else if (handshake == -1) {
close(socket, key);
//如果handshake返回的是SelectionKey.OP_READ，注册读事件到Poller；如果返回的是SelectionKey.OP_WRITE，注册写事件到Poller，进行后续处理
} else if (handshake == SelectionKey.OP_READ) {
socketWrapper.registerReadInterest();
} else if (handshake == SelectionKey.OP_WRITE) {
socketWrapper.registerWriteInterest();
}
} catch (CancelledKeyException cx) {
socket.getPoller().cancelledKey(key);
} catch (VirtualMachineError vme) {
ExceptionUtils.handleThrowable(vme);
} catch (Throwable t) {
log.error(sm.getString("endpoint.processing.fail"), t);
socket.getPoller().cancelledKey(key);
} finally {
socketWrapper = null;
event = null;
//return to cache
if (running && !paused) {
//处理结束后，将SocketProcessor重新放入缓存栈中
processorCache.push(this);
}
}
}
}
这里SocketEvent有OPEN_READ、OPEN_WRITE、STOP、TIMEOUT、DISCONNECT、ERROR六中状态。SocketProcessor对象主要将socket交给Handler来处理请求。
总结
到这里NIO的整个处理流程就大致清楚了，整体流程如下：
NIO整体处理流程
下面深入Handler来看一下Socket请求是如何转换为Request对象，以及如何调用Servlet中的方法。
