【Netty】ChannelPipeline源码分析（五）

前言

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

• Netty 概述（一）
• Netty 架构设计（二）
• Netty Channel 概述（三）
• Netty ChannelHandler（四）

一、ChannelPipeline 接口

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

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

1.1 创建 ChannelPipeline

ChannelPipeline数据管道是与Channel管道绑定的，一个Channel通道对应一个ChannelPipeline，ChannelPipeline是在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)@Overridepublic void initChannel(SocketChannel ch) throws Exception {// 添加ChannelHandler到ChannelPipelinech.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类的实例。DefaultChannelPipeline是ChannelPipeline的默认实现。

pipeline是AbstractChannel的属性，内部维护着一个以AbstractChannelHandlerContext为节点的双向链表，创建的head和tail节点分别指向链表头尾，源码如下：

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();}...}...
}

从上述源码可以看到，TailContext和HeadContext都继承了AbstractChannelHandlerContext，并实现了ChannelHandler接口。AbstractChannelHandlerContext内部维护着next、prev链表指针和入站、出站节点方向等。其中TailContext实现了ChannelInboundHandler，HeadContext实现了ChannelOutboundHandler和ChannelInboundHandler。

1.2 ChannelPipeline 事件传输机制

通过ChannelPipeline的addFirst()方法来添加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);}//...}

1.2.1 处理出站事件

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

public class EchoServerHandler extends ChannelInboundHandlerAdapter {@Overridepublic 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方法。
上述方法最终会调用到DefaultChannelHandlerContext的write()方法。

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()方法，该方法最终调用ChannelOutboundHandler的write方法。

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

至此，处理完成了第一个节点的处理，开始执行下一个节点并不断循环。
所以，处理出站事件时，数据传输的方向是从尾部节点tail到头部节点head。

1.2.2 处理入站事件

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

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

从上述源码可以看到，处理的节点是头部节点head。AbstractChannelHandlerContext.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();}});}}

该方法最终调用ChannelInboundHandler的channelActive方法。

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 接口是联系ChannelHandler和ChannelPipeline 之间的纽带。

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

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

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

3.1 ChannelHandlerContext 与其他组件的关系

下图展示了ChannelPipeline 、Channel、ChannelHandler和ChannelHandlerContext 之间的关系做了如下说明：

1. Channel被绑定到ChannelPipeline 上。
2. 和Channel绑定的ChannelPipeline 包含了所有的ChannelHandler。
3. ChannelHandler。
4. 当添加ChannelHandler到ChannelPipeline 时，ChannelHandlerContext 被创建。

3.2 跳过某些 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);

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

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

在上图中可以看出：

1. 事件传递给ChannelPipeline 的第一个ChannelHandler；
2. ChannelHandler通过关联的ChannelHandlerContext 传递事件给ChannelPipeline 中的下一个ChannelHandler。
3. ChannelHandler通过关联的ChannelHandlerContext 传递事件给ChannelPipeline 中的下一个ChannelHandler。

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

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

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

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

直接调用ChannelHandlerContext 的write()方法，将会把缓冲区发送到下一个ChannelHandler。

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

1. 执行ChannelHandlerContext 方法调用。
2. 事件发送到了下一个ChannelHandler。
3. 经过最后一个ChannelHandler后，事件从ChannelPipeline 中移除。

当调用某个特定的ChannelHandler操作时，它尤为有用。
例如：

public class EchoServerHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {System.out.println(ctx.channel().remoteAddress() + " -> Server :" + msg);// 写消息到管道ctx.write(msg);// 写消息ctx.flush(); // 冲刷消息// 上面两个方法等同于 ctx.writeAndFlush(msg);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {// 当出现异常就关闭连接cause.printStackTrace();ctx.close();}
}

总结

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