前边一遍文章分析了NioEventLoop的实现原理，可以知道NioEventLoop主要跑两类任务：I/O任务和非I/O任务。其中I/O任务主要是进行Select选择出已注册的I/O事件并对这些I/O事件进行处理，执行的具体方法是processSelectedKeys()。下面我们就对这段代码进行具体分析，可参见其中的一个分支processSelectedKeysPlain(Set<SelectionKey> selectedKeys)。
具体代码如下：
遍历唤醒的SelectionKey，并取出对应key注册的attachment。判断attachment的类型：
1）是AbstractNioChannel，对应确定的一个Netty对应的NioChannel，对应执行processSelectedKey(SelectionKey k, AbstractNioChannel ch)方法。
2）是NioTask<SelectableChannel>,一般不会是这种类型，多数为是用户自定义的一个task。
private void processSelectedKeysPlain(Set<SelectionKey> selectedKeys) {
if (selectedKeys.isEmpty()) {
return;
}
Iterator<SelectionKey> i = selectedKeys.iterator();
for (;;) {
final SelectionKey k = i.next();
final Object a = k.attachment();
i.remove();
if (a instanceof AbstractNioChannel) {
processSelectedKey(k, (AbstractNioChannel) a);
} else {
@SuppressWarnings("unchecked")
NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
processSelectedKey(k, task);
}
if (!i.hasNext()) {
break;
}
if (needsToSelectAgain) {
selectAgain();
selectedKeys = selector.selectedKeys();
// Create the iterator again to avoid ConcurrentModificationException
if (selectedKeys.isEmpty()) {
break;
} else {
i = selectedKeys.iterator();
}
}
}
}
我们可以着重来看下processSelectedKey(SelectionKey k, AbstractNioChannel ch)的实现，根据readyOps确定感兴趣的事件类型，执行不同的操作：
1）SelectionKey.OP_CONNECT 连接事件，此处是对于client端而言的，该事件标志着三次握手阶段，client端收到了server端的ack报文。
2）SelectionKey.OP_WRITE 写事件，说明有数据要写。
3）SelectionKey.OP_READ | SelectionKey.OP_ACCEPT 读事件或接收连接事件。SelectionKey.OP_ACCEPT是对于server端而言的，标志有新的连接请求到达。
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
if (!k.isValid()) {
final EventLoop eventLoop;
try {
eventLoop = ch.eventLoop();
} catch (Throwable ignored) {
return;
}
if (eventLoop != this || eventLoop == null) {
return;
}
// close the channel if the key is not valid anymore
unsafe.close(unsafe.voidPromise());
return;
}
try {
int readyOps = k.readyOps();
// We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
// the NIO JDK channel implementation may throw a NotYetConnectedException.
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
// remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
// See https://github.com/netty/netty/issues/924
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
// Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
// Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
ch.unsafe().forceFlush();
}
// Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
// to a spin loop
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
1. OP_CONNECT事件分析
OP_CONNECT事件说明Server端已经回复了客户端建立连接的请求，下一步需要执行三次握手的第三步。具体执行的代码如下。首先将Selector上将OP_CONNECT从注册事件中去掉，然后会调用unsafe.finishConnect()。
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
// remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
// See https://github.com/netty/netty/issues/924
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
重点看一下unsafe.finishConnect()方法。doFinishConnect()会直接调用对应javaChannel的finishConnect()方法，此处不再做复杂介绍。在fulfillConnectPromise方法中，会尝试去判断该channel是否是active状态，是否被人cancell掉。如果当期状态是active的，会调用pipeline的fireChannelActive()方法。
public final void finishConnect() {
assert eventLoop().inEventLoop();
try {
boolean wasActive = isActive();
doFinishConnect();
fulfillConnectPromise(connectPromise, wasActive);
} catch (Throwable t) {
fulfillConnectPromise(connectPromise, annotateConnectException(t, requestedRemoteAddress));
} finally {
// Check for null as the connectTimeoutFuture is only created if a connectTimeoutMillis > 0 is used
// See https://github.com/netty/netty/issues/1770
if (connectTimeoutFuture != null) {
connectTimeoutFuture.cancel(false);
}
connectPromise = null;
}
}
private void fulfillConnectPromise(ChannelPromise promise, boolean wasActive) {
if (promise == null) {
// Closed via cancellation and the promise has been notified already.
return;
}
// Get the state as trySuccess() may trigger an ChannelFutureListener that will close the Channel.
// We still need to ensure we call fireChannelActive() in this case.
boolean active = isActive();
// trySuccess() will return false if a user cancelled the connection attempt.
boolean promiseSet = promise.trySuccess();
// Regardless if the connection attempt was cancelled, channelActive() event should be triggered,
// because what happened is what happened.
if (!wasActive && active) {
pipeline().fireChannelActive();
}
// If a user cancelled the connection attempt, close the channel, which is followed by channelInactive().
if (!promiseSet) {
close(voidPromise());
}
}
2. OP_WRITE事件分析
OP_WRITE事件说明该端有数据需要写出。通过代码可以看到write事件会调用ch.unsafe().forceFlush()方法，接下来直接调用AbstractChannel类的flush0()方法。在这个方法中，最核心的就是调用doWrite(outboundBuffer)，这是一个abstract方法，具体得由对应的Channel子类去实现。举例来说，对于NioSocketChannel类，outboundBuffer中的数据类型是byte，直接调用javaChannel的write方法即可；而对于AbstractNioMessageChannel类，outboundBuffer中的数据类型是Object，最终在调用javaChannel的write方法进行写操作之前需要进行转换将Object转化至byte类型。
protected void flush0() {
if (inFlush0) {
// Avoid re-entrance
return;
}
final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
if (outboundBuffer == null || outboundBuffer.isEmpty()) {
return;
}
inFlush0 = true;
// Mark all pending write requests as failure if the channel is inactive.
if (!isActive()) {
.......
}
try {
doWrite(outboundBuffer);
} catch (Throwable t) {
if (t instanceof IOException && config().isAutoClose()) {
} else {
outboundBuffer.failFlushed(t, true);
}
} finally {
inFlush0 = false;
}
}
3. SelectionKey.OP_READ | SelectionKey.OP_ACCEPT事件分析
Netty将Accept事件和READ事件进行了封装，统一调用unsafe.read()方法进行处理。我们以AbstractNioMessageChannel为例，看一下read()方法进行了哪些操作。
核心过程大致分为3步：
1）doReadMessages(readBuf)将数据读到readBuf。
2）遍历readBuf，对其中的每一个元素调用pipeline的fireChannelRead方法。
3）调用pipeline的fireChannelReadComplete方法。
4）removeReadOp()。
前边我们讨论过说Netty将Accept和Read进行了统一封装，而具体拆分的细节也是通过分别实现abstract方法进行不同的处理。比如，对于NioServerSocketChannel中，doReadMessages(readBuf)会调用SocketUtils.accept方法建立子连接并将该子连接放置到readBuf中，后续的pipeline在处理readBuf时也会有不同的处理，可以可到ServerBootstrapAcceptor中channelRead方法的实现；对于NioUdtMessageConnectorChannel，doReadMessages(readBuf)方法会从javaChannel中读取数据到readBuf
public void read() {
assert eventLoop().inEventLoop();
final ChannelConfig config = config();
final ChannelPipeline pipeline = pipeline();
final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
allocHandle.reset(config);
boolean closed = false;
Throwable exception = null;
try {
try {
do {
int localRead = doReadMessages(readBuf);
if (localRead == 0) {
break;
}
if (localRead < 0) {
closed = true;
break;
}
allocHandle.incMessagesRead(localRead);
} while (allocHandle.continueReading());
} catch (Throwable t) {
exception = t;
}
int size = readBuf.size();
for (int i = 0; i < size; i ++) {
readPending = false;
pipeline.fireChannelRead(readBuf.get(i));
}
readBuf.clear();
allocHandle.readComplete();
pipeline.fireChannelReadComplete();
if (exception != null) {
closed = closeOnReadError(exception);
pipeline.fireExceptionCaught(exception);
}
if (closed) {
inputShutdown = true;
if (isOpen()) {
close(voidPromise());
}
}
} finally {
if (!readPending && !config.isAutoRead()) {
removeReadOp();
}
}
}
