一、pipeline介绍

1. 什么是pipeline

    
pipeline 有管道，流水线的意思，最早使用在 Unix 操作系统中，可以让不同功能的程序相互通讯，使软件更加”高内聚，低耦合”，它以一种”链式模型”来串起不同的程序或组件，使它们组成一条直线的工作流。

2. Netty的ChannelPipeline

    
ChannelPipeline 是处理或拦截channel的进站事件和出站事件的双向链表，事件在ChannelPipeline中流动和传递，可以增加或删除ChannelHandler来实现对不同业务逻辑的处理。通俗的说，ChannelPipeline是工厂里的流水线，ChannelHandler是流水线上的工人。
    
ChannelPipeline在创建Channel时会自动创建，每个Channel都拥有自己的ChannelPipeline。

3. Netty I/O事件的处理过程

image

image

二、ChannelHandlerContext

1. 什么是ChannelHandlerContext

    
ChannelHandlerContext是将ChannelHandler和ChannelPipeline关联起来的上下文环境，每添加一个handler都会创建ChannelHandlerContext实例，管理ChannelHandler在ChannelPipeline中的传播流向。

2. ChannelHandlerContext和ChannelPipeline以及ChannelHandler之间的关系

    
ChannelPipeline依赖于Channel的创建而自动创建，保存了channel，将所有handler组织起来，相当于工厂的流水线。
    
ChannelHandler拥有独立功能逻辑，可以注册到多个ChannelPipeline，是不保存channel的，相当于工厂的工人。
    
ChannelHandlerContext是关联ChannelHandler和ChannelPipeline的上下文环境，保存了ChannelPipeline，控制ChannelHandler在ChannelPipeline中的传播流向，相当于流水线上的小组长。

三、传播Inbound事件

1. Inbound事件有哪些？

image

2. 添加读事件

    
从前面《Netty 源码解析-服务端启动流程解析》和《Netty 源码解析-客户端连接接入及读I/O解析》我们知道，当有新连接接入时，我们执行注册流程，注册成功后，会调用channelRegistered，我们从这个方法开始

   public final void channelRegistered(ChannelHandlerContext ctx) throws Exception {
          initChannel((C) ctx.channel());
          ctx.pipeline().remove(this);
          ctx.fireChannelRegistered();
}

    
initChannel是在服务启动时配置的参数childHandler重写了父类方法

private class IOChannelInitialize extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        System.out.println("initChannel");
        ch.pipeline().addLast(new IdleStateHandler(1000, 0, 0));
        ch.pipeline().addLast(new IOHandler());
    }
}

    
我们回忆一下，pipeline是在哪里创建的

protected AbstractChannel(Channel parent) {
    this.parent = parent;
    unsafe = newUnsafe();
    pipeline = new DefaultChannelPipeline(this);
}

    
当创建channel时会自动创建pipeline

public DefaultChannelPipeline(AbstractChannel channel) {
    if (channel == null) {
        throw new NullPointerException("channel");
    }
    this.channel = channel;

    tail = new TailContext(this);
    head = new HeadContext(this);

    head.next = tail;
    tail.prev = head;
}

    
在这里会创建两个默认的handler，一个InboundHandler --> TailContext，一个OutboundHandler --> HeadContext
    
再看addLast方法

@Override
public ChannelPipeline addLast(ChannelHandler... handlers) {
    return addLast(null, handlers);
}

    
在这里生成一个handler名字，生成规则由handler类名加 ”#0”

  @Override
public ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {
    …
    for (ChannelHandler h: handlers) {
        if (h == null) {
            break;
        }
        addLast(executor, generateName(h), h);
    }
    return this;
}

@Override
public ChannelPipeline addLast(EventExecutorGroup group, final String name, ChannelHandler handler) {
    synchronized (this) {
        checkDuplicateName(name);
        AbstractChannelHandlerContext newCtx = new DefaultChannelHandlerContext(this, group, name, handler);
        addLast0(name, newCtx);
    }
    return this;
}

    
由于pipeline是线程非安全的，通过加锁来保证并发访问的安全，进行handler名称重复性校验，将handler包装成DefaultChannelHandlerContext，最后再添加到pipeline

private void addLast0(final String name, AbstractChannelHandlerContext newCtx) {
    checkMultiplicity(newCtx);

    AbstractChannelHandlerContext prev = tail.prev;
    newCtx.prev = prev;
    newCtx.next = tail;
    prev.next = newCtx;
    tail.prev = newCtx;

    name2ctx.put(name, newCtx);

    callHandlerAdded(newCtx);
}

    
这里分三步
    
(1) 对DefaultChannelHandlerContext进行重复性校验，如果DefaultChannelHandlerContext不是可以在多个pipeline中共享的，且已经被添加到pipeline中，则抛出异常
    
(2) 修改pipeline中的指针
       
添加IdleStateHandler之前
         
HeadContext --> IOChannelInitialize --> TailContext

image

       
添加IdleStateHandler之后
         
HeadContext --> IOChannelInitialize --> IdleStateHandler --> TailContext

image

    
(3) 将handler名和DefaultChannelHandlerContext建立映射关系
    
(4) 回调handler添加完成监听事件
    
最后删除IOChannelInitialize

image

3. pipeline.fireChannelRead()事件解析

    
在这里我们选一个比较典型的读事件解析，其他事件流程基本类似

 private static void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {    
    …
    if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
            unsafe.read();
    }
    …
}

    
当boss线程监听到读事件，会调用unsafe.read()方法

@Override
public final void read() {
    …
    pipeline.fireChannelRead(byteBuf);
    …
}

    
入站事件从head开始，tail结束

@Override
public ChannelPipeline fireChannelRead(Object msg) {
    head.fireChannelRead(msg);
    return this;
}

@Override
public ChannelHandlerContext fireChannelRead(final Object msg) {
    if (msg == null) {
        throw new NullPointerException("msg");
    }

    final AbstractChannelHandlerContext next = findContextInbound();
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeChannelRead(msg);
    } else {
        executor.execute(new OneTimeTask() {
            @Override
            public void run() {
                next.invokeChannelRead(msg);
            }
        });
    }
    return this;
}

    
查找pipeline中下一个Inbound事件

private AbstractChannelHandlerContext findContextInbound() {
    AbstractChannelHandlerContext ctx = this;
    do {
        ctx = ctx.next;
    } while (!ctx.inbound);
    return ctx;
}

image

private void invokeChannelRead(Object msg) {
    try {
        ((ChannelInboundHandler) handler()).channelRead(this, msg);
    } catch (Throwable t) {
        notifyHandlerException(t);
    }
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (readerIdleTimeNanos > 0 || allIdleTimeNanos > 0) {
        reading = true;
        firstReaderIdleEvent = firstAllIdleEvent = true;
    }
    ctx.fireChannelRead(msg);
}

    
将这个channel读事件标识为true，并传到下一个handler

image

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    super.channelRead(ctx, msg);
    System.out.println(msg.toString());
}

    
这里执行IOHandler重写的channelRead()方法,并调用父类channelRead方法

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

    
继续调用事件链上的下一个handler

image

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    try {
        logger.debug(
                "Discarded inbound message {} that reached at the tail of the pipeline. " +
                        "Please check your pipeline configuration.", msg);
    } finally {
        ReferenceCountUtil.release(msg);
    }
}

    
这里会调用TailContext的Read方法，释放msg缓存
    
总结:传播Inbound事件是从HeadContext节点往上传播，一直到TailContext节点结束

四、传播Outbound事件

1. Outbound事件有哪些？

image

2. 解析write事件

    ByteBuf resp = Unpooled.copiedBuffer("hello".getBytes());
    ctx.channel().write(resp);

    
我们在项目中像上面这样直接调用write写数据，并不能直接写进channel，而是写到缓冲区，还要调用flush方法才能将数据刷进channel，或者直接调用writeAndFlush。
    
在这里我们选择比较典型的write事件来解析Outbound流程，其他事件流程类似

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

    
通过上下文绑定的channel直接调用write方法，调用channel相对应的事件链上的handler

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

    
写事件是从tail向head调用，和读事件刚好相反

@Override
public ChannelFuture write(Object msg) {
    return write(msg, newPromise());
}

@Override
public ChannelFuture write(final Object msg, final ChannelPromise promise) {
    ...
     write(msg, false, promise);
    ...
}

private void write(Object msg, boolean flush, ChannelPromise promise) {
    AbstractChannelHandlerContext next = findContextOutbound();
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeWrite(msg, promise);
        if (flush) {
            next.invokeFlush();
        }
    ...
}
...
}

    
经过多次跳转，获取上一个Ounbound事件链的handler

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

image

image

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    unsafe.write(msg, promise);
}

@Override
public final void write(Object msg, ChannelPromise promise) {
    ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
    ...
    outboundBuffer.addMessage(msg, size, promise);
...
}

    
从这里我们看到，最终是把数据丢到了缓冲区，自此netty 的pipeline模型我们解析完毕
    
有关inbound事件和outbound事件的传输, 可通过下图进行归纳:

image

觉得对您有帮助请点 "赞"