根据k的readOps,进行计算决定执行unsafe.read();
// todo 服务端启动后,方法被用用处理新链接, 可以模拟 telnet localhost 8899 新链接的介入 // todo 处理selectedkey // todo netty底层对数据的读写都是 unsafe完成的 private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) { // todo 这个unsafe 也是可channel 也是和Channel进行唯一绑定的对象 final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe(); if (!k.isValid()) { // todo 确保Key的合法 final EventLoop eventLoop; try { eventLoop = ch.eventLoop(); } catch (Throwable ignored) { return; } if (eventLoop != this || eventLoop == null) { // todo 确保多线程下的安全性 return; } unsafe.close(unsafe.voidPromise()); return; } // todo NioServerSocketChannel和selectKey都合法的话, 就进入下面的 处理阶段 try { // todo 获取SelectedKey 的 关心的选项 int readyOps = k.readyOps(); // todo 在read() write()之前我们需要调用 finishConnect() 方法, 否则 NIO JDK抛出异常 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( ); 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(); } // todo 同样是检查 readOps是否为零, 来检查是否出现了 jdk 空轮询的bug if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) { unsafe.read(); } } catch (CancelledKeyException ignored) { unsafe.close(unsafe.voidPromise()); } }下面我们进入unsafe.read();, 直接跳进去,直接进入到了AbstractNioChannel的抽象内部类,因为上面说了,做了向上强制类型转换,我们源码如下:
/** * Special {@link Unsafe} sub-type which allows to access the underlying {@link SelectableChannel} */ public interface NioUnsafe extends Unsafe { /** * Return underlying {@link SelectableChannel} */ SelectableChannel ch(); /** * Finish connect */ void finishConnect(); void forceFlush(); }具体的实现是谁? 因为我们是服务端的channel, 所以实现类是:NioMessageUnsafe, 进入查看他的源码: 下面这段代码真的是挺长的, 它的解析我写在他的下面:
@Override public void read() { // todo 同样是断言, 当前的线程必须是在 EventLoop 里面的线程才有资格执行 assert eventLoop().inEventLoop( ); final ChannelConfig config = config(); final ChannelPipeline pipeline = pipeline(); // todo 用于查看服务端接受的速率, 说白了就是控制服务端是否接着read 客户端的IO事件 final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle(); allocHandle.reset(config); boolean closed = false; Throwable exception = null; try { try { do { // todo 进入 int localRead = doReadMessages(readBuf); if (localRead == 0) { break; } if (localRead < 0) { closed = true; break; } //todo 对读到的连接,进行简单的计数 allocHandle.incMessagesRead(localRead); } while (allocHandle.continueReading()); } catch (Throwable t) { exception = t; } int size = readBuf.size(); for (int i = 0; i < size; i ++) { readPending = false; // todo 处理新的连接的逻辑来到这, 意思是让pipeline中发生事件传播, // todo pipeline是谁的呢? 现在是NioMessageUnsafe 所以是服务端的, // todo 事件是如何传播的呢? head-->>ServerBootStraptAcceptor-->>tail 依次传播, // todo 传播的什么事件? ChannelRead, 也就是说,会去调用 ServerBootStraptAcceptor的ChannelRead方法,跟进去 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 { // Check if there is a readPending which was not processed yet. // This could be for two reasons: // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method // // See https://github.com/netty/netty/issues/2254 if (!readPending && !config.isAutoRead()) { removeReadOp(); } } } } read()三部曲:针对这段代码,我们值关心下面几部分, 这三部分结束, 整个新链接的建立就完成了,
下面三部曲的 大前提都是,当前我们是在AbstractNioMessageChannel
doReadMessages(readBuf)
allocHandle.incMessagesRead(localRead);
pipeline.fireChannelRead(readBuf.get(i));
第一步: 如何创建出jdk原生的 客户端channel,对它做了什么?