Pipeline 设计原理

Channel 与 ChannelPipeline

image.png

一对一关系

一个 Channel 包含了一个 ChannelPipeline

    protected AbstractChannel(Channel parent) {
        this.parent = parent;
        id = newId();
        unsafe = newUnsafe();
        pipeline = newChannelPipeline();
    }

双向链表

而 ChannelPipeline 中又维护了一个由 ChannelHandlerContext 组成的双向链表。这个链表的头是 HeadContext，链表的尾是 TailContext，并且每个 ChannelHandlerContext 中又关联着一个 ChannelHandler。

    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;
    }

HeadContext和TailContext

类图

HeadContex.png
TailContext.png

head 和 tail 即是一个 ChannelHandler，又是一个 ChannelHandlerContext

分析

HeadContext是inbound = false，outbound = true,TailContext与之相反

        HeadContext(DefaultChannelPipeline pipeline) {
            super(pipeline, null, HEAD_NAME, false, true);
            unsafe = pipeline.channel().unsafe();
            setAddComplete();
        }

ChannelInitializer 的添加

初始化

ServerBootstrap server = new ServerBootstrap();
server.childHandler(new ChannelInitializer<SocketChannel>() {
protected void initChannel(SocketChannel client) {
}
})

添加

1 具体过程

ChannelInitializer的调用.png

2 结果

image.png
3 ChannelInitializer类图
image.png

    private static boolean isInbound(ChannelHandler handler) {
        return handler instanceof ChannelInboundHandler;
    }
    private static boolean isOutbound(ChannelHandler handler) {
        return handler instanceof ChannelOutboundHandler;
    }

ChannelInitializer来自于ChannelInboundHandler,所以是inBound

自定义 ChannelHandler 的添加过程

利用ChannelInitializer来添加自定义的ChannelHandler

ChannelHandler.png

给ChannelHandler 命名

检查重复性

DefaultChannelPipeline的addLast检查名字是否重复

    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);
            .....
        }
        callHandlerAdded0(newCtx);
        return this;
    }

默认命名规则

若是MyTomcatHandler则生成默认生成的名字是MyTomcatHandler#0

    private static String generateName0(Class<?> handlerType) {
        return StringUtil.simpleClassName(handlerType) + "#0";
    }

Pipeline 的事件传播机制

Netty 中的传播事件可以分为两种：Inbound 事件和 Outbound 事件

image.png

Outbound 事件传播方式

Outbound 事件都是请求事件(request event)，即请求某件事情的发生，然后通过 Outbound 事件进行通知。 Outbound 事件的传播方向是 tail -> customContext -> head。

    private AbstractChannelHandlerContext findContextOutbound() {
        AbstractChannelHandlerContext ctx = this;
        do {
            ctx = ctx.prev;
        } while (!ctx.outbound);
        return ctx;
    }

Connect.png

形成了Context.connect -> Connect.findContextOutbound -> nextContext.invokeConnect -> nextHandler.connect -> nextContext.connect的循环,最后到headContext结束。

Inbound 事件传播方式

Inbound 事件是一个通知事件,即某件事已经发生了,然后通过 Inbound 事件进行通知。Inbound 通常发生在 Channel 的状态的改变或 IO 事件就绪。 Inbound 的特点是它传播方向是 head -> customContext -> tail。
具体请上面参考ChannelInitializer的调用过程。
形成一个循环Context.fireChannelXXX() -> Connect.findContextInbound() -> nextContext.invokeChannelXXX() -> nextHandler.channelXXX() -> nextContext.fireChannelXXX(),最后到TailContext结束。

Handler 的各种姿势

ChannelHandlerContext

ChannelPipeline存储的是ChannelHandlerContext，ChannelHandlerContext存储的是ChannelHandler

ChannelHandler

ChannelHandler:主要有handlerAdded,handlerRemoved和exceptionCaught方法
ChannelHandlerAdapter是对ChannelHandler的骨架实现+向下一个handler传递。

ChannelInboundHandler

ChannelInboundHandler:当channel状态改变时,回调响应的方法即事件触发。
ChannelInitializer:自定义ChannelHandler.