Netty数据如何在 pipeline 中流动

这里，我们的重点其实就是 pipeline.fireChannelRead(byteBuf);

「DefaultChannelPipeline」

final AbstractChannelHandlerContext head;
//...
head = new HeadContext(this);

public final ChannelPipeline fireChannelRead(Object msg) {
    AbstractChannelHandlerContext.invokeChannelRead(head, msg);
    return this;
}

结合这幅图

可以看到，数据从head节点开始流入，在进行下一步之前，我们先把head节点的功能过一遍

「HeadContext」

final class HeadContext extends AbstractChannelHandlerContext
        implements ChannelOutboundHandler, ChannelInboundHandler {

    private final Unsafe unsafe;

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

    @Override
    public ChannelHandler handler() {
        return this;
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        // NOOP
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        // NOOP
    }

    @Override
    public void bind(
            ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise)
            throws Exception {
        unsafe.bind(localAddress, promise);
    }

    @Override
    public void connect(
            ChannelHandlerContext ctx,
            SocketAddress remoteAddress, SocketAddress localAddress,
            ChannelPromise promise) throws Exception {
        unsafe.connect(remoteAddress, localAddress, promise);
    }

    @Override
    public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        unsafe.disconnect(promise);
    }

    @Override
    public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        unsafe.close(promise);
    }

    @Override
    public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        unsafe.deregister(promise);
    }

    @Override
    public void read(ChannelHandlerContext ctx) {
        unsafe.beginRead();
    }

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

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

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.fireExceptionCaught(cause);
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        invokeHandlerAddedIfNeeded();
        ctx.fireChannelRegistered();
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelUnregistered();

        // Remove all handlers sequentially if channel is closed and unregistered.
        if (!channel.isOpen()) {
            destroy();
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelActive();

        readIfIsAutoRead();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelInactive();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ctx.fireChannelRead(msg);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelReadComplete();

        readIfIsAutoRead();
    }

    private void readIfIsAutoRead() {
        if (channel.config().isAutoRead()) {
            channel.read();
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        ctx.fireUserEventTriggered(evt);
    }

    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelWritabilityChanged();
    }
}

从head节点继承的两个接口看，TA既是一个ChannelHandlerContext，同时又属于inBound和outBound Handler

在传播读写事件的时候，head的功能只是简单地将事件传播下去，如ctx.fireChannelRead(msg);

在真正执行读写操作的时候，例如在调用writeAndFlush()等方法的时候，最终都会委托到unsafe执行，而当一次数据读完，channelReadComplete方法会被调用

pipeline中的inBound事件传播

我们接着上面的 AbstractChannelHandlerContext.invokeChannelRead(head, msg); 这个静态方法看，参数传入了 head，我们知道入站数据都是从 head 开始的，以保证后面所有的 handler 都由机会处理数据流。我们看看这个静态方法内部是怎么样的：

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
    final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeChannelRead(m);
    } else {
        executor.execute(new Runnable() {
            public void run() {
                next.invokeChannelRead(m);
            }
        });
    }
}

调用这个 Context （也就是 head） 的 invokeChannelRead 方法，并传入数据。我们再看看head中 invokeChannelRead 方法的实现，实际上是在headContext的父类AbstractChannelHandlerContext中：

「AbstractChannelHandlerContext」

private void invokeChannelRead(Object msg) {
    if (invokeHandler()) {
        try {
            ((ChannelInboundHandler) handler()).channelRead(this, msg);
        } catch (Throwable t) {
            notifyHandlerException(t);
        }
    } else {
        fireChannelRead(msg);
    }
}

public ChannelHandler handler() {
    return this;
}

上面 handler()就是headContext中的handler,也就是headContext自身，也就是调用 head 的 channelRead 方法。那么这个方法是怎么实现的呢？

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    ctx.fireChannelRead(msg);
}

什么都没做，调用 Context 的 fire 系列方法，将请求转发给下一个节点。我们这里是 fireChannelRead 方法，注意，这里方法名字都挺像的。需要细心区分。下面我们看看 Context 的成员方法 fireChannelRead：「AbstractChannelHandlerContext」

@Override
public ChannelHandlerContext fireChannelRead(final Object msg) {
    invokeChannelRead(findContextInbound(), msg);
    return this;
}

这个是 head 的抽象父类 AbstractChannelHandlerContext 的实现，该方法再次调用了静态 fire 系列方法，但和上次不同的是，不再放入 head 参数了，而是使用 findContextInbound 方法的返回值。从这个方法的名字可以看出，是找到入站类型的 handler。我们看看方法实现：

private AbstractChannelHandlerContext findContextInbound() {
    AbstractChannelHandlerContext ctx = this;
    do {
        ctx = ctx.next;
    } while (!ctx.inbound);
    return ctx;
}

该方法很简单，找到当前 Context 的 next 节点（inbound 类型的）并返回。这样就能将请求传递给后面的 inbound handler 了。我们来看看 invokeChannelRead(findContextInbound(), msg);

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
    final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeChannelRead(m);
    } else {
        executor.execute(new Runnable() {
            public void run() {
                next.invokeChannelRead(m);
            }
        });
    }

}

上面我们找到了next节点（inbound类型的），然后直接调用 next.invokeChannelRead(m);如果这个next是我们自定义的handler,此时我们自定义的handler的父类是AbstractChannelHandlerContext，则又回到了AbstractChannelHandlerContext中实现的invokeChannelRead，代码如下：

「AbstractChannelHandlerContext」

private void invokeChannelRead(Object msg) {
    if (invokeHandler()) {
        try {
            ((ChannelInboundHandler) handler()).channelRead(this, msg);
        } catch (Throwable t) {
            notifyHandlerException(t);
        }
    } else {
        fireChannelRead(msg);
    }
}

public ChannelHandler handler() {
    return this;
}

此时的handler()就是我们自定义的handler了，然后调用我们自定义handler中的 channelRead(this, msg);

请求进来时，pipeline 会从 head 节点开始输送，通过配合 invoker 接口的 fire 系列方法，实现 Context 链在 pipeline 中的完美传递。最终到达我们自定义的 handler。

如果所有的handler都调用了fire系列方法，则会传递到最后一个inbound类型的handler，也就是——tail节点，那我们就来看看tail节点

pipeline中的tail

final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {

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

    @Override
    public ChannelHandler handler() {
        return this;
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception { }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception { }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception { }

    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception { }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        // This may not be a configuration error and so don't log anything.
        // The event may be superfluous for the current pipeline configuration.
        ReferenceCountUtil.release(evt);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        onUnhandledInboundException(cause);
    }

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

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { }
}

正如我们前面所提到的，tail节点的大部分作用即终止事件的传播(方法体为空)

「channelRead」

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节点在发现字节数据(ByteBuf)或者decoder之后的业务对象在pipeline流转过程中没有被消费，落到tail节点，tail节点就会给你发出一个警告，告诉你，我已经将你未处理的数据给丢掉了

总结一下，tail节点的作用就是结束事件传播，并且对一些重要的事件做一些善意提醒.

pipeline中的outBound事件传播

上一节中，我们在阐述tail节点的功能时，忽略了其父类AbstractChannelHandlerContext所具有的功能，这一节中，我们以最常见的writeAndFlush操作来看下pipeline中的outBound事件是如何向外传播的

典型的消息推送系统中，会有类似下面的一段代码

Channel channel = getChannel(userInfo);
channel.writeAndFlush(pushInfo);

这段代码的含义就是根据用户信息拿到对应的Channel，然后给用户推送消息，跟进 channel.writeAndFlush

public ChannelFuture writeAndFlush(Object msg) {
    return pipeline.writeAndFlush(msg);
}

从pipeline开始往外传播

public final ChannelFuture writeAndFlush(Object msg) {
    return tail.writeAndFlush(msg);
}

Channel 中大部分outBound事件都是从tail开始往外传播, writeAndFlush()方法是tail继承而来的方法，我们跟进去

「AbstractChannelHandlerContext」

public ChannelFuture writeAndFlush(Object msg) {
    return writeAndFlush(msg, newPromise());
}

public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) {
    write(msg, true, promise);

    return 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()方法找到下一个outBound()节点

「AbstractChannelHandlerContext」

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

找outBound节点的过程和找inBound节点类似，反方向遍历pipeline中的双向链表，直到第一个outBound节点next，然后调用next.invokeWriteAndFlush(m, promise)

「AbstractChannelHandlerContext」

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

调用该节点的ChannelHandler的write方法，flush方法我们暂且忽略，后面会专门讲writeAndFlush的完整流程

「AbstractChannelHandlerContext」

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

可以看到，数据开始出站，从后向前开始流动，和入站的方向是反的。那么最后会走到哪里呢，当然是走到 head 节点，因为 head 节点就是 outbound 类型的 handler。

「HeadContext」

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

调用了 底层的 unsafe 操作数据，这里，加深了我们对head节点的理解，即所有的数据写出都会经过head节点

当执行完这个 write 方法后，方法开始退栈。逐步退到 unsafe 的 read 方法，回到最初开始的地方，然后继续调用 pipeline.fireChannelReadComplete() 方法。

总结

结束