6.ChannelPipeline

Posted 2020-11-26 spiritsx

• pipeline和handler
  • ChannelPipline
  • ChannelHandler
  • ChannelHandlerContext
  • pipeline的初始化
  • handler的添加和删除
  • handler的传播顺序
    • inbount事件的传播
    • pipeline与context调用传播方法的区别
• 异常的传播
  • 异常的传播路径
  • 异常优雅处理

pipeline和handler

ChannelPipline

pipeline可以译为管道、流水线，正如工厂的流水线一样，ChannelPipline将各种handler串联起来，将IO事件在这些handler中进行传播，每个handler负责一部分逻辑。从ChannelPipeline接口定义的方法可以看出来，它是一个双向链表，处理过程类似于JavaWeb中的filter。这种责任链模式的设计不仅有利于解耦，还能动态调整pipeline中的handler，这一点在前文中的channelInitializerHandler已经有所体现。

ChannelHandler

handler指的是ChannelHandler接口及其子类，是处理读写事件的类，也是实际开发时主要编写的类。ChannelHandler作为跟借口，定义了3个方法和一个注解。

public interface ChannelHandler {

    void handlerAdded(ChannelHandlerContext ctx) throws Exception;

    void handlerRemoved(ChannelHandlerContext ctx) throws Exception;

    void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception;

    @interface Sharable {}
}

从方法的名字不难理解这3个方法分别在handler被添加、移除、抛出异常时回调触发。而@Sharable注解表明某个handler实例可以被多个pipeline共享(也即多个channel共享)。
经过pipeline的后，handler处理过的事件会作为临近handler的事件入口。netty将事件分成了入站事件和出站事件，这里的入和出是相对于netty所属的应用程序而言的，一般来说，由外部触发的事件是inbound事件，而outbound事件是由应用程序主动请求而触发的事件。相应的，handler也被分成inBoundHandler和outBoundHandler两种。顾名思义，inBoundHandler只会处理inBound事件，outBoundHandler只会处理outBound事件。具体的入站和出站事件可以参考ChannelInboundHandler和ChannelOutboundHandler2个接口各自定义的方法。

// inbound事件
fireChannelRegistered()
fireChannelActive()
fireChannelRead(Object)
fireChannelReadComplete()
fireExceptionCaught()
fireUserEventTriggered()
fireChannelWritabilityChanged()
fireChannelInactive()
fireChannelUnregistered()

// outbound事件
bind()
connect()
write()
flush()
read()
disconnect()
close()
deregister()

在上述事件中，别的事件都容易理解，唯独read这个事件出现了3次，容易混淆，所以单独拿出来提一下。
fireChannelRead(Object)和FireChannelReadComplete属于inBound事件，而read属于outBound事件，这表明，read事件是应用程序主动触发的事件。在ChannelOutBoundInvoker关于read方法的注释中也提到，请求将channel中的数据读入第一个inbound缓冲区，然后根据是否还有数据来决定触发channelRead(Object)和channelReadComplete。

ChannelHandlerContext

为了使handler类更关注于实际对数据的逻辑处理，netty将handler与pipeline关联的过程交由ChannelHandlerContext完成。熟悉链表数据结构的都知道，链表的每一个节点都包含数据域和指针域，显然，handler和handlerContext的关系就像数据域和指针域。但context不仅仅只是一个指针域，从它的接口定义可以看出来，hannelHandlerContext一方面将handler包裹起来，继而进行inbound和outbound事件的传播，另一方面继承于attributeMap的attr方法也令其可以自定义一些属性(已经被废弃，转而使用handler的attr方法)。此外，context还可以为handler赋予名称、获取内存分配器，它还持有pipeline的引用，以便在必要时刻从头尾指针重新开始处理。

ChannelHandlerContext extends AttributeMap, ChannelInboundInvoker, ChannelOutboundInvoker{...}

pipeline的初始化

对以上3个类有概述性的了解后，我们先看一下pipeline是如何初始化的。
在channel初始化时，channel的构造函数初始化了一个pipeline。

protected DefaultChannelPipeline(Channel channel) {
    this.channel = ObjectUtil.checkNotNull(channel, "channel");
    succeededFuture = new SucceededChannelFuture(channel, null);
    voidPromise =  new VoidChannelPromise(channel, true);
    tail = new TailContext(this);
    head = new HeadContext(this);
    head.next = tail;
    tail.prev = head;
}

可以看到pipeline在初始化时，添加了Tail和Head2个ChannelHandlerContext，且将这2个节点作为哨兵节点，组成双向链表这样一个数据结构。
两个哨兵的继承关系如下

final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {...}
final class HeadContext extends AbstractChannelHandlerContext implements ChannelOutboundHandler, ChannelInboundHandler {
    private final Unsafe unsafe;
    ...
}

可以看到tail节点只是InboundHandler，而head节点既是InboundHandler又是OutboundHandler。tailContext通常做的是一个收尾的工作，比如异常没有捕获，传递到tail，就会打印日志等等、释放内存等等；而headContext持有一个Unsafe对象，在前文说过，unsafe是实现底层数据读写的一个类，也因此，head在处理inbount事件时，会原封不动的往下传播，而处理outbound事件时，会委托给unsafe进行处理。
这里还有一个小细节。在传播事件时需要判断下一个handler是否可以处理这个事件，netty于是将各种事件用位图的形式区分，采用这种方式大大节省判断操作所需要的额外空间。
// ChannelHandlerMask类定义的部分事件位运算

static final int MASK_EXCEPTION_CAUGHT = 1;
static final int MASK_CHANNEL_REGISTERED = 1 << 1;
static final int MASK_CHANNEL_UNREGISTERED = 1 << 2;
static final int MASK_CHANNEL_ACTIVE = 1 << 3;

利用掩码判断handler处理对应事件

do {
    ctx = ctx.prev;
} while ((ctx.executionMask & mask) == 0);

handler的添加和删除

handler在调用pipeline的addXXX系列方法里添加，以addLast(ChannelHandler... handlers)方法为例，默认情况下，该方法会重载到addLast(EventExecutorGroup group, String name, ChannelHandler handler)方法，默认情况下group和name均为null

@Override
public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
    final AbstractChannelHandlerContext newCtx;
    synchronized (this) {
        checkMultiplicity(handler);
        newCtx = newContext(group, filterName(name, handler), handler);
        addLast0(newCtx);
        if (!registered) {
            newCtx.setAddPending();
            callHandlerCallbackLater(newCtx, true);
            return this;
        }
        EventExecutor executor = newCtx.executor();
        if (!executor.inEventLoop()) {
            callHandlerAddedInEventLoop(newCtx, executor);
            return this;
        }
    }
    callHandlerAdded0(newCtx);
    return this;
}

总的来说可以分为3个步骤：

1. 检查handler是否重复添加，主要是通过handler的@Sharable注解和added字段判断;
2. 创建HandlerContext，并添加到链表中。
3. 回调handlerAdded方法。

handler的删除类似，先通过参数找到对应的handler，然后删除链表中的context节点，最后回调handlerRemove方法。

handler的传播顺序

由于采用了责任链模式，链表节点之间的顺序就显得非常重要了，先看一下inbound事件是如何在pipeline中传播的

inbount事件的传播

inbound以AbstractChannelHandlerContext中的fireChannelRead(Object)方法为例。

public ChannelHandlerContext fireChannelRead(final Object msg) {
    invokeChannelRead(findContextInbound(MASK_CHANNEL_READ), msg);
    return this;
}

可以看出，fireChannelRead做了2件事

1. 通过事件对应的掩码找到下一个inboundHandler
2. 将本节点处理好的数据传播给下一个inboundHandler

// 步骤1
private AbstractChannelHandlerContext findContextInbound(int mask) {
    AbstractChannelHandlerContext ctx = this;
    do {
        ctx = ctx.next;
    } while ((ctx.executionMask & mask) == 0);
    return ctx;
}    
// 步骤2
static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeChannelRead(msg);
    } else {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                next.invokeChannelRead(msg);
            }
        });
    }
}

步骤1的实现是不断通过context的executionMask与事件掩码做与运算，直到与的结果不为0。这表明该context对应的handler具备处理对应事件的能力。此外要注意循环过程中，context是next方向。
步骤2则判断当前线程是否是eventLoop线程，若是，则执行下一个inboundHandlerContext的invokeChannelRead方法，若不是则添加到任务队列里，待eventLoop线程来执行
至于invokeChannelRead方法也很简单，先判断该handler是否已存在于pipeline，然后调用handler的channelRead方法。

private void invokeChannelRead(Object msg) {
    if (invokeHandler()) {
        try {
            ((ChannelInboundHandler) handler()).channelRead(this, msg);
        } catch (Throwable t) {
            notifyHandlerException(t);
        }
    } else {
        fireChannelRead(msg);
    }
}
// 判断是否添加到pipeline中或即将添加到pipeline中
private boolean invokeHandler() {
    int handlerState = this.handlerState;
    return handlerState == ADD_COMPLETE || (!ordered && handlerState == ADD_PENDING);
}

outbound事件传播与inbound类似，只是在通过掩码查询下一个outboundHandler时为prev方向，与inbound相反。具体代码略过。

pipeline与context调用传播方法的区别

pipeline.fireChannelRead()和ChannelHandlerContext.fireChannelRead()在代码中都时常出现，那么它们的区别是什么?
不妨看一下DefaultChannelPipeline的fireChannelRead方法。

public final ChannelPipeline fireChannelRead(Object msg) {
    AbstractChannelHandlerContext.invokeChannelRead(head, msg);
}

可以看出，其将headContext作为参数传入，调用了HandlerContext的invokeChannelRead(AbstractChannelHandlerContext, Object)静态方法，这个静态方法会调用传入的HandlerContext的invokeChannelRead(Object)方法，继而调用Context内部持有的ChannelInboundHandler的channelRead(ChannelHandlerContext, Object)方法。这个方法由子类重写，在这里就是HeadContext重写的方法，它调用传入的ChannelHandlerContext，继续往下传播。

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    ctx.fireChannelRead(msg);
}

而DefaultChannelPipeline的read方法则调用tail的read方法，tail会传播给它的前一个节点。
小结
pipeline调用传播方法时，若是inbound事件，从head开始往tail方向传播，若是outbound事件，从tail开始往head方向传播
context调用传播方法，若是inbound事件，从当前context节点开始往tail方向传播，若是outbound事件，从当前context节点开始往head方向传播

异常的传播

异常的传播路径

在context处理各种事件时，用了channelRead的例子。可以注意到invokeChannelRead方法实现用了一个try-catch的写法。当抛出异常时，会调用notifyHandlerException(Throwable)，代码如下：

private void notifyHandlerException(Throwable cause) {
    if (inExceptionCaught(cause)) {
        if (logger.isWarnEnabled()) {
            logger.warn(
                    "An exception was thrown by a user handler " +
                            "while handling an exceptionCaught event", cause);
        }
        return;
    }
    invokeExceptionCaught(cause);
}

首先调用inExceptionCaught，判断异常是否发生在exceptionCaught方法内。若是，则打印警告日志后直接返回，否则调用invokeExceptionCaught(Throwable)方法。该方法会调用handler复写的exceptionCaught方法。
若复写方法调用了ChannelHandlerContext.fireExceptionCaught方法，则异常会继续往下传播，不论下一个节点是inbound还是outbound。若一直传播到tail，则会打印一个日志，并释放异常占用的内存。

异常优雅处理

在springMvc体系中，通常会有一个包含ControllerAdvice注解的类统一进行异常的处理，在netty中，也可以在pipeline的末尾添加一个异常处理handler统一进行异常处理。甚至可以用策略模式，对不同异常类进行分门别类的处理。