[netty核心类]--ChannelPipeline和ChannelHandler源码分析

女爷i 2022-07-12 08:41 235阅读 0赞

> 本文主要包括：  
> （1）ChannelPipeline的功能说明和源码分析；  
> （2）ChannelHandler的功能说明和源码分析；

在netty中的ChannelPipeline和ChannelHandler的机制就很类似于J2EE中的Servlet和Filter过滤器，这类拦截器就很类似于责任链模式的一种变形。核心就是事件的拦截和用户业务逻辑的定制。

netty中channel过滤器实现原理和servlet的Filter基本一致，netty将Channel的数据管道抽象为ChannelPipeline，消息在ChannelPipeline中流动和传递。ChannelPipeline又持有IO事件拦截器ChannelHandler的链表，由ChannelPipeline中的channelHandler对IO事件进行拦截和处理，可以通过在ChannelPipeline中新增和删除ChannelHandler来实现不同业务逻辑的定制。这样就不需要更改ChannelHandler中的代码，就能实现扩展和修改。

# 1. ChannelPipeline功能说明 #

ChannelPipeline其实就是ChannelHandler的容器，它维护了一个ChannelHandler的链表。

## （1）ChannelPipeline的消息事件处理过程 ##

下图展示了一个消息被ChannelPipeline的ChannelHandler链拦截和处理的全过程：  
![这里写图片描述][SouthEast]

下面对整个消息的处理和发送处理过程进行一个解释：  
（1）先从左下方的开始：底层的SocketChannel.read()方法读取ByteBuf，然后触发ChannelRead事件，由IO线程NioEventLoop调用ChannelPipeline的fireChannelRead(Object msg)方法，将消息ByteBuf传输到ChannelPipeline中；

（2）消息依次经过HeadHandler、ChannelHandler1、ChannelHandler2…..TailHandler拦截和处理，在这个过程中，任何ChannelHandler都可以中断当前的流程，结束消息传递。

（3）调用ChannelHandlerContext的write()方法发送消息，消息从TailHandler开始，途径ChannelHandlerN….ChannelHandler1、HeadHandler，最终被添加到消息发送缓冲区，等待被发送到Channel中，在此过程中也可以中断消息的发送，比如编码失败的时候。

在netty中的事件分为inbound入站事件和outbound出站事件。inbound事件一般是由IO线程触发的，例如TCP链路建立事件、链路关闭事件、读事件、异常通知事件等，对应的是上图的左半部分。触发inbound事件的方法都是调用ChannelHandlerContext里面的方法，具体如下：

//channel注册事件
    ChannelHandlerContext ChannelHandlerContext fireChannelRegistered();
    //channel注销事件
    ChannelHandlerContext ChannelHandlerContext fireChannelUnregistered();
    //TCP链路建立成功，Channel激活事件
    ChannelHandlerContext ChannelHandlerContext fireChannelActive();
    //TCP连接关闭，链路不可用通知事件
    ChannelHandlerContext ChannelHandlerContext fireChannelInactive();
    //异常通知事件
    ChannelHandlerContext ChannelHandlerContext fireExceptionCaught(Throwable cause);
    //用户自定义事件
    ChannelHandlerContext ChannelHandlerContext fireUserEventTriggered(Object evt);
    //读事件
    ChannelHandlerContext ChannelHandlerContext fireChannelRead(Object msg);
    //读操作完成事件
    ChannelHandlerContext ChannelHandlerContext fireChannelReadComplete();
    //channel的可写状态变化通知事件
    ChannelHandlerContext ChannelHandlerContext fireChannelWritabilityChanged();

outbound事件一般都是用户主动发起的网络IO事件，比如用户发起的连接操作，绑定操作、消息发送等操作，对应上图的有半部分。调用ChannelHandlerContext里面的方法，具体如下：

//绑定本地地址事件
    ChannelFuture bind(SocketAddress localAddress);
    //连接服务端事件
    ChannelFuture connect(SocketAddress remoteAddress);
    //发送事件
    ChannelFuture write(Object msg);
    //读事件
    ChannelHandlerContext read();
    //刷新事件
    ChannelHandlerContext flush();
    //断开连接事件
    ChannelFuture disconnect();
    //关闭当前channel事件
    ChannelFuture close();

## （2）自定义拦截器与构建pipeline ##

前面已经说了多次，ChannelPipeline通过ChannelHandler来实现事件的拦截和处理。  
关于这一部分的内容可以查看着一篇博客： [\[netty中的ChannelHandler\]][netty_ChannelHandler] [http://blog.csdn.net/u010853261/article/details/54574440][netty_ChannelHandler]

## （3）ChannelPipeline的主要特性 ##

ChannelPipeline支持运行态动态的添加和删除ChannelHandler。这在某些场景下非常的实用：比如业务高峰期需要对系统做阻塞处理，就可以根据当前系统时间判断，如果处于高峰期就动态将系统阻塞保护channelHandler添加到当前Pipeline中，当高峰期过了之后，就可以动态的删除阻塞保护ChannelHandler。

pipeline是线程安全的，查看DefaultChannelPipeline源码就可以知道，里面的核心函数都使用synchronized同步块加锁了，这意味着N个业务线程可以并发的操作ChannelPipeline而不存在线程安全的问题。但是ChannelHandler却不是线程安全的，所以我们自己必须要保证ChannelHandler的线程安全。

# 2. ChannelPipeline源码分析 #

首先看看ChannelPipeline的类继承关系，比较简单，ChannelPipeline接口默认实现类是DefaultChannelPipeline  
![这里写图片描述][SouthEast 1]

## （1）ChannelPipeline对ChannelHandler的管理 ##

ChannelPipeline是ChannelHandler的容器，负责ChannelHandler的查询、添加、删除、替换等。前面我们就说了，ChannelPipeline里面维护的就是一个AbstractChannelHandlerContext的双链表，而且AbstractChannelHandlerContext与ChannelHandler是一一对应的映射关系。

DefaultChannelPipeline的成员变量

//头结点的handler的名字。
    private static final String HEAD_NAME = generateName0(HeadContext.class);
    //尾结点的handler的名字
    private static final String TAIL_NAME = generateName0(TailContext.class);
    //头结点
    final AbstractChannelHandlerContext head;
    //尾结点
    final AbstractChannelHandlerContext tail;

新增Handler函数分析  
以`ChannelPipeline addBefore(String baseName, String name, ChannelHandler handler)` 函数为例进行分析，这个函数传入的参数有三个，添加到Handler之前的那个handler的名字； 当前待添加的handler 的名字，以及待添加的handler。

该函数会直接调用`addBefore( EventExecutorGroup group, String baseName, String name, ChannelHandler handler)` 重载的函数，源码如下：

public final ChannelPipeline addBefore(
                EventExecutorGroup group, String baseName, String name, ChannelHandler handler) {
        //新插入的handler对应的ChannelHandlerContext
        final AbstractChannelHandlerContext newCtx;
        //baseName对应的handler
        final AbstractChannelHandlerContext ctx;
        //加锁保证Pipeline的线程安全性
        synchronized (this) {
            //校验：被添加的handler确保能在多个ChannelPipeline共享并且没有被添加到pipeline中
            checkMultiplicity(handler);
            //对name和handler过滤：当name为null,就根据handler创建name；
            //如果name非空，就遍历所有已经添加进来的handler的名字，保证名字的唯一性。
            name = filterName(name, handler);
            //根据basename获取对应的ctx
            ctx = getContextOrDie(baseName);
            //创建handler唯一对应的AbstractChannelHandlerContext
            newCtx = newContext(group, name, handler);
            //将新的newCtx插入在ctx的后面，这里其实就是双链表的插入
            addBefore0(ctx, newCtx);
    
            // If the registered is false it means that the channel was not registered on an eventloop yet.
            // In this case we add the context to the pipeline and add a task that will call
            // ChannelHandler.handlerAdded(...) once the channel is registered.
            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;
            }
        }
        //发送新增ChannelHandlerContext的通知
        callHandlerAdded0(newCtx);
        return this;
    }

根据注释作如下新增handler的总结：  
（1）获取插入点baseName所对应的AbstractChannelHandlerContext；

（2）添加synchronized同步块保证线程安全性；

（3）检验被添加的handler的对应的name的合法性，然后创建handler对应的AbstractChannelHandlerContext，再利用双链表插入方法，插入到pipeline维护的AbstractChannelHandlerContext双向链表中。

（4）调用callHandlerAdded0(newCtx); 发送新增handler的通知。

## （2）ChannelPipeline的inbound事件和outbound事件 ##

·inbound事件  
当发生某些IO事件时候，比如链路连接、关闭、读取操作的完成等都会产生一个事件，事件在pipeline中传播。

pipeline中以fireXXXXX命名的方法都是从IO线程流向用户业务(handler) 的inbound事件，它们的实现因功能而异，但是处理步骤类似：  
（1）调用HeadHandler对应的fireXXXX()方法；

（2）执行事件相关的逻辑操作；

以fireChannelActive方法为例，调用head.fireChannelActive()之后，实际上调用的是AbstractChannelHandlerContext.invokeChannelActive(head) 静态方法。

public final ChannelPipeline fireChannelActive() {
        AbstractChannelHandlerContext.invokeChannelActive(head);
        return this;
    }

先看看AbstractChannelHandlerContext.invokeChannelActive(head)源码：

static void invokeChannelActive(final AbstractChannelHandlerContext next) {
        //获取线程执行器
        EventExecutor executor = next.executor();
        //
        if (executor.inEventLoop()) {
            next.invokeChannelActive();
        } else {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    next.invokeChannelActive();
                }
            });
        }
    }

上面的源码可知，最后都会调用`AbstractChannelHandlerContext.invokeChannelActive();` 这个方法。  
看看这个方法的源码：

private void invokeChannelActive() {
        if (invokeHandler()) {
            try {
                ((ChannelInboundHandler) handler()).channelActive(this);
            } catch (Throwable t) {
                notifyHandlerException(t);
            }
        } else {
            fireChannelActive();
        }
    }

当handler已经被添加到pipeline中时，就直接获取ChannelInboundHandler对象，然后调用handler中的channelActive()函数。

当handler还没有被添加进来，就会执行下一个handler。

·outbound事件  
由用户线程发起的IO操作成为outbound事件，比如connect(), write()等操作。pipeline自身并不直接进行IO操作，最终都是由Unsafe和Channel来实现真正的IO操作的。

# 3. ChannelHandler功能说明 #

ChannelHandler就类似于servlet中的Filter过滤器，负责对IO事件进行拦截和处理。基于Handler接口用户可以很方便的进行业务定制，例如打印日志、同一封装异常信息，性能统计和消息编解码等。

Channelhandler支持注解，目前支持的注解有两种：  
1）Sharable：多个ChannelPipeline共用同一个ChannelHandler；

2）Skip：被Skip注解的方法不会被调用，直接忽略。

关于ChannelHandler 我们可以直接继承ChannelInboundHandlerAdapter定义入站事件，继承于ChannelOutboundHandlerAdapter定义出站事件。在netty中定义了很多的编解码器也是基于ChannelHandler实现的，关于这两部分的知识，可以参考另外的博客：  
[ChannelHandler的介绍][netty_ChannelHandler][http://blog.csdn.net/u010853261/article/details/54574440][netty_ChannelHandler]

[netty的编解码器][netty]  
[http://blog.csdn.net/u010853261/article/details/54575626][netty]

# 4. ChannelHandler源码分析 #

对ChannelHandler的源码分析其实也就是对一些编解码器的源码的分析，本文就不再给出。

[SouthEast]: /images/20220710/097c39ae95b34917a4fbec90a65d0d03.png
[netty_ChannelHandler]: http://blog.csdn.net/u010853261/article/details/54574440
[SouthEast 1]: /images/20220710/45b5714a5dc2498790b77514c5ab8e6a.png
[netty]: http://blog.csdn.net/u010853261/article/details/54575626