跳到主要内容

05、Netty学习 - ChannelPipeline

我们在前面的文章中也对ChannelPipeline接口做了初步的介绍。

ChannelPipeline 接口

ChannelPipeline接口采用了责任链设计模式,底层采用双向链表的数据结构,将链上的各个处理器串联起来。客户端每一个请求的到来,ChannelPipeline中所有的处理器都有机会处理它。

每一个新创建的Channel都将会被分配一个新的ChannelPipeline。这项关联是永久性的;Channel既不能附加另一个ChannelPipeline,也不能分离其当前的。

创建 ChannelPipeline

ChannelPipeline数据管道是与Channel管道绑定的,一个Channel通道对应一个ChannelPipelineChannelPipeline是在Channel初始化时被创建。

观察下面这个实例:

public void run() throws Exception {
    EventLoopGroup bossGroup = new NioEventLoopGroup(); // (1)
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap(); // (2)
        b.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class) // (3)
            .childHandler(new ChannelInitializer<SocketChannel>() { // (4)
                @Override
                public void initChannel(SocketChannel ch) throws Exception {

                    // 添加ChannelHandler到ChannelPipeline
                    ch.pipeline().addLast(new DiscardServerHandler());
                }
            })
            .option(ChannelOption.SO_BACKLOG, 128)          // (5)
            .childOption(ChannelOption.SO_KEEPALIVE, true); // (6)

        // 绑定端口,开始接收进来的连接
        ChannelFuture f = b.bind(port).sync(); // (7)

        System.out.println("DiscardServer已启动,端口:" + port);

        // 等待服务器  socket 关闭 。
        // 在这个例子中,这不会发生,但你可以优雅地关闭你的服务器。
        f.channel().closeFuture().sync();
    } finally {
        workerGroup.shutdownGracefully();
        bossGroup.shutdownGracefully();
    }
}

从上述代码中可以看到,当ServerBootstrap初始化后,直接就可以获取到SocketChannel上的ChannelPipeline,而无需手动实例化,因为 Netty 会为每个Channel连接创建一个ChannelPipeline

Channel的大部分子类都继承了AbstractChannel,在创建实例时也会调用AbstractChannel构造器。在AbstractChannel构造器中会创建ChannelPipeline管道实例,核心代码如下:

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

protected DefaultChannelPipeline newChannelPipeline() {
    return new DefaultChannelPipeline(this);
}

从上述代码中可以看出,在创建Channel时,会由Channel创建DefaultChannelPipeline类的实例。DefaultChannelPipelineChannelPipeline的默认实现。

pipelineAbstractChannel的属性,内部维护着一个以AbstractChannelHandlerContext为节点的双向链表,创建的headtail节点分别指向链表头尾,源码如下:

public class DefaultChannelPipeline implements ChannelPipeline {   	

    protected DefaultChannelPipeline(Channel channel) {
        this.channel = (Channel)ObjectUtil.checkNotNull(channel, "channel");
        this.succeededFuture = new SucceededChannelFuture(channel, (EventExecutor)null);
        this.voidPromise = new VoidChannelPromise(channel, true);
        this.tail = new DefaultChannelPipeline.TailContext(this);
        this.head = new DefaultChannelPipeline.HeadContext(this);
        this.head.next = this.tail;
        this.tail.prev = this.head;
    }
    ...
    
    final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {
        TailContext(DefaultChannelPipeline pipeline) {
            super(pipeline, (EventExecutor)null, DefaultChannelPipeline.TAIL_NAME, DefaultChannelPipeline.TailContext.class);
            this.setAddComplete();
        }

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

        HeadContext(DefaultChannelPipeline pipeline) {
            super(pipeline, (EventExecutor)null, DefaultChannelPipeline.HEAD_NAME, DefaultChannelPipeline.HeadContext.class);
            this.unsafe = pipeline.channel().unsafe();
            this.setAddComplete();
        }
          
     	...
          
      }
    
     ...
}

从上述源码可以看到,TailContextHeadContext都继承了AbstractChannelHandlerContext,并实现了ChannelHandler接口。AbstractChannelHandlerContext内部维护着nextprev链表指针和入站、出站节点方向等。其中TailContext实现了ChannelInboundHandlerHeadContext实现了ChannelOutboundHandlerChannelInboundHandler

ChannelPipeline 事件传输机制

通过ChannelPipelineaddFirst()方法来添加ChannelHandler,并为这个ChannelHandler创建一个对应的DefaultChannelHandlerContext实例。

public class DefaultChannelPipeline implements ChannelPipeline {  
    //...
    
	public final ChannelPipeline addFirst(EventExecutorGroup group, String name, ChannelHandler handler) {
        AbstractChannelHandlerContext newCtx;
        synchronized(this) {
            checkMultiplicity(handler);
            name = this.filterName(name, handler);
            newCtx = this.newContext(group, name, handler);
            this.addFirst0(newCtx);
            if (!this.registered) {
                newCtx.setAddPending();
                this.callHandlerCallbackLater(newCtx, true);
                return this;
            }

            EventExecutor executor = newCtx.executor();
            if (!executor.inEventLoop()) {
                this.callHandlerAddedInEventLoop(newCtx, executor);
                return this;
            }
        }

        this.callHandlerAdded0(newCtx);
        return this;
    }
    
    //...
    

    private AbstractChannelHandlerContext newContext(EventExecutorGroup group, String name, ChannelHandler handler) {
        return new DefaultChannelHandlerContext(this, this.childExecutor(group), name, handler);
    }
    
    //...

}

处理出站事件

当处理出站事件时,channelRead()方法的示例如下:

public class EchoServerHandler extends ChannelInboundHandlerAdapter {

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) {
		System.out.println(ctx.channel().remoteAddress() + " -> Server :" + msg);
		
		// 写消息到管道
		ctx.write(msg);// 写消息
	}
	
	//...
}

上述代码中的write()方法会触发一个出站事件,该方法会调用DefaultChannelPipeline上的write()方法。

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

从上述源码可以看到,调用的是DefaultChannelPipeline上尾部节点(tail)的write方法。

上述方法最终会调用到DefaultChannelHandlerContextwrite()方法。

private void write(Object msg, boolean flush, ChannelPromise promise) {
    ObjectUtil.checkNotNull(msg, "msg");

    try {
        if (this.isNotValidPromise(promise, true)) {
            ReferenceCountUtil.release(msg);
            return;
        }
    } catch (RuntimeException var8) {
        ReferenceCountUtil.release(msg);
        throw var8;
    }

    AbstractChannelHandlerContext next = this.findContextOutbound(flush ? 98304 : '耀');
    Object m = this.pipeline.touch(msg, next);
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        if (flush) {
            next.invokeWriteAndFlush(m, promise);
        } else {
            next.invokeWrite(m, promise);
        }
    } else {
        AbstractChannelHandlerContext.WriteTask task = AbstractChannelHandlerContext.WriteTask.newInstance(next, m, promise, flush);
        if (!safeExecute(executor, task, promise, m, !flush)) {
            task.cancel();
        }
    }

}

上述的write()方法会查找下一个出站的节点,也就是当前ChannelHandler后的一个出站类型的ChannelHandler,并调用下一个节点的invokeWrite()方法。

void invokeWrite(Object msg, ChannelPromise promise) {
    if (this.invokeHandler()) {
        this.invokeWrite0(msg, promise);
    } else {
        this.write(msg, promise);
    }

}

接着调用invokeWrite0()方法,该方法最终调用ChannelOutboundHandlerwrite方法。

    private void invokeWrite0(Object msg, ChannelPromise promise) {
        try {
            ((ChannelOutboundHandler)this.handler()).write(this, msg, promise);
        } catch (Throwable var4) {
            notifyOutboundHandlerException(var4, promise);
        }

    }

至此,处理完成了第一个节点的处理,开始执行下一个节点并不断循环。

所以,处理出站事件时,数据传输的方向是从尾部节点tail到头部节点head

处理入站事件

入站事件处理的起点是触发ChannelPipeline fire方法,例如fireChannelActive()方法的示例如下:

public class DefaultChannelPipeline implements ChannelPipeline {   	  
    //...
    
    public final ChannelPipeline fireChannelActive() {
            AbstractChannelHandlerContext.invokeChannelActive(this.head);
            return this;
        }
    //...
}

从上述源码可以看到,处理的节点是头部节点headAbstractChannelHandlerContext.invokeChannelActive方法定义如下:

static void invokeChannelActive(final AbstractChannelHandlerContext next) {
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeChannelActive();
    } else {
        executor.execute(new Runnable() {
            public void run() {
                next.invokeChannelActive();
            }
        });
    }

}

该方法最终调用ChannelInboundHandlerchannelActive方法。

private void invokeChannelActive() {
    if (this.invokeHandler()) {
        try {
            ((ChannelInboundHandler)this.handler()).channelActive(this);
        } catch (Throwable var2) {
            this.invokeExceptionCaught(var2);
        }
    } else {
        this.fireChannelActive();
    }

}

至此完成了第一个节点的处理,开始执行下一个节点的不断循环。

所以,处理入站事件时,数据传输的方向是从头部节点head到尾部节点tail

ChannelPipeline 中的 ChannelHandler

从上述的ChannelPipeline接口源码可以看出,ChannelPipeline是通过addXxx或者removeXxx方法来将ChannelHandler动态的添加到ChannelPipeline中,或者从ChannelPipeline移除ChannelHandler的。那么ChannelPipeline是如何保障并发访问时的安全呢?

addLast方法为例,DefaultChannelPipeline的源码如下:

public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
    AbstractChannelHandlerContext newCtx;
    //synchronized 保障线程安全
    synchronized(this) {
        checkMultiplicity(handler);
        newCtx = this.newContext(group, this.filterName(name, handler), handler);
        this.addLast0(newCtx);
        if (!this.registered) {
            newCtx.setAddPending();
            this.callHandlerCallbackLater(newCtx, true);
            return this;
        }

        EventExecutor executor = newCtx.executor();
        if (!executor.inEventLoop()) {
            this.callHandlerAddedInEventLoop(newCtx, executor);
            return this;
        }
    }

    this.callHandlerAdded0(newCtx);
    return this;
}

从上述源码可以看到,使用synchronized关键字保障了线程的安全访问。其他方法的实现方式也是类似。

ChannelHandlerContext 接口

ChannelHandlerContext接口是联系ChannelHandlerChannelPipeline之间的纽带。

每当有ChannelHandler添加到ChannelPipeline中时,都会创建ChannelHandlerContextChannelHandlerContext的主要功能是管理它所关联的ChannelHandler和在同一个ChannelPipeline中的其他ChannelHandler之间的交互。

例如,ChannelHandlerContext可以通知ChannelPipeline中的下一个ChannelHandler开始执行及动态修改其所属的ChannelPipeline

ChannelHandlerContext中包含了许多方法,其中一些方法也出现在ChannelChannelPipeline中。如果通过ChannelChannelPipeline的实例来调用这些方法,它们就会在整个ChannelPipeline中传播。相比之下,一样的方法在ChannelHandlerContext的实例上调用,就只会从当前的ChannelHandler开始并传播到相关管道中的下一个有处理事件能力的ChannelHandler中。因此,ChannelHandlerContext所包含的事件流比其他类中同样的方法都要短,利用这一点可以尽可能提高性能。

ChannelHandlerContext 与其他组件的关系

下图展示了ChannelPipelineChannelChannelHandlerChannelHandlerContext之间的关系做了如下说明:

 

(1)Channel被绑定到ChannelPipeline上。

(2)和Channel绑定的ChannelPipeline包含了所有的ChannelHandler

(3)ChannelHandler

(4)当添加ChannelHandlerChannelPipeline时,ChannelHandlerContext被创建。

跳过某些 ChannelHandler

下面的代码,展示了从ChannelHandlerContext获取到Channel的引用,并通过调用Channel上的write()方法来触发一个写事件到流中。

ChannelHandlerContext ctx = context;
Channel channel = ctx.channel(); //获取ChannelHandlerContext上的Channel
channel.write(msg);

以下代码展示了从ChannelHandlerContext获取到ChannelPipeline

ChannelHandlerContext ctx = context;
ChannelPipeline pipeline = ctx.pipeline(); //获取ChannelHandlerContext上的ChannelPipeline 
pipeline.write(msg);

上述的两个示例,事件流是一样的。虽然被调用的ChannelChannelPipeline上的write()方法将一直传播事件通过整个ChannelPipeline,但是在ChannelHandler的级别上,事件从一个ChannelHandler到下一个ChannelHandler的移动是由ChannelHandlerContext上的调用完成的。

下图展示了Channel或者ChannelPipeline进行的事件传播机制。

 

在上图中可以看出:

(1)事件传递给ChannelPipeline的第一个ChannelHandler

(2)ChannelHandler通过关联的ChannelHandlerContext传递事件给ChannelPipeline中的下一个ChannelHandler

(3)ChannelHandler通过关联的ChannelHandlerContext传递事件给ChannelPipeline中的下一个ChannelHandler

从上面的流程可以看出,如果通过ChannelChannelPipeline的实例来调用这些方法,它们肯定会在整个ChannelPipeline中传播。

那么是否可以跳过某些处理器呢?答案是肯定的。

通过减少ChannelHandler不感兴趣的事件的传递减少开销,并排除掉特定的对此事件感兴趣的处理器的处理以提升性能。想要实现从一个特定的ChannelHandler开始处理,必须引用与此ChannelHandler的前一个ChannelHandler关联的ChannelHandlerContext。这个ChannelHandlerContext将会调用与自身关联的ChannelHandler的下一个ChannelHandler,代码如下:

ChannelHandlerContext ctx = context;
ctx.write(msg);

直接调用ChannelHandlerContextwrite()方法,将会把缓冲区发送到下一个ChannelHandler

如下图,消息会将从下一个ChannelHandler开始流过ChannelPipeline,绕过所有在它之前的ChannelHandler

 

(1)执行ChannelHandlerContext方法调用。

(2)事件发送到了下一个ChannelHandler

(3)经过最后一个ChannelHandler后,事件从ChannelPipeline中移除。

当调用某个特定的ChannelHandler操作时,它尤为有用。

例如:

public class EchoServerHandler extends ChannelInboundHandlerAdapter {

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) {
		System.out.println(ctx.channel().remoteAddress() + " -> Server :" + msg);

        // 写消息到管道
		ctx.write(msg);// 写消息
		ctx.flush(); // 冲刷消息
		
		// 上面两个方法等同于 ctx.writeAndFlush(msg);
	}
	
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

		// 当出现异常就关闭连接
		cause.printStackTrace();
		ctx.close();
	}
}

总结

以上就是关于ChannelPipeline的源码分析,相信认真看完了,你就明白ChannelPipelineChannelChannelHandlerChannelHandlerContext之间的关系。下节我们继续来剖析 Netty 的源码。