2021SC@SDUSC

下面分析一下出站数据传播的细节。我们从ChannelOutboundHandlerAdapter的write方法开始分析：

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

如果想要实现自己的业务处理逻辑，需要继承ChannelOutboundHandlerAdapter并且重写write方法。处理完数据之后，还需要继续向下传递数据，也就是需要调用ctx.write(msg, promise)方法，它会调用AbstractChannelHandlerContext的等效方法，下面是一个它调用链路上的方法：

    private void write(Object msg, boolean flush, ChannelPromise promise) {
       AbstractChannelHandlerContext next = findContextOutbound();
       final Object m = pipeline.touch(msg, next);
       EventExecutor executor = next.executor();
       if (executor.inEventLoop()) {
           if (flush) {
               next.invokeWriteAndFlush(m, promise);
           } else {
               next.invokeWrite(m, promise);
           }
       } else {
           AbstractWriteTask task;
           if (flush) {
               task = WriteAndFlushTask.newInstance(next, m, promise);
           }  else {
               task = WriteTask.newInstance(next, m, promise);
           }
           safeExecute(executor, task, promise, m);
       }
   }

 

这里最关键的是通过findContextOutbound方法来寻找下一个符合要求的handler：

   private AbstractChannelHandlerContext findContextOutbound() {
       AbstractChannelHandlerContext ctx = this;
       do {
           ctx = ctx.prev;
       } while (!ctx.outbound);
       return ctx;
   }

这里从当前的handler,顺着prev往回找，找下一个outband为真的handler。看到这个方法可能会觉得很眼熟，因为和上一次博客中分析的findContextInbound(MASK_CHANNEL_READ)方法很相似，这个方法在入站事件传播时用于寻找下一个符合要求的ChannelHandler：

    private AbstractChannelHandlerContext findContextInbound(int mask) {
        AbstractChannelHandlerContext ctx = this;
        EventExecutor currentExecutor = executor();
        do {
            ctx = ctx.next;
        } while (skipContext(ctx, currentExecutor, mask, MASK_ONLY_INBOUND));
        return ctx;
    }

其实write事件原理和上次分析的read事件类似，区别就是方向和目标不同。入站事件的方向是next方向，目标是InboundHandler。出站事件的方向与入站事件相反，是prev方向，也就是往回传播，目标是OutboundHandler。所以从代码中我们可以看到，这个寻找是顺着prev的方向往回找，也就是出站事件的传播方向为从后往前。

到这里我们明白了ChannelPipeline的事件传播方向。现在来考虑对于入站数据，如果传递到了tail节点，到头了，会有怎样的处理？对于出站数据，如果传递到了head节点，会有怎样的处理？

来看一下 invokeWriteAndFlush方法，它有一个判断 invokeHandler()方法，下面是它的细节：

    private boolean invokeHandler() {
       int handlerState = this.handlerState;
       return handlerState == ADD_COMPLETE || (!ordered && handlerState == ADD_PENDING);
   }

我们重点关注如果 handlerState == ADD_COMPLETE这个条件成立的时候，会返回true，那么就会执行下面的方法：

           invokeWrite0(msg, promise);


  

private void invokeWrite0(Object msg, ChannelPromise promise) {
       try {
           ((ChannelOutboundHandler) handler()).write(this, msg, promise);
       } catch (Throwable t) {
           notifyOutboundHandlerException(t, promise);
       }
   }

我们先来考虑这个判断条件什么时候会成立。这里需要回到我们 HeadContext和TailContext的构造函数上：

        HeadContext(DefaultChannelPipeline pipeline) {
           super(pipeline, null, HEAD_NAME, false, true);
           unsafe = pipeline.channel().unsafe();
           setAddComplete();

       }
       
       TailContext(DefaultChannelPipeline pipeline) {
           super(pipeline, null, TAIL_NAME, true, false);
           setAddComplete();

       }        

    final void setAddComplete() {
       for (;;) {
           int oldState = handlerState;
           // Ensure we never update when the handlerState is REMOVE_COMPLETE already.
           // oldState is usually ADD_PENDING but can also be REMOVE_COMPLETE when an EventExecutor is used that is not
           // exposing ordering guarantees.
           if (oldState == REMOVE_COMPLETE || HANDLER_STATE_UPDATER.compareAndSet(this, oldState, ADD_COMPLETE)) {
               return;
           }
       }
   }

这里我们可以看到两个构造方法后面都会调用setAddComplete()方法，这个方法的代码我直接贴在后面了。这里表达的意思是进入一个无限循环，我们不会改变更新，直到oldState的状态改变为REMOVE_COMPLETE。然后会执行 HANDLER_STATE_UPDATER.compareAndSet(this, oldState, ADD_COMPLETE)这个原子操作，将handlerState设置为ADD_COMPLETE。这样上面函数的成立条件就解释清楚了。

然后我们来看之后执行的操作：

​

private void invokeWrite0(Object msg, ChannelPromise promise) {
       try {
           ((ChannelOutboundHandler) handler()).write(this, msg, promise);
       } catch (Throwable t) {
           notifyOutboundHandlerException(t, promise);
       }
   }

​

我们先看handler()这个方法，这个方法是返回当前context。也就是说，会执行TailContext的等效方法。比如read，就会执行TailContext的channelRead方法：

        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
           onUnhandledInboundMessage(msg);
       }

    protected void onUnhandledInboundMessage(Object msg) {
       try {
           logger.debug(
                   "Discarded inbound message {} that reached at the tail of the pipeline. " +
                           "Please check your pipeline configuration.", msg);
       } finally {
           ReferenceCountUtil.release(msg);
       }
   }

可以看到消息传递到tail之后，就不会再传递下去了，并且会释放它。出站事件也是类似，当事件传递到head的时候，会调用HeadContext的等效方法，然后就不会再传递下去了。