vlambda博客
学习文章列表

DefaultChannelPipeline#fireChannelRead(Object msg) {
public final ChannelPipeline fireChannelRead(Object msg) {
     AbstractChannelHandlerContext.invokeChannelRead(head, msg);  // @1
        return this;
 }

首先在NIO事件选择器在网络读事件就绪后，会调用底层SocketChanel#read 方法从读缓存中读取字节，在Netty中使用ByteBuf来存储，然后调用DefaultChannelPipeline # fireChannelRead 方法进行事件传播，每个ChannelHandler针对输入进行加工处理，ChannelPipeline因此而得名，有关Netty基于NIO的事件就绪选择实现将在Netty线程模型、IO读写流程部分详细讲解。

从代码@1处可得知，通过AbstractChannelHandlerContext的静态方法invokerChanelRead，从HeadContext处开始执行，

AbstractChannelHandlerContext#invokerChanelRead

static void invokeChannelRead(final AbstractChannelHandlerContext next, final Object msg) {
        ObjectUtil.checkNotNull(msg, "msg");
        EventExecutor executor = next.executor();         // @1
        if (executor.inEventLoop()) {                                // @2
            next.invokeChannelRead(msg);
        } else {
            executor.execute(new Runnable() {              // @3
                @Override
                public void run() {
                    next.invokeChannelRead(msg);
                }
            });
        }
    }

这种写法是Netty处理事件执行的“模板”方法，都是先获取需要执行的线程组(EventLoop),如果当前线程不属于Eventloop，则将任务提交到EventLoop中异步执行，如果在，则直接调用。第一次调用，该next指针为HeadContext，那接下来重点关注一下HeadContext的invokeChannelRead方法。
AbstractChannelHandlerContext#invokeChannelRead

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

代码@1：如果该通道已经成功添加@1，则执行对应的事件@2，否则只是传播事件@3。

传播事件在AbstractChannelHandlerContext的实现思路如下：

AbstractChannelHandlerContext#fireChannelRead
public ChannelHandlerContext fireChannelRead(final Object msg) {
        invokeChannelRead(findContextInbound(), msg);
        return this;
}
private AbstractChannelHandlerContext findContextInbound() {
        AbstractChannelHandlerContext ctx = this;
        do {
            ctx = ctx.next;
        } while (!ctx.inbound);
        return ctx;
 }

上述就从事件链中按顺序提取inbound类型的处理器，上述代码要最终能结束，那么TailContext必须是Inbound类型的事件处理器。

从代码@2中执行完对应的事件处理逻辑后，事件如何向下传播呢？如果需要继续将事件传播的话，请调用ChannelInboundHandlerAdapter 对应的传播事件方法，如上例中的 ChannelInboundHandlerAdapter#fireChannelRead，该方法会将事件链继续往下传播，如果在对应的事件处理中继续调用fireChannelRead，则事件传播则停止传播，也就是并不是事件一定会顺着整个调用链到达事件链的尾部TailContext，在实践中请特别重视。

Netty inbound 事件传播流程图如下：

2.3 源码分析DefaultChannelPipeline$HeadContex

2.3.1 HeadContext声明与构造方法

final class HeadContext extends AbstractChannelHandlerContext
            implements ChannelOutboundHandler, ChannelInboundHandler {    // @1

        private final Unsafe unsafe;                                                                     // @2

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

代码@1：HeadContext实现ChannelInboundHandler与ChannelOutboundHandler，故它的inbound与outbound都返回true。
代码@2：Unsafe，Netty操作类。

2.3.2 handlerAdded、handlerRemoved

public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
}
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
}

ChanelHandler增加与移除事件处理逻辑：不做任何处理。为什么可以不传播呢？其实上文在讲解addFirst方法时已提到，在添加一个ChannelHandler到事件链时，会根据通道是否被注册，如果未注册，会先阻塞执行，DefaultChannelPipeline会保存一条执行链，等通道被注册后处触发执行，HeadContext作为一个非业务类型的事件处理器，对通道的增加与否无需关注。

2.3.3 exceptionCaught

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

通道异常处理事件的处理逻辑：HeadContext的选择是自己不关注，直接将异常事件往下传播。

2.3.4 channelRegistered

public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            invokeHandlerAddedIfNeeded();
            ctx.fireChannelRegistered();
}
final void invokeHandlerAddedIfNeeded() {
        assert channel.eventLoop().inEventLoop();
        if (firstRegistration) {
            firstRegistration = false;
            // We are now registered to the EventLoop. It's time to call the callbacks for the ChannelHandlers,
            // that were added before the registration was done.
            callHandlerAddedForAllHandlers();
        }
    }

通道注册事件处理逻辑：当通道成功注册后，判断是否是第一次注册，如果是第一次注册的话，调用所有的ChannelHandler#handlerAdd事件，因为当通道增加到事件链后，如果该通道还未注册，channelAdd事件不会马上执行，需要等通道注册后才执行，故在这里首先需要执行完挂起（延迟等待的任务）。然后调用fireChannelRegistered沿着事件链传播通道注册成功事件。

2.3.5 channelUnregistered

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

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

通道取消注册事件处理逻辑：首先传播事件，然后判断通道的状态，如果是处于关闭状态（通道调用了close方法），则需要移除所有的ChannelHandler。

2.3.6 channelActive

public void channelActive(ChannelHandlerContext ctx) throws Exception {
     ctx.fireChannelActive();
        readIfIsAutoRead();
 }

通道激活事件的处理逻辑（TCP连接建立成功后触发）：首先传播该事件，如果开启自动读机制(autoRead为true)，则调用Channel#read方法，向NIO Selector注册读事件。

2.3.7 channelInactive

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

通道非激活事件处理逻辑：只传播事件。

2.3.8 channelRead

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

通道读事件处理逻辑：向下传播事件，各个编码器、业务处理器将各自处理业务逻辑。

2.3.9 channelReadComplete

public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.fireChannelReadComplete();
readIfIsAutoRead();
}

通道读完成事件，首先先传播事件，然后如果开启了自动读取的话，继续注册读事件。

2.3.10 userEventTriggered

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

用户自定义事件的处理逻辑：传播事件。

2.3.11 channelWritabilityChanged

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

通道可写状态变更事件的处理逻辑：传播事件。

接下来介绍HeadContext对于ChannelOutboundHander事件的处理逻辑：

2.3.12 bind

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

通过Unsafe实例完成具体的绑定操作，后续会重点分析该方法的实现原理。

由于HeadContex是outbound事件的尾部事件处理器，而且outbound是用户发送的API调用，其最终目的是希望通过Netty完成具体的网络操作，故HeadContex是离Netty底层机制最近的，到了这里，就意味者“应用程序”层面的定制化介绍，最终需要通过HeadContex直接调用Netty的API来完成具体的动作，故HeadContex关于outbound事件的实现，都是通过调用unsafe去完成具体的动作。故后面的方面就不在一一罗列。

2.3 源码分析DefaultChannelPipeline$TailContext

TailContext由于是 inbound事件链的最后一站，故该节点大部分事件都是空实现，其他实现的方法，基本上就是释放一下资源，我们看一下TailContex关于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);
        }
}

最后就是主动调用ReferenceCountUtil.release(msg)释放资源。

Netty事件传播机制就讲解到这里了。

欢迎关注『中间件兴趣圈』，交接中间件、架构设计、实战应用等技术课题。

年度精彩热文TOP3