Netty中ChannelPipeline实际上类似与一条数据管道，负责传递Channel中读取的消息，它本质上是基于责任链模式的设计与实现，无论是IO事件的拦截器，还是用户自定义的ChannelHandler业务逻辑都做为一个个节点被添加到任务链上。

 ChannelPipeline中做为Netty中的数据管道，作用就是通过控制与联通不同的ChannelHandler，传递Channel中的消息。每一个Channel，都对应一个ChannelPipeline作为ChannelHandler的容器，而ChannelHandlerContext则把ChannelHandler的封装成每个节点，以双向链表方式在容器中存在；我们可以通过下图简单看下它们之间的关系。

 使用过Netty的朋友们都清楚，ChannelHandler就是做为拦截器和业务处理逻辑的存在，它会处理Channel中读写的消息；

首先看下ChannelHandler接口的定义

  1. public interface ChannelHandler {
  3.     //当ChannelHandler添加到Pipeline中时调用
  4.     void handlerAdded(ChannelHandlerContext ctx) throws Exception;
  6.     //当ChannelHandler在Pipeline中被移除时调用
  7.     void handlerRemoved(ChannelHandlerContext ctx) throws Exception;
  9.     //在运行过程中 发生异常时调用
  10.     @Deprecated
  11.     void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception;
  13.     @Inherited
  14.     @Documented
  15.     @Target(ElementType.TYPE)
  16.     @Retention(RetentionPolicy.RUNTIME)
  17.     @interface Sharable {
  18.         // no value
  19.     }
  20. }

ChannelHandler在处理或拦截IO操作时，分为出站和入站两个方向，对应channelde读写两个操作,所以Netty中又从ChannelHandler中派生出入站ChannelInboundHandler和出站ChannelOutboundHandler两个接口

ChannelInboundHandler处理入站数据以及各种状态的变化，下面列出了ChannelInboundHandler中数据被接收或者Channel状态发生变化时被调用的方法，这些方法和Channel的生命周期密切相关

  1. public interface ChannelInboundHandler extends ChannelHandler {
  2.     //当Channel注册对应的EventLoop并且能够处理I/O操作是被调用
  3.     void channelRegistered(ChannelHandlerContext ctx) throws Exception;
  5.     //当Channel从它对应的EventLoop上注销，并且无法处理I/O操作是被调用
  6.     void channelUnregistered(ChannelHandlerContext ctx) throws Exception;
  8.     //当Channel已经连接时被调用
  9.     void channelActive(ChannelHandlerContext ctx) throws Exception;
  11.     //当Channel为非活动状态，也就是断开时被调用
  12.     void channelInactive(ChannelHandlerContext ctx) throws Exception;
  14.     //当从Channel读取数据时被调用
  15.     void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception;
  17.     //当Channel的上一个读数据完成后被调用
  18.     void channelReadComplete(ChannelHandlerContext ctx) throws Exception;
  20.     //当调用fireUserEventTriggered方法时被调用
  21.     void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception;
  23.     //当Channel的可写状态发生改变时被调用。用户可以确保写操作不会完成太快
  24.     void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception;
  26.     @Override
  27.     @SuppressWarnings("deprecation")
  28.     //入站操作发生异常时调用
  29.     void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception;
  30. }

ChannelOutboundHandler处理出站操作和数据

  1. public interface ChannelOutboundHandler extends ChannelHandler {
  2.     //当请求将Channel绑定到本地地址时被调用
  3.     void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception;
  5.     //当请求将Channel连接到远程节点时被调用
  6.     void connect(
  7.             ChannelHandlerContext ctx, SocketAddress remoteAddress,
  8.             SocketAddress localAddress, ChannelPromise promise) throws Exception;
  10.     //当请求将Channel从远程节点断开时被调用
  11.     void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception;
  13.     //当请求关闭Channel时被调用
  14.     void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception;
  16.     //当请求从对应的EventLoop中注销时被调用
  17.     void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception;
  19.     //当请求从Channel读取数据时被调用
  20.     void read(ChannelHandlerContext ctx) throws Exception;
  22.     //当请求通过Channel将数据写到远程节点时被调用
  23.     void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception;
  25.     //当请求通过Channel将入队列数据冲刷到远程节点时被调用
  26.     void flush(ChannelHandlerContext ctx) throws Exception;
  27. }

ChannelHandlerContext可以说是ChannelPipeline的核心，它代表了ChannelHandler和ChannelPipeline之间的关联，我们首先要知道一个ChannelPipeline内部会维护一个双向链表，每当一个ChannelHandler被添加到ChannelPipeline中时，它都会被包装成为一个ChannelHandlerContext，组成链表的各个节点。

我们看下ChannelHandlerContext接口中定义的API接口

  1. public interface ChannelHandlerContext extends AttributeMap, ChannelInboundInvoker, ChannelOutboundInvoker {
  3.     /**
  4.      * Return the {@link Channel} which is bound to the {@link ChannelHandlerContext}.
  5.      */
  6.     //每个ChannelHandlerContext都会对一个Channel
  7.     Channel channel();
  9.     /**
  10.      * Returns the {@link EventExecutor} which is used to execute an arbitrary task.
  11.      */
  12.     //返回用于执行的EventExecutor任务
  13.     EventExecutor executor();
  15.     /**
  16.      * The unique name of the {@link ChannelHandlerContext}.The name was used when then {@link ChannelHandler}
  17.      * was added to the {@link ChannelPipeline}. This name can also be used to access the registered
  18.      * {@link ChannelHandler} from the {@link ChannelPipeline}.
  19.      */
  20.     //返回定义的name名称
  21.     String name();
  23.     /**
  24.      * The {@link ChannelHandler} that is bound this {@link ChannelHandlerContext}.
  25.      */
  26.     ChannelHandler handler();
  28.     /**
  29.      * Return {@code true} if the {@link ChannelHandler} which belongs to this context was removed
  30.      * from the {@link ChannelPipeline}. Note that this method is only meant to be called from with in the
  31.      * {@link EventLoop}.
  32.      */
  33.     //如果绑定到ChannelPipeline的ChannelHandler被删除，返回true
  34.     boolean isRemoved();
  36.     //触发下一个ChannelInboundHandler中fireChannelRegistered方法
  37.     @Override
  38.     ChannelHandlerContext fireChannelRegistered();
  39.     //触发下一个ChannelInboundHandler中fireChannelUnregistered方法
  40.     @Override
  41.     ChannelHandlerContext fireChannelUnregistered();
  42.     //触发下一个ChannelInboundHandler中fireChannelActive方法
  43.     @Override
  44.     ChannelHandlerContext fireChannelActive();
  45.     //触发下一个ChannelInboundHandler中fireChannelInactive方法
  46.     @Override
  47.     ChannelHandlerContext fireChannelInactive();
  48.     //触发下一个ChannelInboundHandler中fireExceptionCaught方法
  49.     @Override
  50.     ChannelHandlerContext fireExceptionCaught(Throwable cause);
  51.     //触发下一个ChannelInboundHandler中fireUserEventTriggered方法
  52.     @Override
  53.     ChannelHandlerContext fireUserEventTriggered(Object evt);
  54.     //触发下一个ChannelInboundHandler中fireChannelRead方法
  55.     @Override
  56.     ChannelHandlerContext fireChannelRead(Object msg);
  57.     //触发下一个ChannelInboundHandler中fireChannelReadComplete方法
  58.     @Override
  59.     ChannelHandlerContext fireChannelReadComplete();
  60.     //触发下一个ChannelInboundHandler中fireChannelWritabilityChanged方法
  61.     @Override
  62.     ChannelHandlerContext fireChannelWritabilityChanged();
  63.     //触发下一个ChannelInboundHandler中channelRead方法，如果是最后一个ChannelInboundHandler，则读取完成后触发channelReadComplete
  64.     @Override
  65.     ChannelHandlerContext read();
  66.     //触发下一个ChannelOutboundHandler中flush方法
  67.     @Override
  68.     ChannelHandlerContext flush();
  70.     /**
  71.      * Return the assigned {@link ChannelPipeline}
  72.      */
  73.     ChannelPipeline pipeline();
  75.     /**
  76.      * Return the assigned {@link ByteBufAllocator} which will be used to allocate {@link ByteBuf}s.
  77.      */
  78.     //返回绑定该channel 的 ByteBufAllocator
  79.     ByteBufAllocator alloc();
  81.     /**
  82.      * @deprecated Use {@link Channel#attr(AttributeKey)}
  83.      */
  84.     @Deprecated
  85.     @Override
  86.     //返回Attribute
  87.     <T> Attribute<T> attr(AttributeKey<T> key);
  89.     /**
  90.      * @deprecated Use {@link Channel#hasAttr(AttributeKey)}
  91.      */
  92.     @Deprecated
  93.     @Override
  94.     //是否包含指定的AttributeKey
  95.     <T> boolean hasAttr(AttributeKey<T> key);
  96. } 

在AbstractChannel的构造函数中我们可以看到对ChannelPipeline的初始化

  1.     protected AbstractChannel(Channel parent) {
  2.         this.parent = parent;
  3.         id = newId();
  4.         unsafe = newUnsafe();
  5.         pipeline = newChannelPipeline();//初始化ChannelPipeline
  6.     }

看下newChannelPipeline()内部的实现

  1.     protected DefaultChannelPipeline newChannelPipeline() {
  2.         return new DefaultChannelPipeline(this);
  3.     }

在这里创建了一个DefaultChannelPipeline 对象，并传入Channel对象。DefaultChannelPipeline 实现了ChannelPipeline的接口

进入DefaultChannelPipeline类内部，看下其具体构造

  1.     protected DefaultChannelPipeline(Channel channel) {
  2.         this.channel = ObjectUtil.checkNotNull(channel, "channel");
  3.         succeededFuture = new SucceededChannelFuture(channel, null);
  4.         voidPromise =  new VoidChannelPromise(channel, true);
  6.         tail = new TailContext(this);//定义一个头部节点
  7.         head = new HeadContext(this);//定义一个尾部节点
  9.         //连接头尾节点，构成双向链表
  10.         head.next = tail;
  11.         tail.prev = head;
  12.     }

在这里我们可以看到DefaultChannelPipeline内部通过声明头尾两个Context节点对象，构建了一个双向链表结构我们；其实这里的TailContext与HeadContext都是ChannelHandlerContext接口的具体实现;

通过上面的内容，我们可以看出ChannelPipeline就是一个用于拦截Channel入站和出站事件的ChannelHandler实例链，而ChannelHandlerContext就是这个实例链上的节点，每一个新创建的Channel都会被分配一个新的ChannelPipeline。这篇文章我们对ChannelPipeline的构造和设计进行了大概的总结，其中如有不足与不正确的地方还望指出与海涵。后面我会对ChannelPipeline中ChannelHandler的添加、删除等具体操作与事件如何在管道中流通传递进行具体的分析。

 

关注微信公众号，查看更多技术文章。

 

 

版权声明：本文为dafanjoy原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://www.cnblogs.com/dafanjoy/p/11512304.html