Netty源码分析——pipeline增减节点

前言

pipeline在Netty中的概念，就好比流水线，有一个socket接入，就分配一条流水线，流经想要的工人，处理之后再返回。这篇文章主要是看看pipeline如何增减工人——Handler。

初始化和绑定

代码：

protected AbstractChannel(Channel parent) {
    this.parent = parent;
    id = newId();
    unsafe = newUnsafe();
    pipeline = newChannelPipeline();
}

//newChannelPipeline()
return new DefaultChannelPipeline(this);

//DefaultChannelPipeline构造函数
protected DefaultChannelPipeline(Channel channel) {
    this.channel = ObjectUtil.checkNotNull(channel, "channel");
    succeededFuture = new SucceededChannelFuture(channel, null);
    voidPromise =  new VoidChannelPromise(channel, true);

    tail = new TailContext(this);
    head = new HeadContext(this);

    head.next = tail;
    tail.prev = head;
}

pipeline是DefaultChannelPipeline，默认保存了channel的引用，以及一个头结点和尾节点。注意这里TailContext是一种ChannelInboundHandler，而HeadContext是ChannelOutboundHandler和ChannelInboundHandler。二者都是AbstractChannelHandlerContext的实例。

Pipeline中，每一个节点都是一个ChannelHandlerContext。默认的ChannelHandlerContext是DefaultChannelHandlerContext，这个context中维护了一个ChannelHandler。但是作为特殊的两种ChannelHandlerContext，TailContext和HeadContext都是没有handler的，初始化之后pipeline应该是这样的：

红色的是ChannelOutboundHandler，蓝色的是
ChannelInboundHandler，这种对应关系会延续整篇文章。其中ChannelInboundHandler主要处理inBound事件，主要对应数据流入的处理，比如channelRead。而反过来ChannelOutboundHandler则对应着outBound事件，比如writeAndFlush。

添加节点

添加节点我们举个简单的例子：

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

这时候的pipeline应该如下：

看下DefaultPipeline#addLast：

final AbstractChannelHandlerContext newCtx;
synchronized (this) {、
    //检查是否有重复的handler
    checkMultiplicity(handler);
    //初始化AbstractChannelHandlerContext
    //filterName用来给handler创建唯一的名字
    newCtx = newContext(group, filterName(name, handler), handler);
    //添加
    addLast0(newCtx);
    if (!registered) {
        newCtx.setAddPending();
        callHandlerCallbackLater(newCtx, true);
        return this;
    }

    EventExecutor executor = newCtx.executor();
    if (!executor.inEventLoop()) {
        newCtx.setAddPending();
        executor.execute(new Runnable() {
            @Override
            public void run() {
                callHandlerAdded0(newCtx);
            }
        });
        return this;
    }
}
//回调 以及更新状态为已添加
callHandlerAdded0(newCtx);
return this;

这里是简单的双向链表，多线程操作add会有问题，这里直接加了个锁。

检查是否有重复的handler
创建AbstractChannelHandlerContext。
添加节点
回调

第一步比较简单，就不展开了。校验是通过Sharable注解和ChannelHandlerAdapter的added熟悉来做的。如果一个ChannelHandler没有加Sharable注解，并且之前被添加过（其实是执行过checkMultiplicity方法），则报错。

继续看创建节点。newContext方法其实就是创建了一个DefaultChannelHandlerContext，这个DefaultChannelHandlerContext保存了对应的Handler，注意，AbstractChannelHandlerContext是可以直接区分是哪种Context的，这里的哪种，具体是inBound和outBound两种，这里可以直接根据Handler的类型判断：

DefaultChannelHandlerContext(
        DefaultChannelPipeline pipeline, EventExecutor executor, String name, ChannelHandler handler) {
    super(pipeline, executor, name, isInbound(handler), isOutbound(handler));
    this.handler = handler;
}

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

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

直接判断Handler是那种Handler的实现即可。有一些Handler是双向的，即两种接口都实现了的。比较简单就不展开说了，netty里可以找到很多codec的实现，这就是一种双向的handler。

继续看添加，addLast0：

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

很简单的双向链表的添加节点操作，如果有看不懂的同学，可能要好好补一下你的数据结构了~

注意这里是addLast，其实就是从tail开始往前加。其他所有的add方法都类似就不展开说了。

至此节点添加就讲完了。

删除节点

remove(ChannelHandler handler)方法：

remove(getContextOrDie(handler));

这里需要先循环所有的context节点，寻找一个context.handler = 入参handler的节点，然后从pipeline中移除这个context即可。移除结束同样会回调方法并且更新状态，这里就不展开了。