Netty(ChannelHandler 和 ChannelPipeline)

淩亂°似流年 2022-04-17 01:42 268阅读 0赞

## ChannelHandler家族 ##

### channel的生命周期 ###

Interface Channel定义了一组和ChannelInboundHandler API密切相关的简单但功能强大的状态模型，其Channel主要有4个状态。

<table> 
 <tbody> 
  <tr> 
   <td> <p>状态</p> </td> 
   <td> <p>描述</p> </td> 
  </tr> 
  <tr> 
   <td> <p>ChannelUnregistered</p> </td> 
   <td> <p>Channel 已经被创建，但还未注册到EventLoop</p> </td> 
  </tr> 
  <tr> 
   <td> <p>ChannelRegistered</p> </td> 
   <td> <p>Channel 已经被注册到了EventLoop</p> </td> 
  </tr> 
  <tr> 
   <td> <p>ChannelActive</p> </td> 
   <td> <p>Channel 处于活动状态（已经连接到它的远程节点）。它现在可以接收和发送数据了</p> </td> 
  </tr> 
  <tr> 
   <td> <p>ChannelInactive</p> </td> 
   <td> <p>Channel 没有连接到远程节点</p> </td> 
  </tr> 
 </tbody> 
</table>

当这些状态发生改变时，将会生成对应的事件。这些事件将会被转发给ChannelPipeline中的ChannelHandler，其可以随后对它们做出响应。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70][]

### ChannelHandler的生命周期 ###

interface ChannelHandler定义的生命周期操作如下所示，在ChannelHandler被添加到ChannelPipeline中或者被ChannelPipeline中移除时会调用 这些操作。这些方法中的每一个都接受一个ChannelHandlerContext参数。

<table> 
 <tbody> 
  <tr> 
   <td> <p>状态</p> </td> 
   <td> <p>描述</p> </td> 
  </tr> 
  <tr> 
   <td> <p>handlerAdded</p> </td> 
   <td> <p>当把 ChannelHandler 添加到 ChannelPipeline 中时被调用</p> </td> 
  </tr> 
  <tr> 
   <td> <p>handlerRemoved</p> </td> 
   <td> <p>当从 ChannelPipeline 中移除 ChannelHandler 时被调用</p> </td> 
  </tr> 
  <tr> 
   <td> <p>exceptionCaught</p> </td> 
   <td> <p>当处理过程中在 ChannelPipeline 中有错误时被调用</p> </td> 
  </tr> 
 </tbody> 
</table>

ChannelHandler有两个重要的子接口：

*  ChannelInboundHandler——处理入站数据以及各种状态变化；
 *  ChannelOutboundHandler——处理出站数据并且允许拦截所有的操作。

###  ChannelInboundHandler接口 ###

以下方法将会在数据被接收时或者 与其对应的Channel状态发生改变时被调用

/**
     * {@link ChannelHandler} which adds callbacks for state changes. This allows the user
     * to hook in to state changes easily.
     */
    public interface ChannelInboundHandler extends ChannelHandler {
    
        /**
         * 当Channel已经注册到它的EventLoop并且能够处理I/O时被调用
         */
        void channelRegistered(ChannelHandlerContext ctx) throws Exception;
    
        /**
         * 当Channel从它的EventLoop注销并且无法处理任何I/O时被调用
         */
        void channelUnregistered(ChannelHandlerContext ctx) throws Exception;
    
        /**
         * 当Channel处于活动状态时被调用；Channel已经连接/绑定并且已经就绪
         */
        void channelActive(ChannelHandlerContext ctx) throws Exception;
    
        /**
         * 当Channel离开活动状态并且不再连接它的远程节点时被调用
         */
        void channelInactive(ChannelHandlerContext ctx) throws Exception;
    
        /**
         * 当从Channel读取数据时被调用
         */
        void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception;
    
        /**
         * 当Channel上的一个读操作完成时被调用
         */
        void channelReadComplete(ChannelHandlerContext ctx) throws Exception;
    
        /**
         * 当ChannelnboundHandler.fireUserEventTriggered()方法被调用时被调用，因为一个POJO被传经了ChannelPipeline
         */
        void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception;
    
        /**
         * 当Channel的可写状态发生改变时被调用。用户可以确保写操作不会完成得太快（以避免发生OutOfMemoryError）或者可以在Channel变为再次可写时恢复写入。可以通过调用Channel的isWritable()方法来检测Channel的可写性。与可写性相关的阈值可以通过Channel.config().setWriteHighWaterMark()和Channel.config().setWriteLowWater-Mark()方法来设置
         */
        void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception;
    
        /**
         * 如果抛出一个可抛出的异常对象，则调用。
         */
        @Override
        @SuppressWarnings("deprecation")
        void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception;
    }

### ChannelOutboundHandler接口 ###

出站操作和数据将由ChannelOutboundHandler处理。它的方法将被Channel、ChannelPipeline以及ChannelHandlerContext调用。ChannelOutboundHandler的一个强大的功能是可以按需推迟操作或者事件，这使得可以通过一些复杂的方法来处理请求。例如，如果到远程节点的写入被暂停了，那么你可以推迟冲刷操作并在稍后继续。

public interface ChannelOutboundHandler extends ChannelHandler {
        /**
         * 当请求将Channel绑定到本地地址时被调用
         */
        void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception;
    
        /**
         * 当请求将Channel连接到远程节点时被调用
         */
        void connect(
                ChannelHandlerContext ctx, SocketAddress remoteAddress,
                SocketAddress localAddress, ChannelPromise promise) throws Exception;
    
        /**
         * 当请求将Channel从远程节点断开时被调用
         */
        void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception;
    
        /**
         * 当请求关闭Channel时被调用
         */
        void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception;
    
        /**
         * 当请求将Channel从它的EventLoop注销时被调用
         */
        void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception;
    
        /**
         * 当请求从Channel读取更多的数据时被调用
         */
        void read(ChannelHandlerContext ctx) throws Exception;
    
        /**
         * 当请求通过Channel将数据写到远程节点时被调用
         */
        void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception;
    
        /**
         * 当请求通过Channel将入队数据冲刷到远程节点时被调用
         */
        void flush(ChannelHandlerContext ctx) throws Exception;
    }

**    ChannelPromise**与**ChannelFuture** ChannelOutboundHandler中的大部分方法都需要一个ChannelPromise参数，以便在操作完成时得到通知。ChannelPromise是ChannelFuture的一个子类，其定义了一些可写的方法，如setSuccess()和setFailure()，从而使ChannelFuture不可变。

### ChannelHandler适配器 ###

**ChannelInboundHandlerAdapter**和**ChannelOutboundHandlerAdapter**类作为自己的ChannelHandler的**适配器**类。这两个适配器分别提供了ChannelInboundHandler和ChannelOutboundHandler的基本实现。通过扩展抽象类ChannelHandlerAdapter，它们获得了它们共同的超接口ChannelHandler的方法。生成的类的层次结构如：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 1][]

ChannelHandlerAdapter还提供了实用方法isSharable()。如果其对应的实现被标注为Sharable，那么 这个方法将返回true，表示它可以被添加到多个ChannelPipeline中。（@Sharable注解）

在ChannelInboundHandlerAdapter和ChannelOutboundHandlerAdapter中所提供的方法体调用了其相关联的ChannelHandlerContext上的等效方法，从而将事件转发到了ChannelPipeline中的下一个ChannelHandler中。

### 资源管理 ###

为了诊断潜在的（资源泄漏）问题，Netty提供了class ResourceLeakDetector它将对你应用程序的缓冲区分配做大约1%的采样来检测内存泄露。相关的开销是非常小的。

Netty目前定义了4种泄漏检测级别：

<table> 
 <tbody> 
  <tr> 
   <td> <p>级别</p> </td> 
   <td> <p>描述</p> </td> 
  </tr> 
  <tr> 
   <td> <p>DISABLED</p> </td> 
   <td> <p>禁用泄漏检测。只有在详尽的测试之后才应使用</p> </td> 
  </tr> 
  <tr> 
   <td> <p>SIMPLE</p> </td> 
   <td> <p>使用1%的默认采样率检测并报告任何发现的泄露。这是默认级别，适合绝大部分情况。</p> </td> 
  </tr> 
  <tr> 
   <td> <p>ADVANCED</p> </td> 
   <td> <p>使用默认的采样率，报告所发现的任何的泄露以及对应的消息被访问的位置</p> </td> 
  </tr> 
  <tr> 
   <td> <p>PARANOID</p> </td> 
   <td> <p>类似于ADVANCED，但是其将会对每次（对消息的）访问都进行采样。这对性能将会有很大影响，在调试阶段使用</p> </td> 
  </tr> 
 </tbody> 
</table>

泄露检测级别可以通过将下面的Java系统属性设置为表中的一个值来定义：

java-Dio.netty.leakDetectionLevel=ADVANCED

如果一个消息被消费或者丢弃了，并且没有传递给ChannelPipeline中的下一个ChannelOutboundHandler，那么用户就有责任调用ReferenceCountUtil.release()。如果消息到达了实际的传输层，那么当它被写入时或者Channel关闭时，都将被自动释放。

![20181111145555354.png][]

## ChannelPipeline接口 ##

ChannelPipeline是一个拦截流经Channel的入站和出站事件的Channel-Handler实例链，那么就很容易看出这些ChannelHandler之间的交互是组成一个应用程序数据和事件处理逻辑的核心。

每一个新创建的Channel都将会被分配一个新的ChannelPipeline。这项关联是永久性的；Channel既不能附加另外一个ChannelPipeline，也不能分离其当前的。在Netty组件的生命周期中，这是一项固定的操作，不需要开发人员的任何干预。

根据事件的起源，事件将会被ChannelInboundHandler或者ChannelOutboundHandler处理。随后，通过调用ChannelHandlerContext实现，它将被转发给同一超类型的下一个ChannelHandler。

**ChannelHandlerContext：**

ChannelHandlerContext使得ChannelHandler能够和它的ChannelPipeline以及其他的ChannelHandler交互。ChannelHandler可以通知其所属的ChannelPipeline中的下一个ChannelHandler，甚至可以动态修改它所属的ChannelPipeline。         ChannelHandlerContext具有丰富的用于处理事件和执行I/O 操作的API。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 2][]

当你完成了通过调用ChannelPipeline.add\*()方法将入站处理器（ChannelInboundHandler）和出站处理器（ChannelOutboundHandler）混合添加到ChannelPipeline之后，每一个ChannelHandler从头部到尾端的顺序位置正如同我们方才所定义它们的一样。因此，如果你将图中的处理器（ChannelHandler）从左到右进行编号，那么第一个被入站事件看到的ChannelHandler将是1，而第一个被出站事件看到的ChannelHandler将是 5。

在ChannelPipeline传播事件时，它会测试ChannelPipeline中的下一个Channel-Handler的类型是否和事件的运动方向相匹配。如果不匹配，ChannelPipeline将跳过该ChannelHandler并前进到下一个，直到它找到和该事件所期望的方向相匹配的为止。

### 修改 ChannelPipeline ###

ChannelHandler可以通过添加、删除或者替换其他的ChannelHandler来实时地修改ChannelPipeline的布局。

<table> 
 <tbody> 
  <tr> 
   <td> <p>名称</p> </td> 
   <td> <p>描述</p> </td> 
  </tr> 
  <tr> 
   <td> <p>&nbsp;addFirst</p> <p>&nbsp;addBefore</p> <p>&nbsp;addAfter</p> <p>&nbsp;addLast</p> </td> 
   <td> <p>将一个ChannelHandler 添加到ChannelPipeline 中</p> </td> 
  </tr> 
  <tr> 
   <td> <p>remove</p> </td> 
   <td> <p>将一个ChannelHandler 从 ChannelPipeline 中移除</p> </td> 
  </tr> 
  <tr> 
   <td> <p>replace</p> </td> 
   <td> <p>将 ChannelPipeline 中的一个 ChannelHandler 替换为另一个ChannelHandler</p> </td> 
  </tr> 
 </tbody> 
</table>

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 3][]

**ChannelHandler的执行和阻塞 **

*通常ChannelPipeline中的每一个ChannelHandler都是通过它的EventLoop（I/O 线程）来处理传递给它的事件的。所以至关重要的是不要阻塞这个线程，因为这会对整体的I/O 处理产生负面的影响。但有时可能需要与那些使用阻塞API  的遗留代码进行交互。对于这种情况，ChannelPipeline有一些接受一个EventExecutorGroup的add()方法。如果一个事件被传递给一个自定义的EventExecutorGroup，它将被包含在这个EventExecutorGroup中的某个EventExecutor所处理，从而被从该Channel本身的EventLoop中移除。对于这种用例，Netty提供了一个叫DefaultEventExecutor-Group的默认实现。*

通过类型或者名称来访问ChannelHandler的方法：

<table> 
 <tbody> 
  <tr> 
   <td> <p>名称</p> </td> 
   <td> <p>描述</p> </td> 
  </tr> 
  <tr> 
   <td> <p>&nbsp;get</p> </td> 
   <td> <p>&nbsp;通过类型或者名称返回 ChannelHandler</p> </td> 
  </tr> 
  <tr> 
   <td> <p>&nbsp;context</p> </td> 
   <td> <p>&nbsp;返回和ChannelHandler 绑定的 ChannelHandlerContext</p> </td> 
  </tr> 
  <tr> 
   <td> <p>&nbsp;names</p> </td> 
   <td> <p>&nbsp;返回ChannelPipeline 中所有 ChannelHandler 的名称</p> </td> 
  </tr> 
 </tbody> 
</table>

### 触发事件  ###

ChannelPipeline的API公开了用于调用入站和出站操作的附加方法：

ChannelPipeline的入站操作：

<table style="width:636px;"> 
 <tbody> 
  <tr> 
   <td>方法名称</td> 
   <td>描述</td> 
  </tr> 
  <tr> 
   <td>fireChannelRegistered</td> 
   <td>调用ChannelPipeline中下一个ChannelInboundHandler的channelRegistered(ChannelHandlerContext)方法</td> 
  </tr> 
  <tr> 
   <td>fireChannelUnregistered</td> 
   <td>调用ChannelPipeline中下一个ChannelInboundHandler的fireChannelUnregistered(ChannelHandlerContext)方法</td> 
  </tr> 
  <tr> 
   <td>fireChannelActive</td> 
   <td>调用ChannelPipeline中下一个ChannelInboundHandler的fireChannelActive(ChannelHandlerContext)方法</td> 
  </tr> 
  <tr> 
   <td>fireChannelInactive</td> 
   <td>调用ChannelPipeline中下一个ChannelInboundHandler的fireChannelInactive(ChannelHandlerContext)方法</td> 
  </tr> 
  <tr> 
   <td>fireExceptionCaught</td> 
   <td>调用ChannelPipeline中下一个ChannelInboundHandler的fireExceptionCaught(ChannelHandlerContext)方法</td> 
  </tr> 
  <tr> 
   <td>fireUserEventTriggered</td> 
   <td>调用ChannelPipeline中下一个ChannelInboundHandler的fireUserEventTriggered(ChannelHandlerContext)方法</td> 
  </tr> 
  <tr> 
   <td>fireChannelRead</td> 
   <td>调用ChannelPipeline中下一个ChannelInboundHandler的fireChannelRead(ChannelHandlerContext)方法</td> 
  </tr> 
  <tr> 
   <td>fireChannelReadComplete</td> 
   <td>调用ChannelPipeline中下一个ChannelInboundHandler的fireChannelReadComplete(ChannelHandlerContext)方法</td> 
  </tr> 
  <tr> 
   <td>fireChannelWritabilityChanged</td> 
   <td>调用ChannelPipeline中下一个ChannelInboundHandler的fireChannelWritabilityChanged(ChannelHandlerContext)方法</td> 
  </tr> 
 </tbody> 
</table>

在出站这边，处理事件将会导致底层的套接字上发生一系列的动作。

ChannelPipeline的出站操作：

<table style="width:700px;"> 
 <tbody> 
  <tr> 
   <td>方法名称</td> 
   <td>描述</td> 
  </tr> 
  <tr> 
   <td>bind</td> 
   <td>将Channel绑定到一个本地地址，这将调用ChannelPipeline中的下一个ChannelOutboundHandler的bind(ChannelHandlerContext,Socket-Address, ChannelPromise)方法</td> 
  </tr> 
  <tr> 
   <td>connect</td> 
   <td>将Channel连接到一个远程地址，这将调用ChannelPipeline中的下一个ChannelOutboundHandler的connect(ChannelHandlerContext, SocketAddress, ChannelPromise)方法</td> 
  </tr> 
  <tr> 
   <td>disconnect</td> 
   <td>将Channel断开连接&nbsp; 。这将调用ChannelPipeline中的下一个ChannelOutboundHandler的disconnect(ChannelHandlerContext, ChannelPromise)方法</td> 
  </tr> 
  <tr> 
   <td>close</td> 
   <td>将Channel关闭。 这将调用ChannelPipeline中的下一个ChannelOutboundHandler的close(ChannelHandlerContext, ChannelPromise)方法</td> 
  </tr> 
  <tr> 
   <td>deregister</td> 
   <td>将Channel从它先前所分配的EventExecutor（即EventLoop）中注销。这将调用ChannelPipeline中的下一个ChannelOutboundHandler的deregister(ChannelHandlerContext, ChannelPromise)方法</td> 
  </tr> 
  <tr> 
   <td>flush</td> 
   <td>冲刷Channel所有挂起的写入。这将调用ChannelPipeline中的下一个ChannelOutboundHandler的flush(ChannelHandlerContext)方法</td> 
  </tr> 
  <tr> 
   <td>write</td> 
   <td>将消息写入Channel。这将调用ChannelPipeline中的下一个ChannelOutboundHandler的write(ChannelHandlerContext, Object msg, Channel-Promise)方法。注意：这并不会将消息写入底层的Socket，而只会将它放入队列中。要将它&nbsp; 写入Socket，需要调用flush()或者writeAndFlush()方法</td> 
  </tr> 
  <tr> 
   <td>writeAndFlush</td> 
   <td>这是一个先调用write()方法再接着调用flush()方法的便利方法</td> 
  </tr> 
  <tr> 
   <td>read</td> 
   <td>请求从Channel中读取更多的数据。这将调用ChannelPipeline中的下一个ChannelOutboundHandler的read(ChannelHandlerContext)方法</td> 
  </tr> 
 </tbody> 
</table>

总结：

*  ChannelPipeline保存了与Channel相关联的ChannelHandler；
 *  ChannelPipeline可以根据需要，通过添加或者删除ChannelHandler来动态地修改；
 *  ChannelPipeline有着丰富的API用以被调用，以响应入站和出站事件。

## ChannelHandlerContext接口  ##

ChannelHandlerContext代表了ChannelHandler和ChannelPipeline之间的关联，每当有ChannelHandler添加到ChannelPipeline中时，都会创建ChannelHandlerContext。ChannelHandlerContext的主要功能是管理它所关联的ChannelHandler和在同一个ChannelPipeline中的其他ChannelHandler之间的交互。

ChannelHandlerContext有很多的方法，其中一些方法也存在于Channel和ChannelPipeline本身上，但是有一点重要的不同。如果调用Channel或者ChannelPipeline上的这些方法，它们将沿着整个ChannelPipeline进行传播。而调用位于ChannelHandlerContext上的相同方法，则将从当前所关联的ChannelHandler开始，并且只会传播给位于该ChannelPipeline中的下一个能够处理该事件的ChannelHandler。

public interface ChannelHandlerContext extends AttributeMap, ChannelInboundInvoker, ChannelOutboundInvoker {
    
        /**
         * 返回绑定到这个实例的Channel
         */
        Channel channel();
    
        /**
         * 返回调度事件的EventExecutor
         */
        EventExecutor executor();
    
        /**
         * 返回这个实例的唯一名称
         */
        String name();
    
        /**
         * 返回绑定到这个实例的ChannelHandler
         */
        ChannelHandler handler();
    
        /**
         * 如果所关联的ChannelHandler已经被从ChannelPipeline中移除则返回true
         */
        boolean isRemoved();
    
        /**
         * 触发对下一个ChannelInboundHandler上的fireChannelRegistered()方法的调用
         */
        @Override
        ChannelHandlerContext fireChannelRegistered();
    
        /**
         * 触发对下一个ChannelInboundHandler上的fireChannelUnregistered()方法的调用
         */
        @Override
        ChannelHandlerContext fireChannelUnregistered();
    
        /**
         * 触发对下一个ChannelInboundHandler上的fireChannelActive()方法的调用
         */
        @Override
        ChannelHandlerContext fireChannelActive();
    
        /**
         * 触发对下一个ChannelInboundHandler上的fireChannelInactive()方法的调用
         */
        @Override
        ChannelHandlerContext fireChannelInactive();
    
        /**
         * 触发对下一个ChannelInboundHandler上的fireExceptionCaught()方法的调用
         */
        @Override
        ChannelHandlerContext fireExceptionCaught(Throwable cause);
    
        /**
         * 触发对下一个ChannelInboundHandler上的fireUserEventTriggered()方法的调用
         */
        @Override
        ChannelHandlerContext fireUserEventTriggered(Object evt);
    
        /**
         * 触发对下一个ChannelInboundHandler上的fireChannelRead()方法的调用
         */
        @Override
        ChannelHandlerContext fireChannelRead(Object msg);
    
        /**
         * 触发对下一个ChannelInboundHandler上的fireChannelReadComplete()方法的调用
         */
        @Override
        ChannelHandlerContext fireChannelReadComplete();
    
        /**
         * 触发对下一个ChannelInboundHandler上的fireChannelWritabilityChanged()方法的调用
         */
        @Override
        ChannelHandlerContext fireChannelWritabilityChanged();
    
        /**
         * 将数据从Channel读取到第一个入站缓冲区；如果读取成功则触发一个channelRead事件，并（在最后一个消息被读取完成后）通知ChannelInboundHandler的channelReadComplete(ChannelHandlerContext)方法
         */
        @Override
        ChannelHandlerContext read();
    
        /**
         * 刷新所有挂起的消息。
         */
        @Override
        ChannelHandlerContext flush();
    
        /**
         * 返回这个实例所关联的ChannelPipeline
         */
        ChannelPipeline pipeline();
    
        /**
         * 返回和这个实例相关联的Channel所配置的ByteBufAllocator
         */
        ByteBufAllocator alloc();
    
        /******************补充*************************/
        //write  通过这个实例写入消息并经过ChannelPipeline
        //writeAndFlush  通过这个实例写入并冲刷消息并经过ChannelPipeline
    
    }

使用ChannelHandlerContext的API的时候，请牢记以下两点：

*  ChannelHandlerContext和ChannelHandler之间的关联（绑定）是永远不会改变的，所以缓存对它的引用是安全的；
 *  相对于其他类的同名方法，ChannelHandlerContext的方法将产生更短的事件流，应该尽可能地利用这个特性来获得最大的性能。

###  使用 ChannelHandlerContext ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 4][]

修改客户端的ChildChannelHandler：

private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
    
            protected void initChannel(SocketChannel ch) throws Exception {
                System.out.println("客户端启动……");
                ByteBuf bufs= Unpooled.copiedBuffer("pipeline发送的数据->", Charset.forName("UTF-8"));
                ch.pipeline().write(bufs);//通过调用ChannelPipeline的write方法将数据写入通道，但是不刷新
                ch.pipeline().addLast("text",new ChannelInboundHandlerAdapter() {
                    @Override
                    public void channelActive(ChannelHandlerContext ctx) throws Exception {
                        ctx.channel().write(Unpooled.copiedBuffer("通过ChannelHandlerContext获取的channel发送的消息->",
                                Charset.forName("UTF-8")));//通过ChannelHandlerContext获取的channel发送的消息->
                        CompositeByteBuf messageBuf=Unpooled.compositeBuffer();
                        ByteBuf headerBuf=buf;
                        ByteBuf bodyBuf=buf;
                        messageBuf.addComponent(bodyBuf);//将ByteBuf实例追加到CompositeByteBuf
                        messageBuf.addComponent(headerBuf);
                        for (ByteBuf buf:messageBuf){//遍历所有ByteBuf
                            System.out.println(buf);
                            byte[] req = new byte[buf.readableBytes()];
                            buf.readBytes(req);
                            String body = new String(req, "UTF-8");
                            System.out.println("复合缓冲区："+body);
                        }
                        ctx.writeAndFlush(buf);
                    }
    
                    public void channelRead(ChannelHandlerContext ctx, Object msg)
                            throws Exception {
                        ByteBuf buf = (ByteBuf) msg;
                        ByteBuf copyBuf=((ByteBuf) msg).copy();
    //                    System.out.println(buf.refCnt());//返回此对象的引用计数。如果为0，则表示此对象已被释放。
    //                    buf.release();//释放引用计数对象
                        for (int i = 0; i < buf.capacity(); i++) {
                            byte b=buf.getByte(i);
                            if((char)b>='a'&&(char)b<='z'||(char)b>='A'&&(char)b<='Z'||(char)b==',')
                            System.out.println("i="+(char)b);
                        }
                        int i=buf.forEachByte(new ByteProcessor() {
                            @Override
                            public boolean process(byte value) throws Exception {
                                byte[] b=",".getBytes();
                                if (b[0]!=value)
                                    return true;
                                else
                                    return false;
                            }
                        });
                        System.out.println("i="+i+" value="+(char) buf.getByte(i));
                        ByteBuf sliced = buf.slice(0,2);
                        sliced.setByte(0,(byte)'h');
                        byte[] req = new byte[buf.readableBytes()];
                        buf.readBytes(req);
                        String body = new String(req, "UTF-8");
                        System.out.println(body);
                        ctx.fireChannelRead(copyBuf);
                    }
                });
                ch.pipeline().addLast("text2",new ChannelInboundHandlerAdapter(){
                    public void channelRead(ChannelHandlerContext ctx, Object msg)
                            throws Exception {
                        ByteBuf buf = (ByteBuf) msg;
                        byte[] req = new byte[buf.readableBytes()];
                        buf.readBytes(req);
                        String body = new String(req, "UTF-8");
                        System.out.println("text2:"+body);
                        ByteBuf bufs= Unpooled.copiedBuffer("test2发送的数据", Charset.forName("UTF-8"));
                        ctx.writeAndFlush(bufs);
                        ctx.close();
                    }
                });
    //            ch.pipeline().remove("text2");
            }
        }

![20181111180723371.png][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 5][]

被调用的Channel或ChannelPipeline上的write()方法将一直传播事件通过整个ChannelPipeline，但是在ChannelHandler的级别上，事件从一个ChannelHandler到下一个ChannelHandler的移动是由ChannelHandlerContext上的调用完成的。

要想调用从某个特定的ChannelHandler开始的处理过程，必须获取到在（ChannelPipeline）该ChannelHandler之前的ChannelHandler所关联的ChannelHandlerContext。这个ChannelHandlerContext将调用和它所关联的ChannelHandler之后的ChannelHandler。

消息将从下一个ChannelHandler开始流经ChannelPipeline，绕过了所有前面的ChannelHandler。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 6][]

因为一个ChannelHandler可以从  属于多个ChannelPipeline，所以它也可以绑定到多个ChannelHandlerContext实例。  对于这种用法   指在多个ChannelPipeline中共享同一个ChannelHandler，对应的ChannelHandler必须要使用**@Sharable**注解标注；否则，试图将它添加到多个ChannelPipeline时将会触发异常。显而易见，为了安全地被用于多个并发的Channel（即连接），这样的ChannelHandler必须是线程安全的。

**只应该在确定了你的ChannelHandler是线程安全的时才使用@Sharable注解。**

**在多个ChannelPipeline中安装同一个ChannelHandler的一个常见的原因是用于收集跨越多个Channel的统计信息。**

## 异常处理 ##

### 处理入站异常 ###

如果在处理入站事件的过程中有异常被抛出，那么它将从它在ChannelInboundHandler里被触发的那一点开始流经ChannelPipeline。要想处理这种类型的入站异常，你需要在你的ChannelInboundHandler实现中重写下面的方法。

@Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception {
            System.out.println("服务器异常..");
            cause.printStackTrace();
            ctx.close();
        }

因为异常将会继续按照入站方向流动（就像所有的入站事件一样），所以实现了前面所示逻辑的ChannelInboundHandler通常位于ChannelPipeline的最后。这确保了所有的入站异常都总是会被处理，无论它们可能会发生在ChannelPipeline中的什么位置。

*  ChannelHandler.exceptionCaught()的默认实现是简单地将当前异常转发给ChannelPipeline中的下一个ChannelHandler；
 *  如果异常到达了ChannelPipeline的尾端，它将会被记录为未被处理；
 *  要想定义自定义的处理逻辑，你需要重写exceptionCaught()方法。然后你需要决定是否需要将该异常传播出去。

### 处理出站异常  ###

用于处理出站操作中的正常完成以及异常的选项，都基于以下的通知机制。

*  每个出站操作都将返回一个ChannelFuture。注册到ChannelFuture的ChannelFutureListener将在操作完成时被通知该操作是成功了还是出错了  。
 *  几乎所有的ChannelOutboundHandler上的方法都会传入一个ChannelPromise的实例。作为ChannelFuture的子类，ChannelPromise也可以被分配用于异步通知的监听器。但是，ChannelPromise还具有提供立即通知的可写方法：

ChannelPromise setSuccess();  
   ChannelPromise setFailure(Throwable cause);

添加ChannelFutureListener只需要调用ChannelFuture实例上的addListener(ChannelFutureListener)方法，并且有两种不同的方式可以做到这一点。其中最常用的方式是，调用出站操作（如write()方法）所返回的ChannelFuture上的addListener()方法。

方式一：

ChannelFuture future=ctx.writeAndFlush(bufs);
                        future.addListener(new ChannelFutureListener() {
                            @Override
                            public void operationComplete(ChannelFuture future) throws Exception {
                                if (future.isSuccess())
                                    System.out.println("成功");
                                else{
                                    System.out.println("失败");
                                    future.cause().printStackTrace();
                                    future.channel().close();
                                }
                            }
                        });

方式二：ChannelFutureListener添加到即将作为参数传递给ChannelOutboundHandler的方法的ChannelPromise。

public class OutboundExceptionHandler extends ChannelOutboundHandlerAdapter {
        @Override
        public void
        write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
            promise.addListener(new ChannelFutureListener() {
                @Override
                public void
                operationComplete(ChannelFuture
                                          f) {
                    if (!f.isSuccess()) {
                        f.cause().printStackTrace();
                        f.channel().close();
                    }
                }
            });
        }
    }

**ChannelPromise的可写方法**

通过调用ChannelPromise上的setSuccess()和setFailure()方法，可以使一个操作的状态在ChannelHandler的方法返回给其调用者时便即刻被感知到。

参考《Netty实战》

附：

package netty.in.action;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.ByteBufAllocator;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    import java.nio.charset.Charset;
    
    public class EchoServer {
    
        final ByteBuf bufs= Unpooled.copiedBuffer("Hello,刘德华", Charset.forName("UTF-8"));
        public void bind(int port) throws Exception {
            //配置服务端的NIO线程组
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 100)
                        .childHandler(new ChildChannelHandler());
    
                // 绑定端口，同步等待成功
                ChannelFuture f=b.bind(port).sync();
    
                //等待服务端监听端口关闭
                f.channel().closeFuture().sync();
            } finally {
                //退出，释放线程池资源
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
        private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
    
            protected void initChannel(SocketChannel ch) throws Exception {
                System.out.println("服务端启动……");
                ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg)
                            throws Exception {
                        ByteBuf buf = (ByteBuf) msg;
                        if(buf.hasArray()){
                            byte[] array=buf.array();//返回该缓冲区的备份字节数组。
                            int offset=buf.arrayOffset()+buf.readerIndex();//计算第一个字节的偏移量
                            int length=buf.readableBytes();//获取可读字节数
                            String s=new String(array,offset,length);
                            System.out.println("s="+s);
                        }else{
                            byte[] array = new byte[buf.readableBytes()];//获取可读字节数并分配一个新的数组来保存
                            buf.getBytes(buf.readerIndex(),array);//将字节复制到该数组
                            String s=new String(array,0,buf.readableBytes());
                            System.out.println("直接缓冲区:"+s);
                        }
                        byte[] req = new byte[buf.readableBytes()];
                        buf.readBytes(req);
                        String body = new String(req, "UTF-8");
                        System.out.println(body);
                        bufs.retain();//引用计数器加一
                        ChannelFuture future=ctx.writeAndFlush(bufs);
                        future.addListener(new ChannelFutureListener() {
                            @Override
                            public void operationComplete(ChannelFuture future) throws Exception {
                                if (future.isSuccess())
                                    System.out.println("成功");
                                else{
                                    System.out.println("失败");
                                    future.cause().printStackTrace();
                                    future.channel().close();
                                }
                            }
                        });
    //                    ctx.close();
                    }
                });
            }
        }
    
        public static void main(String[] args) throws Exception {
            int port=8080;
            new EchoServer().bind(port);
        }
    }

package netty.in.action;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.ByteBufProcessor;
    import io.netty.buffer.CompositeByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.util.ByteProcessor;
    
    import java.nio.charset.Charset;
    
    public class EchoClient {
        final ByteBuf buf= Unpooled.copiedBuffer("Hello,王宝强", Charset.forName("UTF-8"));
        public void connect(int port, String host) throws Exception {
            // 配置客户端NIO线程组
            EventLoopGroup group = new NioEventLoopGroup();
            try {
                Bootstrap b = new Bootstrap();
                b.group(group).channel(NioSocketChannel.class)
                        .option(ChannelOption.TCP_NODELAY, true)
                        .handler(new ChildChannelHandler() );
                // 发起异步连接操作
                ChannelFuture f = b.connect(host, port).sync();
                // 等待客户端链路关闭
                f.channel().closeFuture().sync();
            } finally {
                // 优雅退出，释放NIO线程组
                group.shutdownGracefully();
            }
        }
    
        private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
    
            protected void initChannel(SocketChannel ch) throws Exception {
                System.out.println("客户端启动……");
                ByteBuf bufs= Unpooled.copiedBuffer("pipeline发送的数据->", Charset.forName("UTF-8"));
                ch.pipeline().write(bufs);//通过调用ChannelPipeline的write方法将数据写入通道，但是不刷新
                ch.pipeline().addLast("text",new ChannelInboundHandlerAdapter() {
                    @Override
                    public void channelActive(ChannelHandlerContext ctx) throws Exception {
                        ctx.channel().write(Unpooled.copiedBuffer("通过ChannelHandlerContext获取的channel发送的消息->",
                                Charset.forName("UTF-8")));//通过ChannelHandlerContext获取的channel发送的消息->
                        CompositeByteBuf messageBuf=Unpooled.compositeBuffer();
                        ByteBuf headerBuf=buf;
                        ByteBuf bodyBuf=buf;
                        messageBuf.addComponent(bodyBuf);//将ByteBuf实例追加到CompositeByteBuf
                        messageBuf.addComponent(headerBuf);
                        for (ByteBuf buf:messageBuf){//遍历所有ByteBuf
                            System.out.println(buf);
                            byte[] req = new byte[buf.readableBytes()];
                            buf.readBytes(req);
                            String body = new String(req, "UTF-8");
                            System.out.println("复合缓冲区："+body);
                        }
                        ctx.writeAndFlush(buf);
                    }
    
                    public void channelRead(ChannelHandlerContext ctx, Object msg)
                            throws Exception {
                        ByteBuf buf = (ByteBuf) msg;
                        ByteBuf copyBuf=((ByteBuf) msg).copy();
    //                    System.out.println(buf.refCnt());//返回此对象的引用计数。如果为0，则表示此对象已被释放。
    //                    buf.release();//释放引用计数对象
                        for (int i = 0; i < buf.capacity(); i++) {
                            byte b=buf.getByte(i);
                            if((char)b>='a'&&(char)b<='z'||(char)b>='A'&&(char)b<='Z'||(char)b==',')
                            System.out.println("i="+(char)b);
                        }
                        int i=buf.forEachByte(new ByteProcessor() {
                            @Override
                            public boolean process(byte value) throws Exception {
                                byte[] b=",".getBytes();
                                if (b[0]!=value)
                                    return true;
                                else
                                    return false;
                            }
                        });
                        System.out.println("i="+i+" value="+(char) buf.getByte(i));
                        ByteBuf sliced = buf.slice(0,2);
                        sliced.setByte(0,(byte)'h');
                        byte[] req = new byte[buf.readableBytes()];
                        buf.readBytes(req);
                        String body = new String(req, "UTF-8");
                        System.out.println(body);
                        ctx.fireChannelRead(copyBuf);
                    }
                });
                ch.pipeline().addLast("text2",new ChannelInboundHandlerAdapter(){
                    public void channelRead(ChannelHandlerContext ctx, Object msg)
                            throws Exception {
                        ByteBuf buf = (ByteBuf) msg;
                        byte[] req = new byte[buf.readableBytes()];
                        buf.readBytes(req);
                        String body = new String(req, "UTF-8");
                        System.out.println("text2:"+body);
                        ByteBuf bufs= Unpooled.copiedBuffer("test2发送的数据", Charset.forName("UTF-8"));
                        ctx.writeAndFlush(bufs);
                        ctx.close();
                    }
                });
    //            ch.pipeline().remove("text2");
            }
        }
    
        public static void main(String[] args) throws Exception {
            int port = 8080;
            new EchoClient().connect(port, "127.0.0.1");
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70]: /images/20220417/20b61ce5291e4c27bbbaaeda63733cef.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 1]: /images/20220417/808d6c932f054c779fea16fc2300102c.png
[20181111145555354.png]: /images/20220417/c1b4b5df410a43828acf8e7d74222a46.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 2]: /images/20220417/ea041cfb448249b38294dd4b52115ec8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 3]: /images/20220417/c8ada43b2be44c82bff184b1fa79e98e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 4]: /images/20220417/5e08eb758f8f420d8242b38a164a0ed5.png
[20181111180723371.png]: /images/20220417/e0f47ad223ca4f20a0b0f2cc31199736.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 5]: /images/20220417/e2c2fa41b9dd47f19cbd098cfef3e7d9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 6]: /images/20220417/ec6f001d23f142ef95284d2352dca438.png