Netty - Series 1

这是一个后续打算进行长期更新的系列，每篇只包含相关的几个知识点，并不保证完全的准确性。那么作为第一篇，以我比较推崇的Netty框架作为开山作，并希望可以给大家带来一些比较少见但是又比较实用的东西。后续也没有什么特别长远的计划，大概率还是围绕着Netty、RocketMQ等市面上比较常见常用的框架来展开，至于说时间上，也没有定下一个时间线，所以也是随缘更新为主。

<1> writeAndFlush

这是Netty中最常见、或者说最经常使用的API。当我们有数据需要通过网络发送出去时，那么我们从各种教程中，都可以看到这个API的身影。

但是，总有一些比较疑惑的问题，比如说我们自定义一个ChannelHandler，就假定它是ChannelInboundHandler，那么简单定义一个：

public class EchoByDifferentWAF extends ChannelInboundHandlerAdapter {
   
复制

可以看到，有两种方式可以调用writeAndFlush，那么它们既然是被分在两个类里的方法去调用，就一定是有区别的，但是这个方法其实是定义在一个顶层接口：ChannelOutboundInvoker里面的，Channel和ChannelHandlerContext只是继承了这个接口的能力而已。所以，它们各自做了不同的实现，而区别可以简化为：

• Channel里实现的writeAndFlush是从当前ChannelPipeline的TailContext往前找，并逐个调用write。

• 而ChannelHandlerContext里实现的writeAndFlush则是从当前ChannelHandlerContext往前找ChannelOutboundHandler，然后逐个调用它们的write方法。

所以，这里也算是一个小坑，在此提一嘴。并且我们可以从这里进行内容的展开。

上面说到，两个类里实现的writeAndFlush导致的【方法入口】的ChannelHandlerContext是不一样的，但是，从源码翻阅，可以发现它们其实最后都是落到一个相同的方法上，因为，Channel里实际调用的是当前pipeline的tail.writeAndFlush，而tail则是pipeline中的TailContext，所以它们最后的执行逻辑都在AbstractChannelHandlerContext里的writeAndFlush方法。看到相关的定义：

abstract class AbstractChannelHandlerContext implements ChannelHandlerContext, ResourceLeakHint {
  	@Override
  	public ChannelFuture writeAndFlush(Object msg) {
      	return writeAndFlush(msg, newPromise());
    }
  	@Override
  	public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) {
      	write(msg, /* boolean flush */true, promise);
      	return promise;
    }
  	// ...
}
复制

可以看到，最后是委托到统一的一个write方法里，通过一个boolean标识位来判断是否立即触发flush。

再看到write方法，大概的伪代码如下：

private void write(Object msg, boolean flush, ChannelPromise promise) {
		checkMsgNotNullAndPromiseValid();
  	ChannelHandlerContext next = findNextOutboundContext();
  	EventExecutor executor = currentChannel's executor;
  	if(executor.inEventLoop) {
      	flush ? next.invokeWriteAndFlush(msg, promise) :
      					next.invokeWrite(msg, promise);
    } else {
      	wrapToAWriteTaskAndSubmitToExecutor();
    }
}
复制

可以看到，整体的结构还是比较简单的。由于讨论的是writeAndFlush，直接看到invokeWriteAndFlush这里，也是定义在AbstractChannelHandlerContext中的方法，结构也是比较简单的：

void invokeWriteAndFlush(Object msg, ChannelPromise promise) {
  	if(invokeHandler()) {
      	invokeWrite0(msg, promise);
      	invokeFlush0();
    } else {
      	writeAndFlush(msg, promise);
    }
}
复制

这里的if是判断当前Channel的ChannelHandler是否已经填充完成，如果没有的话则再次调用writeAndFlush，可以理解成自旋等待Channel就绪。

往下看到两个invoke，可以看到这两个invoke都是定义在当前类或者父类的，那么在这里这两个invoke方法都是在AbstractChannelHandlerContext里的，逐个来看。

<1.1> invokeWrite0

在Netty中，很爱通过在API名字上加个0来代表真正的逻辑，而在AbstractChannelHandlerContext中，所有的invokeXXX0方法，其实内部结构可以说是一模一样的，伪代码如下：

private void invokeWrite0(Object msg, ChannelPromise promise) {
  	ChannelHandler current = handler();
  	HeadContext head = currentPipeline().head;
  	if(head == current) {
      	head.write(this, msg, promise);
    } else if(current is instance of ChannelDuplexHandler or ChannelOutboundHandler) {
      	((ChannelDuplexHandler or ChannelOutboundHandler)current).write(this. msg, promise);
		}
}
复制

非常的简单，就是调用当前这个ChannelHandler的write方法，这里关联的就是可能存在的用户自定义的ChannelOutboundHandler，那么我们不关心这个，重点在于最后会落到的head.write方法这里。

如果我们点进去，会看到只是简单的调用了unsafe.write方法。这里也需要提一下，在Netty中，每一个Channel都有对应的Unsafe，这个Unsafe是Netty自己定义的Channel的内部接口，主要用来完成一些实际的IO操作。

那么找到我们需要的Unsafe实现：AbstractUnsafe，找到它的write方法，如下：

public final void write(Object msg, ChannelPromise promise) {
  	ChannelOutboundBuffer currentBuffer = this.outboundBuffer;
  	if currentBuffer is null: release buffer and set promise to failure state then return;
  	msg = filter(msg);
  	int size = estimateMsgSize(msg).setToZeroIfLessThanZero();
  	outboundBuffer.addMessage(msg, size, promise);
}
复制

结构也是比较简单的，经过前置的一些检测，引出一个新的类ChannelOutboundBuffer，由于篇幅的问题，不展开这个类，只要知道它的关键数据结构是一个链表即可；链表里的每个元素（Entry）都是一个传进来的msg的封装，并将链表分成了两个部分：flushed和unflushed，也是比较好理解的。

那么这里的addMessage，显而易见就是封装成一个Entry添加到链表中。在多次调用后，链表的结构大概如下图：

那么在这里，invokeWrite0方法也就结束了。这里我还想再提一个点，就是在前面的AbstractChannelHandlerContext的write方法的伪代码中提到的==wrapToAWriteTaskAndSubmitToExecutor==这一步。

这里其实也是Netty异步设计的一个关键所在。并且需要提到的一点是，在Netty中，每个Channel都有对应的一个EventLoop线程，并且在每个Channel初始化的时候会将这个Channel和某一个EventLoop给绑定起来。这是在很多八股文中可以知道的事情，但是，注意到这里的调用对象为==handler.executor()==，那么注意到，在我们对启动类进行配置，对ChannelPipeline中填充自定义的ChannelHandler的时候，实际上是可以为不同的ChannelHandler单独配置线程池的，然后这些ChannelHandler就可以使用区别于IO线程（也就是EventLoop）之外的线程来进行业务处理。那么在此时，走到上面的==AbstractChannelHandlerContext.write()==方法的时候，走到==executor.inEventLoop==就会为false，也就会被封装成一个异步任务，等到绑定的线程在运行的时候再去执行操作。

所以，在我看来，这里就是Netty异步实现writeAndFlush的关键。Netty也是通过这种机制来实现所有的IO操作都是由IO线程也就是EventLoop来完成的。

<1.2> invokeFlush0

那么，在invokeWrite0将数据写入到之后，就到了invokeFlush0，方法的内部结构和invokeWrite0很相似，只是调用的是ChannelOutboundInvoker的flush方法而已。最后一个会被调用的仍然是HeadContext.flush方法，也依然是委托给了它对应的Unsafe去实现真正的调用，具体方法如下：

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

protected void flush0() {
  	// 非volatile变量，因为走到这里的时候只有一个线程在执行，所以不会有多线程竞争的情况
  	if(inFlush0) /* avoid reentrance*/ return;
  	ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
  	if(outboundBuffer is null or empty) return;
  	inFlush0 = true;
  	if(channel is not active) mark all pending write reqeusts are failure;
  	doWrite(outboundBuffer);
  	inFlush0 = false;
}
复制

并不复杂，所以具体的flush0也被我贴了出来，流程也是很显而易见的。需要注意的是在flush中的outboundBuffer.addFlush方法，它只是去将对应OutboundBuffer中的已添加并标记为unflushed的所有Entry从头到尾标记成flushed。并将这些Entry给移动到flushedEntry下，如下图：

这个方法就结束后，就会去到flush0，关键就在doWrite(outboundBuffer)这个方法，这是一个定义在AbstractChannel中的抽象方法，我们只关注最常用的NioSocketChannel的实现即可，位于AbstractNioByteChannel中：

@Override protected void doWrite(ChannelOutboundBuffer in) throws Exception {
  	int writeSpinCount = config().getWRiteSpinCount();
  	do {
      	Object msg = in.current();
      	if(msg == null) {
          	clearOpWrite();
          	return;
        }
      	writeSpinCount -= doWriteInternal(in, msg);
		} while (writeSpinCount > 0);
  	incompleteWrite(writeSpinCount < 0);
}
复制

可以看到也是比较简单的。关键方法就在于doWriteInternal这个方法。这个方法的返回值有1、0以及一个Integer.MAX_VALUE，可以简单的认为

• 0：实际上没有数据写出
• 1：实际上有数据写出
• Integer.MAX_VALUE：Socket缓冲区满了，说明通道暂时不可写

所以，当通道不可写的时候。会立刻跳出do-while循环，后面再展开。

最终会委托到Java网络框架的实现上去执行，以doWriteBytes进行了一个方法的抽象。以最常用的Nio为例，则是NioSocketChannel去执行。

顺便一提，这里的writeSpinCount默认是16，也就是说最多执行16次write操作，如果16次之内写完了，那么会触发clearOpWrite方法，其实就是将对应Channel的SelectionKey中的OP_WRITE事件给清除掉，相当于做一次保险，避免下一次EventLoop运行的时候去关注这个OP_WRITE事件。

那么，为什么这里的incompleteWrite还需要对writeSpinCount的大小进行判断呢？就在于这里涉及到了通道不可写的这种情况。前面提到，当通道不可写时，doWriteInternal会返回Integer.MAX_VALUE，此时writeSpinCount为负数，那么判断结果为true，具体方法如下：

protected final void incompleteWrite(boolean setOpWrite) {
  	if(setOpWrite) setOpWrite();
  	else {
      	clearOpWrite();
      	eventLoop().execute(flushTask);
    }
}
复制

所以，分情况讨论：

• 通道可写，但是已经写了16次到达上限（setOpWrite = false）：

  由于IO线程也就是EventLoop负责管理多个Channel，当单个Channel多次执行耗时的IO操作时，避免长时间占用线程，所以此时该Channel需要让出线程。此时通道仍然是可写的，所以EventLoop不需要关注这个Channel的OP_WRITE事件，需要clear掉，而由于有多出来的数据没有来得及写入到Socket缓冲，所以提交一个异步任务，等到EventLoop空闲的时候就去继续执行flush。如果我们去看flushTask，其实就是执行的flush0，验证了我们的这一说法。

• 通道不可写（setOpWrite = true）：

  通道不可写，则向Selector注册OP_WRITE事件，当Socket缓冲区可写时，说明通道可写，此时EventLoop线程会优先处理OP_WRITE事件而不是OP_READ事件，这就是另外的内容，暂时不多提。

至此，invokeFlush0也算是完成了它的工作，其实这其中还涉及到了Netty通道的高低水位机制，由于不想一次性说的太多，所以没有提及，感兴趣的可以去看看。并且，在Netty实际上读取数据时，和这里的逻辑也是有些许差别的，这些就当作一个坑以后再填。

<2> Inbound和Outbound

先叠个Buff，对于这一节的内容，我是以一个学习者的身份提出这个问题，并结合一些我的见解给读者进行简单的介绍。不能保证内容的准确性，但是考虑到这一个问题几乎没有被提及过，所以写出这一节内容，更多的还是建议大家去自己进行学习。

当我们初次接触Netty的时候，从各类资料中能学习到，最简单的使用就是简单配置Bootstrap/ServerBootstrap，然后在ChannelPipeline中配置自己所需要的ChannelHandler，并且我们也很容易知道，ChannelHandler可以分为ChannelInboundHandler、ChannelOutboundHandler，而从这个名字，我们总是会认为，Inbound、Outbound代表的是应用的数据流向。但是，偶然间在stackoverflow中查找到了一个比较有意思的问题，原帖链接如下：

netty - In Netty4,why read and write both in OutboundHandler - Stack Overflow

这里提到，在Netty4.X版本中，ChannelOutboundHandler接口中同时定义了read、write方法，那么就会感到非常的奇怪，在我们的理解中，outbound应该是只会有write方法的，因为outbound代表了从应用到外界的数据流向，这里不应该有read这个方法。

但是在这个帖子中，作者提到，inbound、outbound并不是依据数据流向来决定的，而是：

• inbound：handle events triggered by external stimuli

  处理外部因素触发的事件

• outbound：intercepts the operations issued by your application

  拦截用户应用程序发起的操作

乍一看，似乎和我们一开始的数据流向based非常的相似。但是，如果我们看到它们各自的API，我们可以发现，在ChannelOutboundHandler中，完全是以==主动方==发起的API，而ChannelInboundHandler中，则完全是以==监听者==监听后的回调。

可以注意到，它们其实关注的点并不是一样的。并且经过debug发现其实是在HeadContext中的channelActive和channelReadComplete回调触发时会调用到ChannelOutboundHandler里的read方法。

触发read方法大致的流程各自如下，都以NIO为例：

• channelActive：

  不论是Server（ServerBootstrap）还是Client端（也就是Bootstrap），最终在启动时都会走到AbstractBootstrap中的initAndRegister这个方法。在这里会执行一系列的操作，往下debug后，最终会触发到ChannelPipeline上的fireChannelActive方法，那么也就是在这里，最后会到上面说的HeadContext中的channelActive方法，触发read。

• channelReadComplete：

  和channelActive的机制相似，只不过这里的入口在于NioEventLoop中，监听到OP_READ事件后会触发unsafe.read方法，不论是服务端的Channel（NioServerSocketChannel）还是普通的Channel（NioSocketChannel），它们各自的Unsafe的实现类最终都会触发ChannelPipeline的fireChannelReadComplete，触发read。

  这里还可以提一嘴的是，NioServerSocketChannel和NioSocketChannel各自的Unsafe实现中的read方法，对于channelRead的fire时机也是不同的。

  • NioServerSocketChannel：将缓冲区的数据读取完之后，逐个触发对应channel的channelRead
  • NioSocketChannel：每完成一次对缓冲区数据的读取，就触发一次对应channel的channelRead

可以看到，基本上都是以ChannelPipeline中的fireXXX方法作为入口，个人认为这也是一种设计模式——门面模式的体现，也就是将方法进行包装，得益于==Netty数据结构==上设计的好处，实际的处理都被更高层级的类给封装了，这些后续我也会考虑是否有必要做个单独的章节来进行总结。

那么对于这个API的具体的讲解以及涉及到的Netty的一些算是误区的地方，就解释到此。其实到这里，基本上就把我想介绍的这一节的内容都说完了。但是，对于这一部分的知识，我更多知识想做个分享，等待其他大佬的解答。而这里，其实我个人还是有点不太能完全理解作者的意图，因为在我看来，如果应用做出了read的操作，那么我的理解是，每次从socket中读取数据都要触发一次ChannelOutboundHandler的read方法，但是实际上并不是这么个机制，不过在已有的机制上来说，这和ChannelInboundHandler中的channelRead回调是有重复的。

提到这个API，并不是说这里埋了个多深奥的技术，更多的是我个人对这个API的设计感到有些疑惑。总体来说，作者的解释个人感觉是能理解，但是在实际使用上来说，个人感觉还是存在一定的迷惑性的。而对此，我不会说这个是Netty整体设计的失误，更多的，我认为是命名上可能存在一些难以让人理解的错误。