netty源码之ChannelPipeline

ゝ一世哀愁。 2022-12-27 07:20 153阅读 0赞

# netty源码之ChannelPipeline #

Netty中ChannelPipeline和ChannelHandler机制类似于Servlet和Filter过滤器，实际上都是职责链模式的一种变形，主要是为了方便事件的拦截和用户业务逻辑的定制。

Netty的Channel过滤器实现原理与Servlet的Filter机制一致，它将Channel的数据管道抽象为ChannelPipeline，消息在ChannelPipeline中流动和传递。ChannelPipeline持有IO事件拦截器ChannelHandler的链表，由ChannelHandler对IO事件进行拦截和处理，可以方便地通过新增和删除ChannelHandler来实现不同的业务逻辑定制，不需要对已有的ChannelHandler进行修改，能够实现对修改封闭和对扩展的支持（开闭原则）。

## ChannelPipeline的结构 ##

ChannelPipeline是ChannelHandler的容器，它负责ChannelHandler的管理和事件拦截与调度。

ChannelPipeline底层使用了一个双向链表来存储ChannelHandler，但并不是直接存储的ChannelHandler，而是ChannelHandlerContext，在ChannelHandlerContext可以直接获取到与之对应的ChannelHandler、ChannelPipeline、Channel。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]

在使用Netty时，我们并不需要手动创建ChannelPipeline，因为使用ServerBootstrap或者Bootstrap启动服务端或者客户端时，Netty会为每个Channel连接创建一个独立的ChannelPipeline。对于使用者而言，只需要将自定义的拦截器加入到ChannelPipeline中即可。

ChannelPipeline支持运行态动态的添加或者删除ChannelHandler，在某些场景下这个特性非常实用。例如当业务高峰期需要对系统做拥塞保护时，就可以根据当前的系统时间进行判断，如果处于业务高峰期，则动态地将系统拥塞保护ChannelHandler添加到当前的ChannelPipeline中，当高峰期过去之后，就可以动态删除拥塞保护ChannelHandler了。

ChannelPipeline是线程安全的,这意味着N个业务线程可以并发地操作ChannelPipeline而不存在多线程并发问题。但是，ChannelHandler却不是线程安全的，这意味着尽管ChannelPipeline是线程安全的，但是用户仍然需要自己保证ChannelHandler的线程安全。

怎么保证ChannelHandler的线程安全？

1.  每个Channel尽量保证都拥有自己的Handler，而不是Handler在多个Channel之间共享，也就是下面的代码不推荐使用，这么写那么这个ServerHandler必须用注解`@ChannelHandler.Sharable`明确表明这是一个共享的handler，而且是线程安全的。

ServerBootstrap b = new ServerBootstrap();
    ServerHandler serverHandler = new ServerHandler()；
    b.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .childHandler(new ChannelInitializer<SocketChannel>() { 
                @Override
                public void initChannel(SocketChannel ch) { 
                    ch.pipeline().addLast(serverHandler);
                }
            });

而是推荐这么使用：`ch.pipeline().addLast(new ServerHandler())`;

1.  尽量不使用共享资源，如果使用了共享资源，在操作时需要对共享资源进行同步，如加锁。

## ChannelPipeline的事件处理 ##

Netty中的事件分为inbound（入站）事件和outbound（出站）事件。

inbound事件通常由IO线程触发，例如TCP链路建立事件、链路关闭事件、读事件、异常通知事件等。触发inbound事件的方法如下：

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

outbound事件通常是由用户主动发起的网络IO操作，例如用户发起的连接操作、绑定操作、消息发送等操作。触发outbound事件的方法如下:

*  ChannelHandlerContext.bind(Socket.Address, ChannelPromise)：绑定本地地址事件。
 *  ChannelHandlerContext.connect(SocketAddress，SocketAddress， ChannelPromise)：连接服务端事件。
 *  ChannelHandlerContext.write(Object, ChannelPromise)：发送事件。
 *  ChannelHandlerContext.flush()：刷新事件。
 *  ChannelHandlerContext.read()：读事件。
 *  ChannelHandlerContext.disconnect(ChannelPromise)：断开连接事件。
 *  ChannelHandlerContext.close(ChannelPromise)：关闭当前Channel事件。

## DefaultChannelPipeline ##

ChannelPipeline的实现其实只有一个DefaultChannelPipeline。

### DefaultChannelPipeline对ChannelHandler的管理 ###

ChannelPipeline是ChannelHandler的管理容器，必然要负责ChannelHandler的查询、添加、替换和删除，所以它里面的方法很多都是和管理ChannelHandler有关的，比如addXXX、removeXXX、replaceXXX等。

ChannelPipeline内部维护了一个ChannelHandler的链表和迭代器，所以从本质上来说，上述的管理操作基本上就是对链表的操作。但是ChannelHandler本身并不是链表上的元素，所以在具体实现上是把ChannelHandler包装进ChannelHandlerContext的实例DefaultChannelHandlerContext，然后把ChannelHandlerContext作为元素来组成链表。

下面是DefaultChannelPipeline.addLast()方法的源码：

@Override
    public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) { 
        final AbstractChannelHandlerContext newCtx;
        synchronized (this) { 
            checkMultiplicity(handler); // 检查handler是否是共享的，如果是，必须加@Sharable注解，否则会抛出异常
    
            // filterName()方法会检查handler是否已经存在
            // 创建DefaultChannelHandlerContext
            newCtx = newContext(group, filterName(name, handler), handler);
    
            // 添加至链表尾部
            addLast0(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()) { 
                callHandlerAddedInEventLoop(newCtx, executor);
                return this;
            }
        }
        // 回调handler.handlerAdd()方法
        callHandlerAdded0(newCtx);
        return this;
    }

由于ChannelPipeline支持运行期动态修改，因此存在两种潜在的多线程并发访问场景。

1.  IO线程和用户业务线程的并发访问;
2.  用户多个线程之间的并发访问。

为了保证ChannelPipeline的线程安全性，需要通过线程安全容器或者锁来保证并发访问的安全，此处Netty直接使用了synchronized关键字，保证同步块内的所有操作的原子性。在加入链表之前，Netty还会检查该Handler是否已经被添加过及其名字是否有重复，如果该Handler不是共享的，而且被添加过抛出异常，如果名字重复，也会抛出异常。

完成链表操作后，后面的部分基本上做的就是一件事，执行方法callHandlerAdded0，只是根据条件不同进行同步或者异步执行，callHandlerAdded0的主要作用是在handler添加被加入链表之后做一些额外工作，Netty本身对这个方法的实现，在ChannelHandlerAdapter类中，是个空操作，这里如果我们需要做一些业务逻辑, 可以通过重写该方法进行实现。

callHandlerAddedInEventLoop和callHandlerAdded0的区别：

*  callHandlerAdded0直接使用当前线程来回调handler.handlerAdd()方法。
 *  callHandlerAddedInEventLoop是使用context中所携带的线程池来回调handler.handlerAdd()方法。

也就说如果addLast()方法是由内部EventLocp中的线程调用，那么就会直接回调handler.handlerAdd()方法，如果是外部用户线程调用，那么就会放入到当前EventLocp的任务队列中，等待其中的线程进行回调。

### inbound事件 ###

当发生某个I/O事件的时候，例如链路建立、链路关闭、读取操作完成等，都会产生一个事件，事件在pipeline中得到传播和处理，它是事件处理的总入口。由于网络IO相关的事件有限，因此Netty对这些事件进行了统一抽象,Netty自身和用户的ChannelHandler会对感兴趣的事件进行拦截和处理。

pipeline中以fireXXX命名的方法都是从IO线程流向用户业务Handler的inbound事件，它们的实现因功能而异，但是处理步骤类似，总结如下。

1.  调用HeadHandler对应的fireXXX方法;
2.  执行事件相关的逻辑操作。

inbound事件对应的方法如下：

ChannelPipeline fireChannelRegistered();
    
    ChannelPipeline fireChannelUnregistered();
    
    ChannelPipeline fireChannelActive();
    
    ChannelPipeline fireChannelInactive();
    
    ChannelPipeline fireExceptionCaught(Throwable cause);
    
    ChannelPipeline fireUserEventTriggered(Object event);
    
    ChannelPipeline fireChannelRead(Object msg);
    
    ChannelPipeline fireChannelReadComplete();
    
    ChannelPipeline fireChannelWritabilityChanged();

来看一下fireChannelActive()方法的源码：

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

可见pipeline的fireChannelActive对应的方法都是委托双向链接中的head.invokeChannelActive()进行处理，实际上会触发对应handler.channelActive()，其他方法也类似。

### outbound事件 ###

由用户线程或者代码发起的IO操作被称为outbound事件，事实上inbound和outbound是Netty自身根据事件在pipeline中的流向抽象出来的术语，在其他NIO框架中并没有这个概念。outbound事件包括发起绑定、发起连接、发起读写、刷新数据、发起关闭、发起断连等等。

Pipeline本身并不直接进行IO操作，最终都是由Unsafe和Channel来实现真正的IO操作的。Pipeline负责将IO事件通过TailHandler进行调度和传播，最终调用Unsafe的IO方法进行I/O操作。

outbound事件对应的方法如下：

ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise);
    
    ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise);
    
    ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise);
    
    ChannelFuture disconnect(ChannelPromise promise);
    
    ChannelFuture close(ChannelPromise promise);
    
    ChannelFuture deregister(ChannelPromise promise);
    
    ChannelOutboundInvoker read();
    
    ChannelFuture write(Object msg);
    
    ChannelFuture write(Object msg, ChannelPromise promise);
    
    ChannelOutboundInvoker flush();
    
    ChannelFuture writeAndFlush(Object msg, ChannelPromise promise);
    
    ChannelFuture writeAndFlush(Object msg);

来看一下write()方法的源码：

@Override
    public final ChannelFuture write(Object msg) { 
        return tail.write(msg);
    }

可见pipeline的出站对应的方法都是委托双向链表中的tail进行处理。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/7866755ee74d4afc9879056920272f03.png