【Netty】原理分析：ChannelPipeline（转）

你的名字 2023-01-01 04:00 160阅读 0赞

> 原文链接：[https://blog.csdn.net/zxhoo/article/details/17264263][https_blog.csdn.net_zxhoo_article_details_17264263]。本文对部分内容进行了重新排版和补充

[ChannelPipeline][] 实际上应该叫做 ChannelHandlerPipeline，可以把 ChannelPipeline 看成是一个 ChandlerHandler 的容器。当需要对 Channel 进行某种处理的时候，Pipeline 负责依次调用每一个 Handler 进行处理。每个 Channel 都有一个属于自己的 Pipeline，调用 `Channel#pipeline()` 方法可以获得 Channel 的 Pipeline，调用`Pipeline#channel()`方法可以获得 Pipeline 的 Channel。

![img][]

### 1.ChannelPipeline 如何管理 Handler ###

ChannelPipeline 的方法有很多，其中一部分是用来管理 ChannelHandler 的，如下面这些：

ChannelPipeline addFirst(String name, ChannelHandler handler);
    ChannelPipeline addLast(String name, ChannelHandler handler);
    ChannelPipeline addBefore(String baseName, String name, ChannelHandler handler);
    ChannelPipeline addAfter(String baseName, String name, ChannelHandler handler);
    ChannelPipeline remove(ChannelHandler handler);
    ChannelHandler remove(String name);
    ChannelHandler removeFirst();
    ChannelHandler removeLast();
    ChannelPipeline replace(ChannelHandler oldHandler, String newName, ChannelHandler newHandler);
    ChannelHandler replace(String oldName, String newName, ChannelHandler newHandler);
    ChannelHandler first();
    ChannelHandler last();
    ChannelHandler get(String name);

根据上面的方法，能够大概想象的到 Pipeline 按照什么样的方式组织 Handler – **链表**。

但是，ChannelPipeline 并不是直接管理 ChannelHandler，而是通过 [**ChannelHandlerContext**][ChannelHandlerContext] 来间接管理（即链表的 Node 实际是 ChannelHandlerContext）。

调用`ChannelHandlerContext#channel()`方法可以得到和 Context 绑定的 Channel，调用`ChannelHandlerContext#handler()`方法可以得到和Context绑定的Handler。  
![img][img 1]

这一点通过 ChannelPipeline 的默认实现 DefaultChannelPipeline 可以看出来，下面是它的部分代码：

// ChannelPipeline 是接口，DefaultChannelPipeline 是默认实现
    final class DefaultChannelPipeline implements ChannelPipeline {
        
        // ...
     
        final DefaultChannelHandlerContext head; // 头结点
        final DefaultChannelHandlerContext tail; // 尾结点
     
        // ...
    }

上面的 DefaultChannelHandlerContext 是 ChannelHandlerContext 的默认实现，下面是它的部分代码：

final class DefaultChannelHandlerContext extends DefaultAttributeMap implements ChannelHandlerContext {
        
     
        volatile DefaultChannelHandlerContext next; // 前驱指针
        volatile DefaultChannelHandlerContext prev; // 后向指针
     
        private final boolean inbound;
        private final boolean outbound;
        private final AbstractChannel channel;
        private final DefaultChannelPipeline pipeline;
        private final String name;
        private final ChannelHandler handler; // Handler
        private boolean removed;
     
        // ...
    }

从上面的两段代码可以看出，在 DefaultPipeline 内部，由 DefaultChannelHandlerContext 实际上组成了一个**双向链表**：

![img][img 2]

再来看看 DefaultChannelPipeline 的构造函数：

public DefaultChannelPipeline(AbstractChannel channel) {
        
            if (channel == null) {
        
                throw new NullPointerException("channel");
            }
            this.channel = channel;
     
            TailHandler tailHandler = new TailHandler();
            tail = new DefaultChannelHandlerContext(this, null, generateName(tailHandler), tailHandler);
     
            HeadHandler headHandler = new HeadHandler(channel.unsafe());
            head = new DefaultChannelHandlerContext(this, null, generateName(headHandler), headHandler);
     
            head.next = tail;
            tail.prev = head;
        }

可以看到，DefaultChinnelPipeline 内部使用了两个特殊的 Handler 来表示 Handler 链的头和尾：

![img][img 3]

所以，我们添加的自定义 ChannelHandler 会插入到 head 和 tail 之间，以 addLast 为例:

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]

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

可以看到，addLast0 做的就是基本的链表指针操作，我们添加的自定义 ChannelHandler 会插入到 head 和 tail 之间。

> 注：插入的顺序取决于我们添加时的先后顺序。

### 2.事件的传播过程 ###

为了搞清楚事件如何在 Pipeline 里传播，让我们从 Channel 的抽象子类 AbstractChannel 开始

1）下面是 AbstractChannel\#write() 方法的实现：

final class DefaultChannelPipeline implements ChannelPipeline {
        
        // ...
        @Override
        public ChannelFuture write(Object msg) {
        
            return tail.write(msg);
        }
        // ...
    }

AbstractChannel 直接调用了 Pipeline 的 write() 方法：

![img][img 4]

2）再看 DefaultChannelPipeline 的 write() 方法实现：

因为 write 是个 outbound 事件，所以 DefaultChannelPipeline 直接找到 **tail** 部分的 context，调用其 write() 方法：

![img][img 5]

3）接着看 DefaultChannelHandlerContext 的 write() 方法：

final class DefaultChannelHandlerContext extends DefaultAttributeMap implements ChannelHandlerContext {
        
        // ...
        @Override
        public ChannelFuture write(Object msg) {
        
            return write(msg, newPromise());
        }
     
        @Override
        public ChannelFuture write(final Object msg, final ChannelPromise promise) {
        
            if (msg == null) {
        
                throw new NullPointerException("msg");
            }
     
            validatePromise(promise, true);
     
            write(msg, false, promise);
     
            return promise;
        }
     
        private void write(Object msg, boolean flush, ChannelPromise promise) {
        
        	// 拿到前驱 ChannelHandlerContext
            DefaultChannelHandlerContext next = findContextOutbound();
            // 调用前驱的 write()
            next.invokeWrite(msg, promise);
            if (flush) {
        
                next.invokeFlush();
            }
        }
     
        private DefaultChannelHandlerContext findContextOutbound() {
        
            DefaultChannelHandlerContext ctx = this;
            do {
        
            	// 拿到前驱 ChannelHandlerContext
                ctx = ctx.prev;
                
            } while (!ctx.outbound);// 循环结束条件是找到 ChannelOutBoundHandler
            return ctx;
        }
     
        private void invokeWrite(Object msg, ChannelPromise promise) {
        
            try {
        
            	// 调用前驱 ChannelHandler 的 write()
                ((ChannelOutboundHandler) handler).write(this, msg, promise);
            } catch (Throwable t) {
        
                notifyOutboundHandlerException(t, promise);
            }
        }
     
        // ...
    }

context 的 write() 方法沿着 context 链往前找，直至找到一个 outbound 类型的 context 为止，然后调用其 invokeWrite() 方法：

![img][img 6]

invokeWrite() 接着调用 handler 的 write() 方法：  
![img][img 7]

4）最后看看 ChannelOutboundHandlerAdapter 的 write() 方法实现：

public class ChannelOutboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelOutboundHandler {
        
        // ...
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        
            ctx.write(msg, promise);
        }
        // ...
    }

默认的实现调用了 context 的 write() 方法而不做任何处理，这样 write 事件就沿着 outbound 链继续传播：

![img][img 8]

可见，Pipeline 的事件传播，是靠 Pipeline，Context 和 Handler 共同协作完成的。

[https_blog.csdn.net_zxhoo_article_details_17264263]: https://blog.csdn.net/zxhoo/article/details/17264263
[ChannelPipeline]: http://netty.io/4.0/api/io/netty/channel/ChannelPipeline.html
[img]: /images/20221120/928674cfc67c4f47a4871f62efa8c81a.png
[ChannelHandlerContext]: http://netty.io/4.0/api/io/netty/channel/ChannelHandlerContext.html
[img 1]: /images/20221120/dbedc99519ac49838e67180af8d92a0c.png
[img 2]: /images/20221120/2c718e40855b47628d80b71d894cb747.png
[img 3]: https://img-blog.csdn.net/20131212120733046?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvenhob28=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: https://img-blog.csdnimg.cn/20201231150640915.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw==,size_16,color_FFFFFF,t_70#pic_center
[img 4]: https://img-blog.csdn.net/20131212133508828?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvenhob28=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast#pic_center
[img 5]: https://img-blog.csdn.net/20131212134216687?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvenhob28=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast#pic_center
[img 6]: https://img-blog.csdn.net/20131212140450171?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvenhob28=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast#pic_center
[img 7]: https://img-blog.csdn.net/20131212140949687?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvenhob28=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast#pic_center
[img 8]: https://img-blog.csdn.net/20131212141601265?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvenhob28=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast