Netty(十二)——ChannelPipeline之观

秒速五厘米 2022-04-13 06:23 256阅读 0赞

前边文章的代码中，我们经常往pipeline中添加ChannelHandler来进行前后顺序控制处理实际业务。是不是类似Servlet和Filter过滤器，利用职责链模式的思想，通过一个一个的Handler进行业务传递处理。下边这个图感觉挺合适：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70][]

好，这篇文章主要对ChannelPipeline进行总结学习，首先看下总结思维导图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70 1][]

一，功能说明：

1，ChannelPipeline和ChannelHandler的事件处理模型：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70 2][]

如图，a，一个消息被socketChannel read()方法读取ByteBuf出发ChannelRead事件，由I/O线程NioEventLoop调用ChannelPipeline的fireChannelRead方法，将消息传输到ChannelPipeline中；b，消息一次被ChannelHandler处理，这个过程中任何ChannelHandler都可以中断当前的流程，结束消息的传递；c，处理完请求后，进行响应，调用ChannelHandlerContext的write方法发送消息，然后消息倒着在ChannelHandler中进行依次处理，直到最后哦socket进行write。

Netty中事件分为inbound事件（通常由I/O线程触发，例如TCP链路建立事件、链路关闭事件、读事件、异常通知时间等，对应上图的左半部分）和outbound事件（通常由用户主动发起的网络I/O操作，例如用户发起的连接操作、绑定操作、消息发送等操作，对应上图的右半部分）

<table>  inbound事件方法  
 <thead> 
  <tr> 
   <th>事件方法</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ChannelHandlerContext fireChannelRegistered();</td> 
   <td>Channel注册事件</td> 
  </tr> 
  <tr> 
   <td>ChannelHandlerContext fireChannelActive();</td> 
   <td>Tcp链路建立成功，Channel激活事件</td> 
  </tr> 
  <tr> 
   <td>ChannelHandlerContext fireChannelRead(Object msg);</td> 
   <td>读事件</td> 
  </tr> 
  <tr> 
   <td>ChannelHandlerContext fireChannelReadComplete();</td> 
   <td>读操作完成通知事件</td> 
  </tr> 
  <tr> 
   <td>ChannelHandlerContext fireExceptionCaught(Throwable cause);</td> 
   <td>异常通知事件</td> 
  </tr> 
  <tr> 
   <td>ChannelHandlerContext fireUserEventTriggered(Object event);</td> 
   <td>用户自定义事件</td> 
  </tr> 
  <tr> 
   <td>ChannelHandlerContext fireChannelWritabilityChanged();</td> 
   <td>Channel的可写状态变化通知事件</td> 
  </tr> 
  <tr> 
   <td>ChannelHandlerContext fireChannelInactive();</td> 
   <td>Tcp连接关闭，链路不可用通知事件</td> 
  </tr> 
 </tbody> 
</table>

<table>  outbound事件方法  
 <thead> 
  <tr> 
   <th>事件方法</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise);</td> 
   <td>绑定本地地址事件</td> 
  </tr> 
  <tr> 
   <td>ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise);</td> 
   <td>连接服务端事件</td> 
  </tr> 
  <tr> 
   <td>ChannelFuture write(Object msg, ChannelPromise promise);</td> 
   <td>发送事件</td> 
  </tr> 
  <tr> 
   <td>ChannelHandlerContext flush();</td> 
   <td>刷新事件</td> 
  </tr> 
  <tr> 
   <td>ChannelHandlerContext read();</td> 
   <td>读事件</td> 
  </tr> 
  <tr> 
   <td>ChannelFuture disconnect(ChannelPromise promise);</td> 
   <td>断开连接事件</td> 
  </tr> 
  <tr> 
   <td>ChannelFuture close(ChannelPromise promise);</td> 
   <td>关闭当前Channel事件</td> 
  </tr> 
 </tbody> 
</table>

2，自定义拦截器：也就是前边我们经常写的各种业务处理handler，例如：处理编解码的，处理半包问题的，处理加解密的，处理分隔符的……  通常我们只需要继承ChannelHandlerAdapter类覆盖自己关心的方法即可。前边的例子写了非常多，大家可以回去看下，这里简单举个：

/**
     * 拦截active时间，进行日志打印
     */
    public class MyInboundHandler extends ChannelHandlerAdapter {
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("tcp connection");
            ctx.fireChannelActive();
        }
    
    }

3，构建pipeline：其实回顾前边的例子，大家应该可以总结出来，其实不需要我们进行创建。Netty中使用ServerBootstrap或者bootstrap启动服务端或者客户端时，会为每个Channel链接创建一个独立的pipeline，我们只需要将自定义的Handler加入到pipeline中即可。（前边我们都是通过addLast()进行添加保证一定的顺序）。

4，主要特性：ChannelPipeline支持动态的添加或者删除ChannelHandler。一些业务场景会非常使用的。ChannelPipeline是线程安全，即可以运行多个业务线程并发操作ChannelPipeline，不存在多线程并发问题。但是ChannelHandler却不是线程安全，需要我们自己进行编码控制。

二，源码阶段：

1，类关系图：非常简单的，一个接口，一个默认实现类。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70 3][]

2，对ChannelHandler的管理：ChannelPipeline是ChannelHandler的管理容器，负责ChannelHandler的增删改查。类似Map的容器实现，看下新增方法吧，其它的也可以这样跟踪理解一下：

/**
     * 1,添加方法
     **/
    @Override
    public ChannelPipeline addBefore(String baseName, String name, ChannelHandler handler) {
            return addBefore((ChannelHandlerInvoker) null, baseName, name, handler);
    }
    
    /**
     * 1.1,调用添加方法
     **/
        @Override
        public ChannelPipeline addBefore(
                ChannelHandlerInvoker invoker, String baseName, String name, ChannelHandler handler) {
            //注意synchronized关键字，来保证并发安全，保证同步块内所有操作的原子性。
            synchronized (this) {
                //第一步   
                AbstractChannelHandlerContext ctx = getContextOrDie(baseName);
                //第二步 
                name = filterName(name, handler);
                //第三步 
                addBefore0(name, ctx, new DefaultChannelHandlerContext(this, invoker, name, handler));
            }
            return this;
        }
    
        //第一步，根据baseName获取它对应的DefaultChannelHandlerContext.
        private AbstractChannelHandlerContext getContextOrDie(String name) {
            AbstractChannelHandlerContext ctx = (AbstractChannelHandlerContext) context(name);
            if (ctx == null) {
                throw new NoSuchElementException(name);
            } else {
                return ctx;
            }
        }
    
        @Override
        public ChannelHandlerContext context(String name) {
            if (name == null) {
                throw new NullPointerException("name");
            }
    
            synchronized (this) {
                return name2ctx.get(name);
            }
        }
    
        //第二步：对新增的handler名进行重复性校验
        private String filterName(String name, ChannelHandler handler) {
            if (name == null) {
                return generateName(handler);
            }
    
            if (!name2ctx.containsKey(name)) {
                return name;
            }
    
            throw new IllegalArgumentException("Duplicate handler name: " + name);
        }
    
        //第三部：进行添加
        private void addBefore0(
                final String name, AbstractChannelHandlerContext ctx, AbstractChannelHandlerContext newCtx) {
            //校验
            checkMultiplicity(newCtx);
            //相当于向一个链表中添加了项应该好理解
            newCtx.prev = ctx.prev;
            newCtx.next = ctx;
            ctx.prev.next = newCtx;
            ctx.prev = newCtx;
    
            name2ctx.put(name, newCtx);
    
            callHandlerAdded(newCtx);
        }
        //进行ctx校验：!h.isSharable() && h.added
        private static void checkMultiplicity(ChannelHandlerContext ctx) {
            ChannelHandler handler = ctx.handler();
            if (handler instanceof ChannelHandlerAdapter) {
                ChannelHandlerAdapter h = (ChannelHandlerAdapter) handler;
                if (!h.isSharable() && h.added) {
                    throw new ChannelPipelineException(
                            h.getClass().getName() +
                            " is not a @Sharable handler, so can't be added or removed multiple times.");
                }
                h.added = true;
            }
        }

3，inbound事件：上边已经介绍了功能，这边以fireChannelActive为例，看下：

/**
         * 调用head.fireChannelActive()后，判断当前channel是否自动读取，如果是，调用read方法
         **/
        @Override
        public ChannelPipeline fireChannelActive() {
            head.fireChannelActive();
    
            if (channel.config().isAutoRead()) {
                channel.read();
            }
    
            return this;
        }

4，outbound事件：上边介绍了其功能，我们来用connect举例看下：

//Pipeline本身不直接进行I/O操作的，前边看Channel和Unsafe的时候，都是有它俩进行真正IO操作的。通过方法追踪，最终是可以验证到unsafe身上的。
        @Override
        public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress) {
            return tail.connect(remoteAddress, localAddress);
        }
    
        @Override
        public void connect(
                    ChannelHandlerContext ctx,
                    SocketAddress remoteAddress, SocketAddress localAddress,
                    ChannelPromise promise) throws Exception {
            unsafe.connect(remoteAddress, localAddress, promise);
        }

ChannelPipelie其实还是比较简单的，主要做了Handler的管理，还有就是各种事件的触发响应。有没有感觉就像“管事的”，管着工作怎么流转，到什么地方需要处理什么工序，有什么事情通知给他，他根据不同事情，找不同的人进行处理。嗯，大概就是这样。大家怎么看。。。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70]: /images/20220413/fcfe9da3385d444d9b5888f43036e93d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70 1]: /images/20220413/255318494db946888647f88909d8a68d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70 2]: /images/20220413/f3f2317f0759473eba626edd017d0937.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpdWppYWhhbjYyOTYyOQ_size_16_color_FFFFFF_t_70 3]: /images/20220413/ebbb3683991f416093d6864aea8327c1.png