Netty启动源码NioEventLoop剖析accept剖析read剖析write剖析

学习链接

NIO&Netty - 专栏

• Netty核心技术十–Netty 核心源码剖析
• Netty核心技术九–TCP 粘包和拆包及解决方案
• Netty核心技术七–Google Protobuf
• Netty核心技术六–Netty核心模块组件
• Netty核心技术五–Netty高性能架构设计

聊聊Netty那些事儿 - 专栏

• 一文搞懂Netty发送数据全流程 | 你想知道的细节全在这里

netty源码解析 - 系列

• Netty源码分析 （一）----- NioEventLoopGroup
• Netty源码分析 （二）----- ServerBootstrap
• Netty源码分析 （三）----- 服务端启动源码分析
• Netty源码分析 （四）----- ChannelPipeline
• Netty源码分析 （五）----- 数据如何在 pipeline 中流动
• Netty源码分析 （六）----- 客户端接入accept过程
• Netty源码分析 （七）----- read过程 源码分析
• Netty源码分析 （八）----- write过程 源码分析
• Netty源码分析 （九）----- 拆包器的奥秘
• Netty源码分析 （十）----- 拆包器之LineBasedFrameDecoder
• Netty源码分析 （十一）----- 拆包器之LengthFieldBasedFrameDecoder
• Netty源码分析 （十二）----- 心跳服务之 IdleStateHandler 源码分析

1. 源码分析

1.1 启动剖析

我们就来看看 netty 中对下面的代码是怎样进行处理的

（先明确Java nio的基础步骤如下，而netty在启动过程中，也是需要做下面的事情的）

//1 netty 中使用 NioEventLoopGroup （简称 nio boss 线程）来封装线程和 selector
Selector selector = Selector.open(); //2 创建 NioServerSocketChannel，同时会初始化它关联的 handler，以及为原生 ssc 存储 config
NioServerSocketChannel attachment = new NioServerSocketChannel();//3 创建 NioServerSocketChannel 时，创建了 java 原生的 ServerSocketChannel
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); 
serverSocketChannel.configureBlocking(false);//4 启动 nio boss 线程执行接下来的操作//5 注册（仅关联 selector 和 NioServerSocketChannel），未关注事件
// （注意，这里将NioServerSocketChannel作为附件绑定到了selectionKey上，当此ServerSocketChannel有可连接事件时，就可以获取到此selectionKey，从而获取到对应的NioServerSocketChannel）
SelectionKey selectionKey = serverSocketChannel.register(selector, 0, attachment);// （ServerBootstrapAcceptor是ChannelInboundHandlerAdapter入站类型的处理器）
//6 head -> 初始化器 -> ServerBootstrapAcceptor -> tail，初始化器是一次性的，只为添加 acceptor//7 绑定端口
serverSocketChannel.bind(new InetSocketAddress(8080));//8 触发 channel active 事件，在 head 中关注 op_accept 事件
selectionKey.interestOps(SelectionKey.OP_ACCEPT);

源码入口，可以从下面的代码进入

（暂时先不看NioEventLoopGroup，而Selector是存在于NioEventLoop中的，所以selector.open暂时不看）

public class TestSourceServer {public static void main(String[] args) {new ServerBootstrap().group(new NioEventLoopGroup()).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) {ch.pipeline().addLast(new LoggingHandler());}}).bind(8880); // 以bind为入口}
}

AbstractBootstrap#doBind

入口 io.netty.bootstrap.ServerBootstrap#bind

关键代码 io.netty.bootstrap.AbstractBootstrap#doBind

（1、注意main线程和nio线程的切换；

​ 2、initAndRegister 对应 nio中 创建ServerSocketChannel 和 把ServerSocketChannel注册到selector上

​ 3、doBind0 对应 nio中 bind监听端口）

private ChannelFuture doBind(final SocketAddress localAddress) {// 1. 执行初始化和注册 regFuture 会由 initAndRegister 设置其是否完成，从而回调 3.2 处代码final ChannelFuture regFuture = initAndRegister();final Channel channel = regFuture.channel();if (regFuture.cause() != null) {return regFuture;}// 2. 因为 initAndRegister 是异步执行，需要分两种情况来看，调试时也需要通过 suspend 断点类型加以区分// 2.1 如果已经完成（如果前面做的比较快，就进入这个if块）if (regFuture.isDone()) {ChannelPromise promise = channel.newPromise();// 3.1 立刻调用 doBind0doBind0(regFuture, channel, localAddress, promise);return promise;} // 2.2 还没有完成else {final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);// 3.2 回调 doBind0regFuture.addListener(new ChannelFutureListener() {// （这个operationComplete是由nio线程来调用的）@Overridepublic void operationComplete(ChannelFuture future) throws Exception {Throwable cause = future.cause();if (cause != null) {// 处理异常...promise.setFailure(cause);} else {promise.registered();// 3. 由注册线程去执行 doBind0doBind0(regFuture, channel, localAddress, promise);}}});return promise;}
}

AbstractBootstrap#initAndRegister

关键代码 io.netty.bootstrap.AbstractBootstrap#initAndRegister

final ChannelFuture initAndRegister() {Channel channel = null;try {//（1、这里会去调用NioServerSocketChannel的无参构造方法去得到channel// 2、在NioServerSocketChannel的无参构造方法中会去创建javanio的ServerSocketChannel,//    并且将该ServerSocketChannel维护在NioServerSocketChannel中，//    并配置为非阻塞模式，感兴趣的事件是OP_ACCEPT，但是还没注册到selector上，//    只是维护了这些基本信息到NioServerSocketChannel。//    并且在NioServerSocketChannel的构造方法中会去创建NioServerSocketChannelConfig//    维护到NioServerSocketChannel中。//    并且NioServerSocketChannel的构造方法中会去创建DefaultChannelPipeline//    维护到NioServerSocketChannel中）channel = channelFactory.newChannel();// 1.1 初始化 - 做的事就是添加一个初始化器 ChannelInitializerinit(channel);} catch (Throwable t) {// 处理异常...return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);}// 1.2 注册 - 做的事就是将原生 channel 注册到 selector 上// （这里会从eventLoopGroup中挑选出1个eventLoop来注册ServerSocketChannel）ChannelFuture regFuture = config().group().register(channel);if (regFuture.cause() != null) {// 处理异常...}return regFuture;
}

ServerBootstrap#init

关键代码 io.netty.bootstrap.ServerBootstrap#init

（这里会给ServerSocketChannel的pipeline中添加1个ChannelInitializer初始化器，该初始化器只会执行1次，后续将会移除掉。）

// 这里 channel 实际上是 NioServerSocketChannel
void init(Channel channel) throws Exception {final Map<ChannelOption<?>, Object> options = options0();synchronized (options) {setChannelOptions(channel, options, logger);}final Map<AttributeKey<?>, Object> attrs = attrs0();synchronized (attrs) {for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {@SuppressWarnings("unchecked")AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();channel.attr(key).set(e.getValue());}}ChannelPipeline p = channel.pipeline();final EventLoopGroup currentChildGroup = childGroup;final ChannelHandler currentChildHandler = childHandler;final Entry<ChannelOption<?>, Object>[] currentChildOptions;final Entry<AttributeKey<?>, Object>[] currentChildAttrs;synchronized (childOptions) {currentChildOptions = childOptions.entrySet().toArray(newOptionArray(0));}synchronized (childAttrs) {currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0));}// 为 NioServerSocketChannel 的pipeline 添加初始化器!!!// （1、该初始化器的initChannel方法只会执行1次，后续该初始化器将会移除掉，//      移除动作是在ChannelInitializer#initChannel中操作的。//  2、注意该初始化器的initChannel方法在此处尚未被调用。//  3、initChannel方法的调用时机是在AbstractChannel的register0方法中，//     在做完将channel注册到selector上之后的//     pipeline.invokeHandlerAddedIfNeeded()这句代码调用的）p.addLast(new ChannelInitializer<Channel>() {@Overridepublic void initChannel(final Channel ch) throws Exception {final ChannelPipeline pipeline = ch.pipeline();// （这里意味着可以通过配置给config1个handler，//  从而给serverSocketChannel的pipeline添加1个handler）ChannelHandler handler = config.handler();if (handler != null) {pipeline.addLast(handler);}// 1、初始化器的职责是将 ServerBootstrapAcceptor 加入至 NioServerSocketChannel// 2、ServerBootstrapAcceptor的作用是在selector触发可连接事件时，建立连接// 3、保证添加这个动作是在nio线程中完成的ch.eventLoop().execute(new Runnable() {@Overridepublic void run() {pipeline.addLast(new ServerBootstrapAcceptor(ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));}});}});
}

AbstractUnsafe#register

关键代码 io.netty.channel.AbstractChannel.AbstractUnsafe#register

public final void register(EventLoop eventLoop, final ChannelPromise promise) {// 一些检查，略...AbstractChannel.this.eventLoop = eventLoop;// （判断当前线程是不是eventLoop的线程，因为顺着刚刚的逻辑，当前还在主线程中，所以走else）if (eventLoop.inEventLoop()) {register0(promise);} else {try {// 首次执行 execute 方法时，会启动 nio 线程，之后注册等操作在 nio 线程上执行// 因为只有一个 NioServerSocketChannel 因此，也只会有一个 boss nio 线程// 这行代码完成的事实是 main -> nio boss 线程的切换eventLoop.execute(new Runnable() {@Overridepublic void run() {// （该方法在nio线程上执行，并且注意promise传进去了，用于通知其它线程）register0(promise);}});} catch (Throwable t) {// 日志记录...closeForcibly();closeFuture.setClosed();safeSetFailure(promise, t);}}
}

#####- *AbstractUnsafe#register0

io.netty.channel.AbstractChannel.AbstractUnsafe#register0

private void register0(ChannelPromise promise) {try {if (!promise.setUncancellable() || !ensureOpen(promise)) {return;}boolean firstRegistration = neverRegistered;// 1.2.1 【将原生的 nio channel 绑定到 selector 上】，//       注意此时没有注册 selector 关注事件，附件为 NioServerSocketChanneldoRegister();neverRegistered = false;registered = true;// 1.2.2 执行 NioServerSocketChannel 初始化器的 initChannel//（1、调用到ServerBootstrap的init方法中为NioServerSocketChannel的//    pipeline添加的初始化器的initChannel方法。//  2、该initChannel方向pipeline中添加了ServerBootstrapAcceptor这个入站处理器）pipeline.invokeHandlerAddedIfNeeded();// （给promise对象1个成功的结果，这样前面的监听就能收到这个结果触发operationComplete方法，//   就会去通知前面在AbstractBootstrap#doBind方法中注册的监听去做doBind0绑定监听端口）// 回调 3.2 io.netty.bootstrap.AbstractBootstrap#doBind0safeSetSuccess(promise);pipeline.fireChannelRegistered();// 对应 server socket channel 还未绑定，isActive 为 falseif (isActive()) {if (firstRegistration) {pipeline.fireChannelActive();} else if (config().isAutoRead()) {beginRead();}}} catch (Throwable t) {// Close the channel directly to avoid FD leak.closeForcibly();closeFuture.setClosed();safeSetFailure(promise, t);}
}protected void doRegister() throws Exception {boolean selected = false;for (;;) {try {// 将java的channel注册到了eventLoop的selector上// （此时，尚未注册感兴趣的事件。同时，注意当前this作为附件绑定到了selectionKey）selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);return;} catch (CancelledKeyException e) {if (!selected) {eventLoop().selectNow();selected = true;} else {throw e;}}}}

- ChannelInitializer#initChannel

关键代码 io.netty.channel.ChannelInitializer#initChannel

private boolean initChannel(ChannelHandlerContext ctx) throws Exception {if (initMap.add(ctx)) { // Guard against re-entrance.try {// 1.2.2.1 执行初始化！！！（调用前面添加的初始化器的initChannel方法）initChannel((C) ctx.channel());} catch (Throwable cause) {exceptionCaught(ctx, cause);} finally {// 1.2.2.2 移除初始化器！！！（调用完成后，移除初始化器）ChannelPipeline pipeline = ctx.pipeline();if (pipeline.context(this) != null) {pipeline.remove(this);}}return true;}return false;
}

AbstractBootstrap#doBind0

关键代码 io.netty.bootstrap.AbstractBootstrap#doBind0

// 3.1 或 3.2 执行 doBind0
private static void doBind0(final ChannelFuture regFuture, final Channel channel,final SocketAddress localAddress, final ChannelPromise promise) {// 1、确保执行是在nio eventLoop线程中执行// 2、绑定会从pipe的tail开始找，最终会到headContext中调用到AbstractUnsafe的bind方法channel.eventLoop().execute(new Runnable() {@Overridepublic void run() {if (regFuture.isSuccess()) {channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);} else {promise.setFailure(regFuture.cause());}}});
}

AbstractUnsafe#bind

关键代码 io.netty.channel.AbstractChannel.AbstractUnsafe#bind

public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {assertEventLoop();if (!promise.setUncancellable() || !ensureOpen(promise)) {return;}if (Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) &&localAddress instanceof InetSocketAddress &&!((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress() &&!PlatformDependent.isWindows() && !PlatformDependent.maybeSuperUser()) {// 记录日志...}boolean wasActive = isActive();try {// 3.3 【执行端口绑定】doBind(localAddress);} catch (Throwable t) {safeSetFailure(promise, t);closeIfClosed();return;}// 从这里可以看出是绑定端口后，再去触发active事件的// 当前serverSocketChannel的pipeline已经添加了head-acceptor-tail处理器链, // 并且已经绑定好端口了，所以这里触发pipeline上所有handler的active事件，// 接下来，去看HeadContext#channelActive方法if (!wasActive && isActive()) {invokeLater(new Runnable() {@Overridepublic void run() {// 3.4 【触发 active 事件】pipeline.fireChannelActive();}});}safeSetSuccess(promise);
}

- NioServerSocketChannel#doBind

3.3 关键代码 io.netty.channel.socket.nio.NioServerSocketChannel#doBind

protected void doBind(SocketAddress localAddress) throws Exception {if (PlatformDependent.javaVersion() >= 7) {// 调用java原生channel的绑定端口的方法javaChannel().bind(localAddress, config.getBacklog());} else {javaChannel().socket().bind(localAddress, config.getBacklog());}
}

- HeadContext#channelActive

3.4 关键代码 io.netty.channel.DefaultChannelPipeline.HeadContext#channelActive

public void channelActive(ChannelHandlerContext ctx) {// 触发所有handler的active事件ctx.fireChannelActive();// 从这里可以看出是所有handler的active触发之后，再将channel注册感兴趣的事件的// 触发 read ，目的是为了触发channel的事件注册，注册OP_ACCEPT事件，//            见AbstractNioChannel#doBeginRead// (注意: NioServerSocketChannel 上的 read 不是读取数据，只是为了触发 channel 的事件注册)readIfIsAutoRead();
}

– AbstractNioChannel#doBeginRead

关键代码 io.netty.channel.nio.AbstractNioChannel#doBeginRead

protected void doBeginRead() throws Exception {// Channel.read() or ChannelHandlerContext.read() was calledfinal SelectionKey selectionKey = this.selectionKey;if (!selectionKey.isValid()) {return;}readPending = true;final int interestOps = selectionKey.interestOps();// readInterestOp 取值是 16，在NioServerSocketChannel创建时初始化好，代表关注 accept 事件if ((interestOps & readInterestOp) == 0) {// 注册感兴趣的事件！！！selectionKey.interestOps(interestOps | readInterestOp);}
}

1.2 NioEventLoop 剖析

NioEventLoop的重要组成

1、在NioEventLoop类中有成员变量

private Selector selector; 
private Selector unwrappedSelector;

2、在NioEventLoop的父类SingleThreadEventExecutor中有成员变量：

private volatile Thread thread;// 使用的跟上面同1个thread
private final Executor executor; // 由于eventLoop是单线程，其它的任务先放在taskQueue任务队列中，然后由单线程依次执行
private final Queue<Runnable> taskQueue;

3、在NioEventLoop的父类的父类AbstractScheduledEventExecutor中有成员变量

// 用来处理定时任务
PriorityQueue<ScheduledFutureTask<?>> scheduledTaskQueue;

NioEventLoop 线程不仅要处理 IO 事件，还要处理 Task（包括普通任务和定时任务）

selector何时创建

在NioEventLoop的唯一构造方法中，创建了selector

NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler,EventLoopTaskQueueFactory taskQueueFactory, EventLoopTaskQueueFactory tailTaskQueueFactory) {super(parent, executor, false, newTaskQueue(taskQueueFactory), newTaskQueue(tailTaskQueueFactory),rejectedExecutionHandler);this.provider = ObjectUtil.checkNotNull(selectorProvider, "selectorProvider");this.selectStrategy = ObjectUtil.checkNotNull(strategy, "selectStrategy");// 在这里创建selectorfinal SelectorTuple selectorTuple = openSelector();this.selector = selectorTuple.selector;this.unwrappedSelector = selectorTuple.unwrappedSelector;
}

那为什么会有2个selector呢？

因为netty要把selector里面原来的selectionKey的set实现改为用数组实现，因为数组遍历的性能比set好！

nio线程在何时启动

（当首次调用eventLoop的execute方法时，会启动线程，并且state状态位控制只会启动1次。）

public class TestEventLoop {public static void main(String[] args) {NioEventLoopGroup group = new NioEventLoopGroup();group.next()// 入口.execute(() -> {System.out.println("Hello");});}}

SingleThreadEventExecutor#execute

提交任务代码 io.netty.util.concurrent.SingleThreadEventExecutor#execute

public void execute(Runnable task) {if (task == null) {throw new NullPointerException("task");}// 判断当前线程是否是eventLoop的thread，很显然，现在eventLoop的thread是nullboolean inEventLoop = inEventLoop();// 添加任务，其中队列使用了 jctools 提供的 mpsc 无锁队列addTask(task);if (!inEventLoop) {// inEventLoop 如果为 false 表示由其它线程来调用 execute，// 即首次调用时，需要向 eventLoop 提交首个任务，启动死循环，会执行到下面的 doStartThreadstartThread();if (isShutdown()) {// 如果已经 shutdown，做拒绝逻辑，代码略...}}if (!addTaskWakesUp && wakesUpForTask(task)) {// 如果线程由于 IO select 阻塞了，添加任务的线程需要负责唤醒 NioEventLoop 线程wakeup(inEventLoop);}
}private void startThread() {if (state == ST_NOT_STARTED) {if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {boolean success = false;try {// 启动线程doStartThread();success = true;} finally {if (!success) {STATE_UPDATER.compareAndSet(this, ST_STARTED, ST_NOT_STARTED);}}}}
}private void doStartThread() {assert thread == null;// 这个executor是在 MultithreadEventExecutorGroup的构造方法中初始化的，//               （直接new的ThreadPerTaskExecutor）executor.execute(new Runnable() {@Overridepublic void run() {// 将线程池的当前线程保存在成员变量中，以便后续使用// 将thread线程设置为执行线程// （所以eventLoopgroup中的executor属性的线程和thread属性是同一个线程）thread = Thread.currentThread();if (interrupted) {thread.interrupt();}boolean success = false;updateLastExecutionTime();try {// 调用外部类 SingleThreadEventExecutor 的 run 方法，进入死循环，run 方法见下// 【启动 EventLoop 主循环 】SingleThreadEventExecutor.this.run();success = true;} catch (Throwable t) {logger.warn("Unexpected exception from an event executor: ", t);} finally { // 清理工作，代码略... }}});
}@Override
protected void wakeup(boolean inEventLoop) {// !eventLoop的理解: 只有其它非nio线程提交任务，才会有机会去唤醒selector停止阻塞//                  （因为如果是eventLoop自己提交任务给自己，在提交的时候，//                    当前eventLoop正在运行，没有阻塞，所以不需要唤醒selector）// wakenUp的理解: 当多个其它非nio线程提交任务，那么只会将selector唤醒1次if (!inEventLoop && wakenUp.compareAndSet(false, true)) {// 唤醒 select 阻塞线程// （这个wakeup调用后，如果selector正在select，那么直接唤醒，//   如果selector还没select，那么该selector去select时，就不会阻塞。//   类似于LockSupport的park和unpark。）selector.wakeup();}
}

*NioEventLoop#run

io.netty.channel.nio.NioEventLoop#run 主要任务是执行死循环，不断看有没有新任务，有没有定时任务，有没有 IO 事件，如果有，则执行。

// 死循环执行
for (;;) {try {try {//calculateStrategy 的逻辑如下：/* [代码] hasTasks ? selectNow() : SelectStrategy.SELECT; */// 当有任务时, 会执行一次selectNow(),去获取看看是否有io事件，//           并且会清除上一次的wakeup结果, 无论有没有IO事件,都会跳过switch// （因为有任务的话，即便没有io事件，也得干活，所以没有必要阻塞了）// 当没有任务，会匹配 SelectStrategy.SELECT，看是否应当阻塞// （因为没有任务的话，那就等有io事件了，再干活，所以就设置超时阻塞，//   同时还要看在阻塞期间有其它非nio线程提交任务，并唤醒selector。//   那么默认阻塞多久呢？那就需要看NioEventLoop#select(boolean oldWakenUp)方法//    默认是阻塞1s + 0.5ms，不过还得看有没有定时任务。）switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {case SelectStrategy.CONTINUE: // -2continue;case SelectStrategy.BUSY_WAIT: // -3case SelectStrategy.SELECT: // -1// 因为IO线程和提交任务线程都有可能执行 wakeup，而 wakeup 属于比较昂贵的操作，// 因此使用了一个原子布尔对象 wakenUp，它取值为 true 时，表示该由当前线程唤醒// 进行 select 阻塞，并设置唤醒状态为 falseboolean oldWakenUp = wakenUp.getAndSet(false);// 这里select方法中会调用select(timeoutMillis)阻塞，那么什么时候唤醒呢？// 当有io事件时自动唤醒// 或者超时自动唤醒// 或者有任务提交时,手动唤醒以便及时处理io事件以外的普通任务// 如果在这个位置，非 EventLoop 线程抢先将 wakenUp 置为 true，并 wakeup// 下面的 select 方法不会阻塞// 等 runAllTasks 处理完成后，到再循环进来这个阶段新增的任务会不会及时执行呢?// 因为 oldWakenUp 为 true，因此下面的 select 方法就会阻塞，直到超时// 才能执行，让 select 方法无谓阻塞select(oldWakenUp);if (wakenUp.get()) {selector.wakeup();}default:}} catch (IOException e) {rebuildSelector0();handleLoopException(e);continue;}// 有任务 或者 正在等待io事件但io事件还没来就被唤醒了 或者 io事件来了cancelledKeys = 0;needsToSelectAgain = false;// （如果eventLoop在执行非io任务的事件过长，势必会影响到io事件的处理）// ioRatio 默认是 50final int ioRatio = this.ioRatio;// 如果ioRatio设置为 100，那么会让普通任务都运行完。// 如果ioRatio不设置为100，那么会根据io事件处理的运行时间，算出普通任务可以运行的时间，//                       算出的这个时间仅仅是用来判断要不要继续运行下1个普通任务，//                       因此，如果1个普通任务本身耗时就特别长，//                            这里是没有中断这个任务的说法的，而且还得任务响应中断才行。if (ioRatio == 100) {try {processSelectedKeys();} finally {// ioRatio 为 100 时，总是运行完所有非 IO 任务runAllTasks();}} else {                final long ioStartTime = System.nanoTime();try {processSelectedKeys();} finally {// 记录 io 事件处理耗时final long ioTime = System.nanoTime() - ioStartTime;// 运行非 IO 任务，一旦超时会退出 runAllTasksrunAllTasks(ioTime * (100 - ioRatio) / ioRatio);}}} catch (Throwable t) {handleLoopException(t);}try {if (isShuttingDown()) {closeAll();if (confirmShutdown()) {return;}}} catch (Throwable t) {handleLoopException(t);}
}

NioEventLoop#select

io.netty.channel.nio.NioEventLoop#select

private void select(boolean oldWakenUp) throws IOException {Selector selector = this.selector;try {int selectCnt = 0;long currentTimeNanos = System.nanoTime();// 计算等待时间// * 没有 scheduledTask定时任务，超时时间为 1s// * 有 scheduledTas定时任务k，超时时间为 `下一个定时任务执行时间 - 当前时间`long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);for (;;) {long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;// 如果超时，退出循环if (timeoutMillis <= 0) {if (selectCnt == 0) {selector.selectNow();selectCnt = 1;}break;}// 如果期间又有task退出循环，如果没这个判断，那么任务就会等到下次select超时时才能被执行// wakenUp.compareAndSet(false, true) 是让非 NioEventLoop 不必再执行 wakeupif (hasTasks() && wakenUp.compareAndSet(false, true)) {selector.selectNow();selectCnt = 1;break;}// select 有限时阻塞// 注意 nio 有 bug，当 bug 出现时，select 方法即使没有时间发生，也不会阻塞住，// 导致不断空轮询，cpu 占用 100%// （所以就用了 selectCnt ++ 来统计次数，因为如果bug发生的话，循环会很快，//   这样selectCnt就会猛增，就检测到了）int selectedKeys = selector.select(timeoutMillis);// 计数加 1selectCnt ++;// 醒来后，如果有 IO 事件、或是由非 EventLoop 线程唤醒，或者有任务，退出循环if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks()|| hasScheduledTasks()) {break;}if (Thread.interrupted()) {// 线程被打断，退出循环// 记录日志selectCnt = 1;break;}long time = System.nanoTime();if (time-TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {// 如果超时，计数重置为 1，下次循环就会 breakselectCnt = 1;} // 计数超过阈值，由 io.netty.selectorAutoRebuildThreshold 指定，默认 512// 这是为了解决 nio 空轮询 bug// （重新创建1个selector，来替换原来的selector，来解决nio空轮询bug）else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 && selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {// 重建 selectorselector = selectRebuildSelector(selectCnt);selectCnt = 1;break;}currentTimeNanos = time;}if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS) {// 记录日志}} catch (CancelledKeyException e) {// 记录日志}
}

NioEventLoop#processSelectedKeys

处理 keys io.netty.channel.nio.NioEventLoop#processSelectedKeys

private void processSelectedKeys() {if (selectedKeys != null) {// 通过反射将 Selector 实现类中的就绪事件集合替换为 SelectedSelectionKeySet // SelectedSelectionKeySet 底层为数组实现，可以提高遍历性能（原本为 HashSet）processSelectedKeysOptimized();} else {processSelectedKeysPlain(selector.selectedKeys());}
}private void processSelectedKeysOptimized() {// 遍历所有的selectionKeyfor (int i = 0; i < selectedKeys.size; ++i) {final SelectionKey k = selectedKeys.keys[i];// 获取完就置为nullselectedKeys.keys[i] = null;// 附件就是 NioServerSocketChannelfinal Object a = k.attachment();if (a instanceof AbstractNioChannel) {// 处理selectionKeyprocessSelectedKey(k, (AbstractNioChannel) a);} else {@SuppressWarnings("unchecked")NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;processSelectedKey(k, task);}if (needsToSelectAgain) {selectedKeys.reset(i + 1);selectAgain();i = -1;}}
}

- NioEventLoop#processSelectedKey

io.netty.channel.nio.NioEventLoop#processSelectedKey

private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();// 当 key 取消或关闭时会导致这个 key 无效if (!k.isValid()) {// 无效时处理...return;}try {int readyOps = k.readyOps();// 连接事件// （这个是客户端需要监听处理的事件，服务端就不用管）if ((readyOps & SelectionKey.OP_CONNECT) != 0) {int ops = k.interestOps();ops &= ~SelectionKey.OP_CONNECT;k.interestOps(ops);unsafe.finishConnect();}// OP_ACCEPT 和 OP_CONNECT的理解：// ServerBootstrap用来处理服务器端，而Bootstrap用于客户端。当服务器绑定端口后，会注册OP_ACCEPT事件，等待客户端连接。一旦有连接进来，就会触发这个事件，然后创建子Channel来处理通信。而客户端在连接服务器时，会发起非阻塞的连接操作，这时候会注册OP_CONNECT，当连接建立完成后，触发该事件，之后就可以进行读写操作了// OP_CONNECT只是在连接过程中注册，一旦连接成功，就会触发，之后可能需要修改感兴趣的事件为OP_READ等。// 需要注意，当连接失败时，OP_CONNECT也会触发，这时候需要处理异常情况。比如，在Netty中，连接失败会触发相应的异常处理机制，比如channel的exceptionCaught方法// 总结来说，OP_ACCEPT是服务器端用于接收新连接，而OP_CONNECT是客户端用于处理连接建立完成的事件// 可读：当客户端或服务端的缓冲区有接收到数据，这时候，就会通知程序有数据可以读了，然后这里就会触发read事件，然后handler就使用channel去读取数据，假设这里在handler里面只读了1半数据，然后就不读了，就是说还有数据没有读，但是这个时候，就去处理下1个selectionKey，那么当调用下1次selector.select方法时，仍然会由于有数据要读取，而被唤醒，仍然是对应该channel的selectionKey的可读事件。// 可写：当客户端或服务端需要将数据发送出去，这时候，需要订阅可写事件，当发送缓冲区可写时，就会触发这个事件，然后使用channel将数据写出到缓冲区，假设1次没写完，那就需要继续订阅可写事件，直到全部数据写完，然后数据全部写完之后，取消订阅可写事件，然后又有数据需要发送，就再次订阅可写事件，写完之后，就取消订阅可写事件。// 可写事件if ((readyOps & SelectionKey.OP_WRITE) != 0) {ch.unsafe().forceFlush();}// 可读 或 可接入事件if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {// （这个方法同时处理 可接入 和 可读 事件，因为如果是NioServerSocketChannel它感兴趣的是OP_ACCEPT事件，而如果是NioSocketChannel它感兴趣的是OP_READ事件，对应的unsafe是不一样的。）// 如果是可接入 AbstractNioMessageChannel.NioMessageUnsafe#read// 如果是可读   AbstractNioByteChannel.NioByteUnsafe#readunsafe.read();}} catch (CancelledKeyException ignored) {unsafe.close(unsafe.voidPromise());}
}

1.3 accept 剖析

其中，前面3步，在NioEventLoop#processSelectedKey中已经做过分析了。

nio 中如下代码，在 netty 中的流程

//1 阻塞直到事件发生
selector.select();Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
while (iter.hasNext()) {    //2 拿到一个事件SelectionKey key = iter.next();//3 如果是 accept 事件if (key.isAcceptable()) {//4 执行 acceptSocketChannel channel = serverSocketChannel.accept();channel.configureBlocking(false);//5 关注 read 事件channel.register(selector, SelectionKey.OP_READ);}// ...
}

入口代码

// 服务端
public class TestSourceServer {public static void main(String[] args) {new ServerBootstrap().group(new NioEventLoopGroup()).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) {ch.pipeline().addLast(new LoggingHandler());}}).bind(8888);}
}

// 客户端
public class TestSourceClient {public static void main(String[] args) {new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) {ch.pipeline().addLast(new LoggingHandler());}}).connect(new InetSocketAddress("localhost", 8888));}}

AbstractNioMessageChannel.NioMessageUnsafe#read

先来看可接入事件处理（accept）

io.netty.channel.nio.AbstractNioMessageChannel.NioMessageUnsafe#read

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 {// doReadMessages中执行了accept并创建【NioSocketChannel】作为消息放入readBuf// readBuf 是一个 ArrayList 用来缓存消息// （看下面的doReadMessages方法）int localRead = doReadMessages(readBuf);if (localRead == 0) {break;}if (localRead < 0) {closed = true;break;}// localRead 为 1，就一条消息，即接收一个客户端连接allocHandle.incMessagesRead(localRead);} while (allocHandle.continueReading());} catch (Throwable t) {exception = t; // 忽略暂时的异常}int size = readBuf.size();for (int i = 0; i < size; i ++) {readPending = false;// 触发 read 事件，让NioServerSocketChannel的pipelin上的 handler 处理，// 这时 肯定交给 ServerBootstrapAcceptor#channelReadpipeline.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();}}
}@Override
protected int doReadMessages(List<Object> buf) throws Exception {// 获取到SocketChannelSocketChannel ch = SocketUtils.accept(javaChannel());try {if (ch != null) {// 创建NioSocketChannelbuf.add(new NioSocketChannel(this, ch));return 1;}} catch (Throwable t) {logger.warn("Failed to create a new channel from an accepted socket.", t);try {ch.close();} catch (Throwable t2) {logger.warn("Failed to close a socket.", t2);}}return 0;
}

ServerBootstrapAcceptor#channelRead

关键代码 io.netty.bootstrap.ServerBootstrap.ServerBootstrapAcceptor#channelRead

public void channelRead(ChannelHandlerContext ctx, Object msg) {// 这时的 msg 是 NioSocketChannelfinal Channel child = (Channel) msg;// NioSocketChannel 添加  childHandler 即初始化器// （这里添加的是初始化器）child.pipeline().addLast(childHandler);// 设置选项setChannelOptions(child, childOptions, logger);for (Entry<AttributeKey<?>, Object> e: childAttrs) {child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());}try {// 注册 NioSocketChannel 到 nio worker 线程，接下来的处理也移交至 nio worker 线程childGroup.register(child).addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture future) throws Exception {if (!future.isSuccess()) {forceClose(child, future.cause());}}});} catch (Throwable t) {forceClose(child, t);}
}

AbstractChannel.AbstractUnsafe#register

又回到了熟悉的 io.netty.channel.AbstractChannel.AbstractUnsafe#register 方法

public final void register(EventLoop eventLoop, final ChannelPromise promise) {// 一些检查，略...AbstractChannel.this.eventLoop = eventLoop;if (eventLoop.inEventLoop()) {register0(promise);} else {try {// 这行代码完成的事实是 nio boss -> nio worker 线程的切换！！！eventLoop.execute(new Runnable() {@Overridepublic void run() {// 调用注册的方法register0(promise);}});} catch (Throwable t) {// 日志记录...closeForcibly();closeFuture.setClosed();safeSetFailure(promise, t);}}
}

*AbstractUnsafe#register0

io.netty.channel.AbstractChannel.AbstractUnsafe#register0

private void register0(ChannelPromise promise) {try {if (!promise.setUncancellable() || !ensureOpen(promise)) {return;}boolean firstRegistration = neverRegistered;// 这里面在 AbstractNioChannel 的 doRegister()方法中会将channel注册到selector上doRegister();neverRegistered = false;registered = true;//【关键代码，注意初始化器执行前后。这里将会为NiosocketChannel添加自定义的handler。】// 执行初始化器，执行前 pipeline 中只有 head -> 初始化器 -> tailpipeline.invokeHandlerAddedIfNeeded();// 执行后就是 head -> logging handler -> my handler -> tail// 上面将客户端的channel已经配置好了，所以通知promise已经成功设置了safeSetSuccess(promise);// 触发handler的 channelRegistered事件pipeline.fireChannelRegistered();if (isActive()) {if (firstRegistration) {// 触发 pipeline 上 active 事件// （这里就会在HeadContext#channelActive中让channel关注可读事件）pipeline.fireChannelActive();} else if (config().isAutoRead()) {beginRead();}}} catch (Throwable t) {closeForcibly();closeFuture.setClosed();safeSetFailure(promise, t);}
}

- HeadContext#channelActive

回到了熟悉的代码 io.netty.channel.DefaultChannelPipeline.HeadContext#channelActive

public void channelActive(ChannelHandlerContext ctx) {ctx.fireChannelActive();// 触发read (NioSocketChannel 这里 read，只是为了触发 channel 的事件注册，还未涉及数据读取)// （注册可读事件）readIfIsAutoRead();
}private void readIfIsAutoRead() {if (channel.config().isAutoRead()) {// 进入该调用，经过pipeline逐链调用，会来到HeadContext的read方法channel.read();}
}@Override
public void read(ChannelHandlerContext ctx) {// 接着这里调用到了AbstractNioChannel#doBeginReadunsafe.beginRead();
}

–AbstractNioChannel#doBeginRead

io.netty.channel.nio.AbstractNioChannel#doBeginRead

protected void doBeginRead() throws Exception {// Channel.read() or ChannelHandlerContext.read() was calledfinal SelectionKey selectionKey = this.selectionKey;if (!selectionKey.isValid()) {return;}readPending = true;// 这时候 interestOps 是 0final int interestOps = selectionKey.interestOps();if ((interestOps & readInterestOp) == 0) {// 关注 read 事件！！！selectionKey.interestOps(interestOps | readInterestOp);}
}

1.4 read 剖析

接着NioEventLoop#processSelectedKey的那节，当对方发送消息来时。

AbstractNioByteChannel.NioByteUnsafe#read

再来看可读事件 io.netty.channel.nio.AbstractNioByteChannel.NioByteUnsafe#read，注意发送的数据未必能够一次读完，因此会触发多次 nio read 事件，一次事件内会触发多次 pipeline read，一次事件会触发一次 pipeline read complete

public final void read() {final ChannelConfig config = config();if (shouldBreakReadReady(config)) {clearReadPending();return;}final ChannelPipeline pipeline = pipeline();// io.netty.allocator.type 决定 allocator 的实现final ByteBufAllocator allocator = config.getAllocator();// 用来分配 byteBuf，确定单次读取大小final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();allocHandle.reset(config);ByteBuf byteBuf = null;boolean close = false;try {do {byteBuf = allocHandle.allocate(allocator);// 上面还是空的byteBuf// 读取（这里就会将缓冲区中的数据读取到byteBuf中，//      调用NioSocketChannel#doReadBytes）allocHandle.lastBytesRead(doReadBytes(byteBuf));if (allocHandle.lastBytesRead() <= 0) {byteBuf.release();byteBuf = null;close = allocHandle.lastBytesRead() < 0;if (close) {readPending = false;}break;}allocHandle.incMessagesRead(1);readPending = false;// 触发read事件,让pipeline上的handler处理,这时是处理 NioSocketChannel上的 handlerpipeline.fireChannelRead(byteBuf);byteBuf = null;} // 是否要继续循环while (allocHandle.continueReading());allocHandle.readComplete();// 触发 read complete 事件pipeline.fireChannelReadComplete();if (close) {closeOnRead(pipeline);}} catch (Throwable t) {handleReadException(pipeline, byteBuf, t, close, allocHandle);} finally {if (!readPending && !config.isAutoRead()) {removeReadOp();}}
}

NioSocketChannel#doReadBytes

protected int doReadBytes(ByteBuf byteBuf) throws Exception {final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();allocHandle.attemptedBytesRead(byteBuf.writableBytes());// 读取数据return byteBuf.writeBytes(javaChannel(), allocHandle.attemptedBytesRead());
}

MaxMessageHandle#continueReading

io.netty.channel.DefaultMaxMessagesRecvByteBufAllocator.MaxMessageHandle#continueReading(io.netty.util.UncheckedBooleanSupplier)

public boolean continueReading(UncheckedBooleanSupplier maybeMoreDataSupplier) {return // 一般为 trueconfig.isAutoRead() &&// respectMaybeMoreData 默认为 true// maybeMoreDataSupplier 的逻辑是如果预期读取字节与实际读取字节相等，返回 true(!respectMaybeMoreData || maybeMoreDataSupplier.get()) &&// 小于最大次数，maxMessagePerRead 默认 16totalMessages < maxMessagePerRead &&// 实际读到了数据totalBytesRead > 0;
}

1.5 write剖析

public class TestSourceServer {public static void main(String[] args) {new ServerBootstrap().group(new NioEventLoopGroup()).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) {ch.pipeline().addLast(new LoggingHandler());}}).bind(8888);}
}

public class TestSourceClient {public static void main(String[] args) {new Bootstrap().group(new NioEventLoopGroup()).channel(NioSocketChannel.class).handler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) {ch.pipeline().addLast(new LoggingHandler());ch.pipeline().addLast(new ChannelDuplexHandler() {@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {ctx.channel().write("halo");}});}}).connect(new InetSocketAddress("localhost", 8888));}}

HeadContext

public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {unsafe.write(msg, promise);
}

write：写队列

我们来看看channel中unsafe的write方法，先来看看其中的一个属性

AbstractUnsafe

protected abstract class AbstractUnsafe implements Unsafe {private volatile ChannelOutboundBuffer outboundBuffer = new ChannelOutboundBuffer(AbstractChannel.this);

我们来看看 ChannelOutboundBuffer 这个类

public final class ChannelOutboundBuffer {private final Channel channel;private ChannelOutboundBuffer.Entry flushedEntry;private ChannelOutboundBuffer.Entry unflushedEntry;private ChannelOutboundBuffer.Entry tailEntry;

ChannelOutboundBuffer内部维护了一个Entry链表，并使用Entry封装msg。其中的属性我们下面会详细讲

我们回到正题，接着看 unsafe.write(msg, promise);

AbstractUnsafe

@Override
public final void write(Object msg, ChannelPromise promise) {assertEventLoop();ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;int size;try {msg = filterOutboundMessage(msg);size = pipeline.estimatorHandle().size(msg);if (size < 0) {size = 0;}} catch (Throwable t) {safeSetFailure(promise, t);ReferenceCountUtil.release(msg);return;}outboundBuffer.addMessage(msg, size, promise);
}

1.调用 filterOutboundMessage() 方法，将待写入的对象过滤，把非ByteBuf对象和FileRegion过滤，把所有的非直接内存转换成直接内存DirectBuffer

@Override
protected final Object filterOutboundMessage(Object msg) {if (msg instanceof ByteBuf) {ByteBuf buf = (ByteBuf) msg;if (buf.isDirect()) {return msg;}return newDirectBuffer(buf);}if (msg instanceof FileRegion) {return msg;}throw new UnsupportedOperationException("unsupported message type: " + StringUtil.simpleClassName(msg) + EXPECTED_TYPES);
}

2.接下来，估算出需要写入的ByteBuf的size
3.最后，调用 ChannelOutboundBuffer 的addMessage(msg, size, promise) 方法，所以，接下来，我们需要重点看一下这个方法干了什么事情

ChannelOutboundBuffer

public void addMessage(Object msg, int size, ChannelPromise promise) {// 创建一个待写出的消息节点Entry entry = Entry.newInstance(msg, size, total(msg), promise);if (tailEntry == null) {flushedEntry = null;tailEntry = entry;} else {Entry tail = tailEntry;tail.next = entry;tailEntry = entry;}if (unflushedEntry == null) {unflushedEntry = entry;}incrementPendingOutboundBytes(size, false);
}

想要理解上面这段代码，必须得掌握写缓存中的几个消息指针，如下图

hannelOutboundBuffer 里面的数据结构是一个单链表结构，每个节点是一个 Entry，Entry 里面包含了待写出ByteBuf 以及消息回调 promise，下面分别是三个指针的作用

1.flushedEntry 指针表示第一个被写到操作系统Socket缓冲区中的节点
2.unFlushedEntry 指针表示第一个未被写入到操作系统Socket缓冲区中的节点
3.tailEntry指针表示ChannelOutboundBuffer缓冲区的最后一个节点

初次调用 addMessage 之后，各个指针的情况为

fushedEntry指向空，unFushedEntry和 tailEntry 都指向新加入的节点

第二次调用 addMessage之后，各个指针的情况为

第n次调用 addMessage之后，各个指针的情况为

可以看到，调用n次addMessage，flushedEntry指针一直指向NULL，表示现在还未有节点需要写出到Socket缓冲区，而unFushedEntry之后有n个节点，表示当前还有n个节点尚未写出到Socket缓冲区中去

flush：刷新写队列

不管调用channel.flush()，还是ctx.flush()，最终都会落地到pipeline中的head节点

HeadContext

@Override
public void flush(ChannelHandlerContext ctx) throws Exception {unsafe.flush();
}

之后进入到AbstractUnsafe

AbstractUnsafe

public final void flush() {assertEventLoop();ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;if (outboundBuffer == null) {return;}outboundBuffer.addFlush();flush0();
}

flush方法中，先调用 outboundBuffer.addFlush();

ChannelOutboundBuffer

public void addFlush() {Entry entry = unflushedEntry;if (entry != null) {if (flushedEntry == null) {flushedEntry = entry;}do {flushed ++;if (!entry.promise.setUncancellable()) {int pending = entry.cancel();decrementPendingOutboundBytes(pending, false, true);}entry = entry.next;} while (entry != null);unflushedEntry = null;}
}

可以结合前面的图来看，首先拿到 unflushedEntry 指针，然后将 flushedEntry 指向unflushedEntry所指向的节点，调用完毕之后，三个指针的情况如下所示

相当于所有的节点都即将开始推送出去

接下来，调用 flush0();

AbstractUnsafe

protected void flush0() {doWrite(outboundBuffer);
}

发现这里的核心代码就一个 doWrite，继续跟

AbstractNioByteChannel

protected void doWrite(ChannelOutboundBuffer in) throws Exception {int writeSpinCount = -1;boolean setOpWrite = false;for (;;) {// 拿到第一个需要flush的节点的数据Object msg = in.current();if (msg instanceof ByteBuf) {// 强转为ByteBuf，若发现没有数据可读，直接删除该节点ByteBuf buf = (ByteBuf) msg;boolean done = false;long flushedAmount = 0;// 拿到自旋锁迭代次数if (writeSpinCount == -1) {writeSpinCount = config().getWriteSpinCount();}// 自旋，将当前节点写出for (int i = writeSpinCount - 1; i >= 0; i --) {int localFlushedAmount = doWriteBytes(buf);if (localFlushedAmount == 0) {setOpWrite = true;break;}flushedAmount += localFlushedAmount;if (!buf.isReadable()) {done = true;break;}}in.progress(flushedAmount);// 写完之后，将当前节点删除if (done) {in.remove();} else {break;}} }
}

这里略微有点复杂，我们分析一下

1.第一步，调用current()先拿到第一个需要flush的节点的数据

ChannelOutBoundBuffer

public Object current() {Entry entry = flushedEntry;if (entry == null) {return null;}return entry.msg;
}

2.第二步,拿到自旋锁的迭代次数

if (writeSpinCount == -1) {writeSpinCount = config().getWriteSpinCount();
}

3.自旋的方式将ByteBuf写出到jdk nio的Channel

for (int i = writeSpinCount - 1; i >= 0; i --) {int localFlushedAmount = doWriteBytes(buf);if (localFlushedAmount == 0) {setOpWrite = true;break;}flushedAmount += localFlushedAmount;if (!buf.isReadable()) {done = true;break;}
}

doWriteBytes 方法跟进去

protected int doWriteBytes(ByteBuf buf) throws Exception {final int expectedWrittenBytes = buf.readableBytes();return buf.readBytes(javaChannel(), expectedWrittenBytes);
}

我们发现，出现了 javaChannel()，表明已经进入到了jdk nio Channel的领域，我们来看看 buf.readBytes(javaChannel(), expectedWrittenBytes);

public int readBytes(GatheringByteChannel out, int length) throws IOException {this.checkReadableBytes(length);int readBytes = this.getBytes(this.readerIndex, out, length);this.readerIndex += readBytes;return readBytes;
}

我们来看关键代码 this.getBytes(this.readerIndex, out, length)

private int getBytes(int index, GatheringByteChannel out, int length, boolean internal) throws IOException {this.checkIndex(index, length);if (length == 0) {return 0;} else {ByteBuffer tmpBuf;if (internal) {tmpBuf = this.internalNioBuffer();} else {tmpBuf = ((ByteBuffer)this.memory).duplicate();}index = this.idx(index);tmpBuf.clear().position(index).limit(index + length);//将tmpBuf中的数据写到out中return out.write(tmpBuf);}
}

我们来看看out.write(tmpBuf)

public int write(ByteBuffer src) throws IOException {ensureOpen();if (!writable)throw new NonWritableChannelException();synchronized (positionLock) {int n = 0;int ti = -1;try {begin();ti = threads.add();if (!isOpen())return 0;do {n = IOUtil.write(fd, src, -1, nd);} while ((n == IOStatus.INTERRUPTED) && isOpen());return IOStatus.normalize(n);} finally {threads.remove(ti);end(n > 0);assert IOStatus.check(n);}}
}

和read实现一样，SocketChannelImpl的write方法通过IOUtil的write实现：关键代码 n = IOUtil.write(fd, src, -1, nd);

static int write(FileDescriptor var0, ByteBuffer var1, long var2, NativeDispatcher var4) throws IOException {//如果是DirectBuffer，直接写，将堆外缓存中的数据拷贝到内核缓存中进行发送if (var1 instanceof DirectBuffer) {return writeFromNativeBuffer(var0, var1, var2, var4);} else {//非DirectBuffer//获取已经读取到的位置int var5 = var1.position();//获取可以读到的位置int var6 = var1.limit();assert var5 <= var6;//申请一个原buffer可读大小的DirectByteBufferint var7 = var5 <= var6 ? var6 - var5 : 0;ByteBuffer var8 = Util.getTemporaryDirectBuffer(var7);int var10;try {var8.put(var1);var8.flip();var1.position(var5);//通过DirectBuffer写，将堆外缓存的数据拷贝到内核缓存中进行发送int var9 = writeFromNativeBuffer(var0, var8, var2, var4);if (var9 > 0) {var1.position(var5 + var9);}var10 = var9;} finally {//回收分配的DirectByteBufferUtil.offerFirstTemporaryDirectBuffer(var8);}return var10;}
}

代码逻辑我们就不再讲了，代码注释已经很清楚了，这里我们关注一点，我们可以看看我们前面的一个方法 filterOutboundMessage()，将待写入的对象过滤，把非ByteBuf对象和FileRegion过滤，把所有的非直接内存转换成直接内存DirectBuffer

说明到了这一步所有的 var1 意境是直接内存DirectBuffer，就不需要走到else，就不需要write两次了

4.删除该节点

节点的数据已经写入完毕，接下来就需要删除该节点

ChannelOutBoundBuffer

public boolean remove() {Entry e = flushedEntry;Object msg = e.msg;ChannelPromise promise = e.promise;int size = e.pendingSize;removeEntry(e);if (!e.cancelled) {ReferenceCountUtil.safeRelease(msg);safeSuccess(promise);}// recycle the entrye.recycle();return true;
}

首先拿到当前被flush掉的节点(flushedEntry所指)，然后拿到该节点的回调对象 ChannelPromise, 调用 removeEntry()方法移除该节点

private void removeEntry(Entry e) {if (-- flushed == 0) {flushedEntry = null;if (e == tailEntry) {tailEntry = null;unflushedEntry = null;}} else {flushedEntry = e.next;}
}

这里的remove是逻辑移除，只是将flushedEntry指针移到下个节点，调用完毕之后，节点图示如下

writeAndFlush: 写队列并刷新

理解了write和flush这两个过程，writeAndFlush 也就不难了

public final ChannelFuture writeAndFlush(Object msg) {return tail.writeAndFlush(msg);
}public ChannelFuture writeAndFlush(Object msg) {return writeAndFlush(msg, newPromise());
}public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) {write(msg, true, promise);return promise;
}private void write(Object msg, boolean flush, ChannelPromise promise) {AbstractChannelHandlerContext next = findContextOutbound();EventExecutor executor = next.executor();if (executor.inEventLoop()) {if (flush) {next.invokeWriteAndFlush(m, promise);} else {next.invokeWrite(m, promise);}} 
}

可以看到，最终，通过一个boolean变量，表示是调用 invokeWriteAndFlush，还是 invokeWrite，invokeWrite便是我们上文中的write过程

private void invokeWriteAndFlush(Object msg, ChannelPromise promise) {invokeWrite0(msg, promise);invokeFlush0();
}

可以看到，最终调用的底层方法和单独调用 write 和 flush 是一样的

private void invokeWrite(Object msg, ChannelPromise promise) {invokeWrite0(msg, promise);
}private void invokeFlush(Object msg, ChannelPromise promise) {invokeFlush0(msg, promise);
}

由此看来，invokeWriteAndFlush基本等价于write方法之后再来一次flush