[netty源码分析]--ChannelPipeline源码分析(一)

Love The Way You Lie 2022-09-29 06:33 327阅读 0赞

在前面我们已经分析过了netty的线程模型，了解了当线程池启动之后就会处于一种事件循环的状态，不断调用select() 函数，查询已注册的channel中的就绪事件。

下面我们就来研究一下当IO事件被获取之后是怎么在ChannelPipeline中的handler传递运行的。

# 1. Channel和ChannelPipeline的关系 #

在netty中，每个Channel 有且仅有一个ChannelPipeline与之对应，他们的组成关系如下图所示：

![这里写图片描述][SouthEast]

从上图我们也可以看出，每一个Channel都与一个ChannelPipeline对应。 而且在ChannelPipeline 中又维护了一个由ChannelHandlerContext 组成的双向链表。这个链表的头是 HeadContext, 链表的尾是 TailContext, 并且每个 ChannelHandlerContext 又与一个 ChannelHandler一一对应。上面的图示给了我们一个对 ChannelPipeline 的直观认识, 但是实际上 Netty 实现的 Channel 是否真的是这样的呢? 我们继续用源码说话。

在分析netty的启动引导类源码的时候我们可以知道，在执行bind()函数时会对channel进行初始化并注册，其中进行到后面会调用基类AbstractChannel的构造函数：

protected AbstractChannel(Channel parent) {
        this.parent = parent;
        unsafe = newUnsafe();
        this.pipeline = new DefaultChannelPipeline(this);
    }

我们可以知道AbstractChannel 有一个 pipeline 字段, 在构造器中会初始化它为 DefaultChannelPipeline的实例. 这里的代码就印证了一点: 每个 Channel 都有一个 ChannelPipeline，并且是一一对应的关系。

下面我们就来关注一下DefaultChannelPipeline的构造器初始化的过程，先给出源码：

protected DefaultChannelPipeline(Channel channel) {
        //channel绑定操作;
        this.channel = ObjectUtil.checkNotNull(channel, "channel");
        succeededFuture = new SucceededChannelFuture(channel, null);
        voidPromise =  new VoidChannelPromise(channel, true);
    
        // channelpipeline 双向链表的 尾部
        tail = new TailContext(this);
        // channelpipeline 双向链表的 首部
        head = new HeadContext(this);
        //将收尾结点链接成双向链表
        head.next = tail;
        tail.prev = head;
    }

构造器中, 首先将与DefaultChannelPipeline关联的 Channel 保存到字段 channel 中, 然后实例化两个 ChannelHandlerContext, 一个是 HeadContext 实例 head, 另一个是 TailContext 实例 tail。 接着将 head 和 tail 互相指向, 构成一个双向链表。

特别注意到, 我们在开始的示意图中, head 和 tail 并没有包含 ChannelHandler, 这是因为 HeadContext 和 TailContext 继承于 AbstractChannelHandlerContext 的同时也实现了 ChannelHandler 接口了, 因此它们有 Context 和 Handler 的双重属性.

# 2. ChannelPipeline的初始化 #

前面讲到了ChannelPipeline的默认实现类的构造器，这里我们从两个点：ChannelPipeline的实例化以及 ChannelInitializer的添加来分析。

## ChannelPipeline的实例化 ##

在实例化一个 Channel 时, 会伴随着一个 ChannelPipeline 的实例化, 并且此 Channel 会与这个 ChannelPipeline 相互关联, 这一点可以通过NioSocketChannel 的父类 AbstractChannel 的构造器予以佐证:

protected AbstractChannel(Channel parent) {
        this.parent = parent;
        unsafe = newUnsafe();
        pipeline = new DefaultChannelPipeline(this);
    }

当实例化一个 Channel(这里以 EchoClient 为例, 那么 Channel 就是 NioSocketChannel), 其 pipeline 字段就是我们新创建的 DefaultChannelPipeline 对象, 那么我们就来看一下 DefaultChannelPipeline 的构造方法吧:

可以看到, 在 DefaultChannelPipeline 的构造方法中, 将传入的 channel 赋值给字段 this.channel, 接着又实例化了两个特殊的字段: tail 与 head. 这两个字段是一个双向链表的头和尾. 其实在 DefaultChannelPipeline 中, 维护了一个以 AbstractChannelHandlerContext 为节点的双向链表, 这个链表是 Netty 实现 Pipeline 机制的关键。

回顾一下 head 和 tail 的类层次结构:

![这里写图片描述][SouthEast 1]

HeadContext的类继承结构

![这里写图片描述][SouthEast 2]

TailContext的类继承结构

从类继承图可以看出：head 实现了 ChannelOutboundHandler接口，作为出站事件的起点。tail实现了ChannelInboundHandler 接口，作为入站事件的起点。并且head和tail都实现了 ChannelHandlerContext 接口, 因此可以说 head 和 tail 即是一个 ChannelHandler, 又是一个 ChannelHandlerContext。

下面看看这两个类的构造器：

HeadContext(DefaultChannelPipeline pipeline) {
        // 第四,五个参数分别是FALSE，TRUE, 表示这是一个出站 context
        super(pipeline, null, HEAD_NAME, false, true);
        unsafe = pipeline.channel().unsafe();
        setAddComplete();
    }
    
    TailContext(DefaultChannelPipeline pipeline) {
        // 第四,五个参数分别是TRUE，FALSE, 表示这是一个出站 context
        super(pipeline, null, TAIL_NAME, true, false);
        setAddComplete();
    }

它调用了父类 AbstractChannelHandlerContext 的构造器：

AbstractChannelHandlerContext(DefaultChannelPipeline pipeline, EventExecutor executor, String name,
                                      boolean inbound, boolean outbound) {
        this.name = ObjectUtil.checkNotNull(name, "name");
        this.pipeline = pipeline;
        this.executor = executor;
        this.inbound = inbound;
        this.outbound = outbound;
        // Its ordered if its driven by the EventLoop or the given Executor is an instanceof OrderedEventExecutor.
        ordered = executor == null || executor instanceof OrderedEventExecutor;
    }

headContext传入参数 inbound = false, outbound = true.  
TailContext 的构造器与 HeadContext 的相反, 它调用了父类 AbstractChannelHandlerContext 的构造器, 并传入参数 inbound = true, outbound = false。

即 header 是一个 outboundHandler, 而 tail 是一个inboundHandler, 关于这一点, 大家要特别注意, 因为在后面的分析中, 我们会反复用到 inbound 和 outbound 这两个属性.

## ChannelInitializer 的添加 ##

上面一小节中, 我们已经分析了 Channel 的组成, 其中我们了解到, 最开始的时候 ChannelPipeline 中含有两个 ChannelHandlerContext(同时也是 ChannelHandler), 但是这个 Pipeline并不能实现什么特殊的功能, 因为我们还没有给它添加自定义的 ChannelHandler.通常来说, 我们在初始化 Bootstrap, 会添加我们自定义的 ChannelHandler, 就以我们熟悉的 EchoClient 来举例吧:

Bootstrap b = new Bootstrap();
    b.group(group)
     .channel(NioSocketChannel.class)
     .option(ChannelOption.TCP_NODELAY, true)
     .handler(new ChannelInitializer<SocketChannel>() {
         @Override
         public void initChannel(SocketChannel ch) throws Exception {
             ChannelPipeline p = ch.pipeline();
             p.addLast(new EchoClientHandler());
         }
     });

上面代码的初始化过程, 相信大家都不陌生。在调用 handler 时, 传入了 ChannelInitializer 对象, 它提供了一个 initChannel 方法供我们初始化 ChannelHandler. 那么这个初始化过程是怎样的呢? 下面我们就来揭开它的神秘面纱.

ChannelInitializer 实现了 ChannelHandler, 那么它是在什么时候添加到 ChannelPipeline 中的呢? 进行了一番搜索后, 我们发现它是在 Bootstrap.init() 方法中添加到 ChannelPipeline 中的.其代码如下:

void init(Channel channel) throws Exception {
        ChannelPipeline p = channel.pipeline();
        p.addLast(handler());
        ...
    }

上面的代码将 handler() 返回的 ChannelHandler 添加到 Pipeline 中, 而 handler() 返回的是handler 其实就是我们在初始化 Bootstrap 调用 handler 设置的 ChannelInitializer 实例, 因此这里就是将 ChannelInitializer 插入到了 Pipeline 的末端.此时 Pipeline 的结构如下图所示:

![这里写图片描述][SouthEast 3]

有朋友可能就有疑惑了, 我明明插入的是一个 ChannelInitializer 实例, 为什么在 ChannelPipeline 中的双向链表中的元素却是一个 ChannelHandlerContext? 为了解答这个问题，我们继续在代码中寻找答案吧。我们刚才提到， 在 Bootstrap.init 中会调用 p.addLast() 方法, 将 ChannelInitializer 插入到链表末端:

public final ChannelPipeline addLast(ChannelHandler... handlers) {
        return addLast(null, handlers);
    }
    
    //向下调用
    public final ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {
        if (handlers == null) {
            throw new NullPointerException("handlers");
        }
        //迭代添加handlers到 pipeline中.
        for (ChannelHandler h: handlers) {
            if (h == null) {
                break;
            }
            addLast(executor, null, h);
        }
        return this;
    }

上面其实就是迭代handlers， 将每个handler插入到pipeline中：  
下面的源码只保留核心部分

public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
        // 新添加到 pipeline 双链表中的结点.
        final AbstractChannelHandlerContext newCtx;
        synchronized (this) {
            checkMultiplicity(handler);
            //实例化结点，将结点与handler一一绑定。
            newCtx = newContext(group, filterName(name, handler), handler);
            //插入结点
            addLast0(newCtx);
            //...............
    }

上面通过调用newContext()方式 实例化一个新的AbstractChannelHandlerContext，源码如下：

private AbstractChannelHandlerContext newContext(EventExecutorGroup group, String name, ChannelHandler handler) {
       return new DefaultChannelHandlerContext(this, childExecutor(group), name, handler);
    }

这里就是调用了AbstractChannelHandlerContext默认实现类DefaultChannelHandlerContext，构造器如下：

DefaultChannelHandlerContext(
                DefaultChannelPipeline pipeline, EventExecutor executor, String name, ChannelHandler handler) {
        super(pipeline, executor, name, isInbound(handler), isOutbound(handler));
        if (handler == null) {
            throw new NullPointerException("handler");
        }
        this.handler = handler;
    }

DefaultChannelHandlerContext 的构造器中, 调用了两个很有意思的方法: isInbound 与 isOutbound, 这两个方法是做什么的呢?

private static boolean isInbound(ChannelHandler handler) {
        return handler instanceof ChannelInboundHandler;
    }
    
    private static boolean isOutbound(ChannelHandler handler) {
        return handler instanceof ChannelOutboundHandler;
    }

从源码中可以看到, 当一个 handler 实现了 ChannelInboundHandler 接口, 则 isInbound 返回真; 相似地, 当一个 handler 实现了 ChannelOutboundHandler 接口, 则 isOutbound 就返回真.  
而这两个 boolean 变量会传递到父类 AbstractChannelHandlerContext 中, 并初始化父类的两个字段: inbound 与 outbound.

那么这里的 ChannelInitializer 所对应的 DefaultChannelHandlerContext 的 inbound 与 inbound 字段分别是什么呢? 那就看一下 ChannelInitializer 到底实现了哪个接口不就行了? 如下是 ChannelInitializer 的类层次结构图:  
![这里写图片描述][SouthEast 4]

可以清楚地看到, ChannelInitializer 仅仅实现了 ChannelInboundHandler 接口, 因此这里实例化的 DefaultChannelHandlerContext 的 inbound = true, outbound = false.

不就是 inbound 和 outbound 两个字段嘛, 为什么需要这么大费周章地分析一番? 其实这两个字段关系到 pipeline 的事件的流向与分类, 因此是十分关键的, 不过我在这里先卖个关子, 后面我们再来详细分析这两个字段所起的作用. 在这里, 读者只需要记住, ChannelInitializer 所对应的 DefaultChannelHandlerContext 的 inbound = true, outbound = false 即可.

实例化了AbstractChannelHandlerContext之后就插入pipeline的结点，插入的位置就是双链表的尾部，其实是tail尾结点的前面，也就是类似于尾插法。

private void addLast0(AbstractChannelHandlerContext newCtx) {
        AbstractChannelHandlerContext prev = tail.prev;
        newCtx.prev = prev;
        newCtx.next = tail;
        prev.next = newCtx;
        tail.prev = newCtx;
    }

## 自定义 ChannelHandler 的添加过程 ##

在上一小节中, 我们已经分析了一个 ChannelInitializer 如何插入到 Pipeline 中的, 接下来就来探讨一下 ChannelInitializer 的 initChannel()方法在哪里被调用？ChannelInitializer 的作用, 以及我们自定义的 ChannelHandler 是如何插入到 Pipeline 中的.

在 分析Netty 源码 Bootstrap 的 channel 的注册过程中, 我们已经分析过 Channel 的注册过程了, 这里我们再简单地复习一下:

（1）首先在 AbstractBootstrap.initAndRegister中, 通过 group().register(channel), 调用 MultithreadEventLoopGroup.register 方法

（2）在MultithreadEventLoopGroup.register 中, 通过 next() 获取一个可用的 SingleThreadEventLoop, 然后调用它的 register

（3）在 SingleThreadEventLoop.register 中, 通过 channel.unsafe().register(this, promise) 来获取 channel 的 unsafe() 底层操作对象, 然后调用它的 register.

（4）在 AbstractUnsafe.register 方法中, 调用 register0 方法注册 Channel

（5）在 AbstractUnsafe.register0 中, 调用 AbstractNioChannel\#doRegister 方法

（6）AbstractNioChannel.doRegister 方法通过 javaChannel().register(eventLoop().selector, 0, this) 将 Channel 对应的 Java NIO SockerChannel 注册到一个 eventLoop 的 Selector 中, 并且将当前 Channel 作为 attachment.

而我们自定义 ChannelHandler 的添加过程, 发生在 AbstractUnsafe.register0 中, 在这个方法中调用了 pipeline.fireChannelRegistered() 方法, 其实现如下:

public final ChannelPipeline fireChannelRegistered() {
        AbstractChannelHandlerContext.invokeChannelRegistered(head);
        return this;
    }

上面代码很简单，直接调用`AbstractChannelHandlerContext.invokeChannelRegistered(head);` 上面的参数是 head 结点，也就是从头结点head开始在 pipeline中传递事件了。看看该函数的核心部分源码：

static void invokeChannelRegistered(final AbstractChannelHandlerContext next) {
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            //核心部分
            next.invokeChannelRegistered();
        } else {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    next.invokeChannelRegistered();
                }
            });
        }
    }

上文中`invokeChannelRegistered(AbstractChannelHandlerContext);` 里面又调用了`AbstractChannelHandlerContext.invokeChannelRegistered();` 这其实就是函数的重载，看看源码：

private void invokeChannelRegistered() {
        if (invokeHandler()) {
            try {
                ((ChannelInboundHandler) handler()).channelRegistered(this);
            } catch (Throwable t) {
                notifyHandlerException(t);
            }
        } else {
            fireChannelRegistered();
        }
    }

这个时候就是调用handler() 返回的最初其实就是head相关的 ChannelHandler，然后调用ChannelHandler.channelRegistered(AbstractChannelHandlerContext)， 看到这里, 读者朋友是否会觉得有点眼熟呢? ChannelInitializer.channelRegistered() 这个方法我们在第一章的时候已经大量地接触了, 但是我们并没有深入地分析这个方法的调用过程, 那么在这里读者朋友应该对它的调用有了更加深入的了解了吧.

public final void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        if (initChannel(ctx)) {
            ctx.pipeline().fireChannelRegistered();
        } else {
            ctx.fireChannelRegistered();
        }
    }

initChannel 这个方法我们很熟悉了吧, 它就是我们在初始化 Bootstrap 时, 调用 handler 方法传入的匿名内部类所实现的方法:

.handler(new ChannelInitializer<SocketChannel>() {
         @Override
         public void initChannel(SocketChannel ch) throws Exception {
             ChannelPipeline p = ch.pipeline();
             p.addLast(new EchoClientHandler());
         }
     });

因此当调用了这个方法后, 我们自定义的 ChannelHandler 就插入到 Pipeline 了, 此时的 Pipeline 如下图所示:

![这里写图片描述][SouthEast 5]

当添加了自定义的 ChannelHandler 后, 会删除 ChannelInitializer 这个 ChannelHandler, 即在initChannel(ctx)函数里面调用 remove(this), 因此最后的 Pipeline 如下:

![这里写图片描述][SouthEast 6]

好了, 到了这里, 我们的 自定义 ChannelHandler 的添加过程 也分析的查不多了。

[SouthEast]: /images/20220708/364e38df995f4cc1b7c295c91d1e281d.png
[SouthEast 1]: /images/20220708/072a42c2504c477980049a837807f8e1.png
[SouthEast 2]: /images/20220708/63d63b26e7724f779dc74643b4fcd92f.png
[SouthEast 3]: /images/20220708/d60309f3fa614e5892197de982f305d4.png
[SouthEast 4]: /images/20220708/1b65304bdfe641edaa5ca07b6b9ac47b.png
[SouthEast 5]: /images/20220708/63ae323d4aa749eda8859f424b4d6f64.png
[SouthEast 6]: /images/20220708/d6cb860ceebe4d148e4a93c9fd04283c.png