码迷,mamicode.com
首页 > 其他好文 > 详细

6.ChannelPipeline

时间:2019-12-30 00:11:53      阅读:93      评论:0      收藏:0      [点我收藏+]

标签:调用   safe   ace   区分   invoke   ret   出现   决定   promise   

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统一进行异常处理。甚至可以用策略模式,对不同异常类进行分门别类的处理。

6.ChannelPipeline

标签:调用   safe   ace   区分   invoke   ret   出现   决定   promise   

原文地址:https://www.cnblogs.com/spiritsx/p/12116923.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!