【Netty】(8）---理解ChannelPipeline

时间  2020-06-29
标签 Netty 理解 channelpipeline 栏目 Netty 繁體版
原文   原文链接

ChannelPipeline

     ChannelPipeline不是单独存在,它确定会和Channel、ChannelHandler、ChannelHandlerContext关联在一块儿,因此有关概念这里一块儿讲。html

1、ChannelHandler

  一、概念

 先看图java

ChannelHandler下主要是两个子接口设计模式

      ChannelInboundHandler(入站): 处理输入数据和Channel状态类型改变。promise

                                      适配器: ChannelInboundHandlerAdapter（适配器设计模式）异步

                                      经常使用的: SimpleChannelInboundHandleride

    ChannelOutboundHandler(出站): 处理输出数据函数

                                      适配器: ChannelOutboundHandlerAdapteroop

每个Handler都必定会处理出站或者入站(可能二者都处理数据),例如对于入站的Handler可能会继承SimpleChannelInboundHandler或者ChannelInboundHandlerAdapter,学习

而SimpleChannelInboundHandler又是继承于ChannelInboundHandlerAdapter，最大的区别在于SimpleChannelInboundHandler会对没有外界引用的资源进行必定的清理,this

而且入站的消息能够经过泛型来规定。

这里为何有设配器模式呢？

      咱们在写自定义Handel时候,不多会直接实现上面两个接口,由于接口中有不少默认方法须要实现,因此这里就采用了设配器模式,ChannelInboundHandlerAdapter和

ChannelInboundHandlerAdapter就是设配器模式的产物,让它去实现上面接口,实现它全部方法。那么你本身写自定义Handel时,只要继承它,就无须重写上面接口的全部方法了。

 二、Channel 生命周期(执行顺序也是从上倒下)

   （1）channelRegistered: channel注册到一个EventLoop。

   （2）channelActive: 变为活跃状态（链接到了远程主机），能够接受和发送数据

   （3）channelInactive: channel处于非活跃状态，没有链接到远程主机

   （4）channelUnregistered: channel已经建立，可是未注册到一个EventLoop里面，也就是没有和Selector绑定

 三、ChannelHandler 生命周期

      handlerAdded： 当 ChannelHandler 添加到 ChannelPipeline 调用

  handlerRemoved： 当 ChannelHandler 从 ChannelPipeline 移除时调用

 exceptionCaught： 当 ChannelPipeline 执行抛出异常时调用

 

2、ChannelPipeline

  一、概念

先看图

       ChannelPipeline类是ChannelHandler实例对象的链表，用于处理或截获通道的接收和发送数据。它提供了一种高级的截取过滤模式（相似serverlet中的filter功能），让用

户能够在ChannelPipeline中彻底控制一个事件以及如何处理ChannelHandler与ChannelPipeline的交互。

       对于每一个新的通道Channel，都会建立一个新的ChannelPipeline，并将器pipeline附加到channel中。

下图描述ChannelHandler与pipeline中的关系，一个io操做能够由一个ChannelInboundHandler或ChannelOutboundHandle进行处理，并经过调用ChannelInboundHandler

处理入站io或经过ChannelOutboundHandler处理出站IO。

二、经常使用方法

      addFirst(...)   //添加ChannelHandler在ChannelPipeline的第一个位置
     addBefore(...)   //在ChannelPipeline中指定的ChannelHandler名称以前添加ChannelHandler
      addAfter(...)   //在ChannelPipeline中指定的ChannelHandler名称以后添加ChannelHandler
       addLast(...)   //在ChannelPipeline的末尾添加ChannelHandler
        remove(...)   //删除ChannelPipeline中指定的ChannelHandler
       replace(...)   //替换ChannelPipeline中指定的ChannelHandler

ChannelPipeline能够动态添加、删除、替换其中的ChannelHandler，这样的机制能够提升灵活性。示例:

1.    ChannelPipeline pipeline = ch.pipeline(); 
2.    FirstHandler firstHandler = new FirstHandler(); 
3.    pipeline.addLast("handler1", firstHandler); 
4.    pipeline.addFirst("handler2", new SecondHandler()); 
5.    pipeline.addLast("handler3", new ThirdHandler()); 
6.    pipeline.remove("“handler3“"); 
7.    pipeline.remove(firstHandler); 
8.    pipeline.replace("handler2", "handler4", new FourthHandler());<span style="font-family:Microsoft YaHei;font-size:14px;">

 三、入站出站Handler执行顺序

  通常的项目中，inboundHandler和outboundHandler有多个，在Pipeline中的执行顺序？

   重点记住: InboundHandler顺序执行，OutboundHandler逆序执行。

   问题: 下面的handel的执行顺序？

          ch.pipeline().addLast(new InboundHandler1());
          ch.pipeline().addLast(new OutboundHandler1());
          ch.pipeline().addLast(new OutboundHandler2());
          ch.pipeline().addLast(new InboundHandler2());
  或者：
          ch.pipeline().addLast(new OutboundHandler1());
          ch.pipeline().addLast(new OutboundHandler2());
          ch.pipeline().addLast(new InboundHandler1());
          ch.pipeline().addLast(new InboundHandler2());

其实上面的执行顺序都是同样的:

 InboundHandler1--> InboundHandler2 -->OutboundHandler2 -->OutboundHandler1

结论

      1）InboundHandler顺序执行，OutboundHandler逆序执行

      2）InboundHandler之间传递数据，经过ctx.fireChannelRead(msg)

      3）InboundHandler经过ctx.write(msg)，则会传递到outboundHandler

      4)  使用ctx.write(msg)传递消息，Inbound须要放在结尾，在Outbound以后，否则outboundhandler会不执行；

           可是使用channel.write(msg)、pipline.write(msg)状况会不一致，都会执行,那是由于channel和pipline会贯穿整个流。

      5)  outBound和Inbound谁先执行，针对客户端和服务端而言，客户端是发起请求再接受数据，先outbound再inbound，服务端则相反。

 

3、ChannelHandlerContext

    ChannelPipeline并非直接管理ChannelHandler，而是经过ChannelHandlerContext来间接管理，这一点经过ChannelPipeline的默认实现DefaultChannelPipeline能够看出来。

    DefaultChannelHandlerContext和DefaultChannelPipeline是ChannelHandlerContext和ChannelPipeline的默认实如今DefaultPipeline内部

DefaultChannelHandlerContext组成了一个双向链表。 咱们看下DefaultChannelPipeline的构造函数:

/**
  * 能够看到,DefaultChinnelPipeline 内部使用了两个特殊的Hander 来表示Handel链的头和尾。
  */
 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它的Handel头部和尾部的Handel是固定的,咱们所添加的Handel是添加在这个头和尾以前的Handel。（下面这个图更加清晰）

 

4、几者关系

先大体说下什么是Channel

一般来讲, 全部的 NIO 的 I/O 操做都是从 Channel 开始的. 一个 channel 相似于一个 stream。在Netty中,Channel是客户端和服务端创建的一个链接通道。

虽然java Stream 和 NIO Channel都是负责I/O操做,但他们仍是有许多区别的:

      1）咱们能够在同一个 Channel 中执行读和写操做, 然而同一个 Stream 仅仅支持读或写。

     2）Channel 能够异步地读写, 而 Stream 是阻塞的同步读写。

     3）Channel 老是从 Buffer 中读取数据, 或将数据写入到 Buffer 中。

 几者的关系图以下:

总结:

      一个Channel包含一个ChannelPipeline，建立Channel时会自动建立一个ChannelPipeline，每一个Channel都有一个管理它的pipeline，这关联是永久性的。

这点从源码中就能够看出,我以前写的博客里有说到:【Netty】5 源码 Bootstrap。每个ChannelPipeline中能够包含多个ChannelHandler。全部ChannelHandler

都会顺序加入到ChannelPipeline中,ChannelHandler实例与ChannelPipeline之间的桥梁是ChannelHandlerContext实例。

 

5、整个传播流程

     如今将上面的整个传播流程,经过源码大体走一遍。

     为了搞清楚事件如何在Pipeline里传播，让咱们从Channel的抽象子类AbstractChannel开始，下面是AbstractChannel#write()方法的实现：

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {
    // ...
    @Override
    public Channel write(Object msg) {
        return pipeline.write(msg);
    }
    // ...
}

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

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

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

由于write是个outbound事件，因此DefaultChannelPipeline直接找到tail部分的context，调用其write()方法：

 

接着看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) {
        DefaultChannelHandlerContext next = findContextOutbound();
        next.invokeWrite(msg, promise);
        if (flush) {
            next.invokeFlush();
        }
    }
 
    private DefaultChannelHandlerContext findContextOutbound() {
        DefaultChannelHandlerContext ctx = this;
        do {
            ctx = ctx.prev;
        } while (!ctx.outbound);
        return ctx;
    }
 
    private void invokeWrite(Object msg, ChannelPromise promise) {
        try {
            ((ChannelOutboundHandler) handler).write(this, msg, promise);
        } catch (Throwable t) {
            notifyOutboundHandlerException(t, promise);
        }
    }
 
    // ...
}

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

 

  invokeWrite()接着调用handler的write()方法

 

最后看看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链继续传播：

 

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

 

参考 

   Netty4学习笔记-- ChannelPipeline（很是感谢做者分享,让我对事件传播有了更清晰的认识）

 

 

若是一我的充满快乐，正面的思想，那么好的人事物就会和他共鸣，并且被他吸引过来。一样，一我的老带悲伤，倒霉的事情也会跟过来。

                                                                                    ——在本身心情低落的时候，告诫本身不要把负能量带给别人。（大校18） 

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程