导读
原创文章，转载请注明出处。
本文源码地址：https://gitee.com/wangjianxin199003/netty-source-code-analysis.git
本文所使用的netty版本4.1.6.Final：带注释的netty源码，https://gitee.com/wangjianxin199003/netty
Pipeline
这个词翻译过来就是“流水线”的意思，读到这里有了解过设计模式的同学应该已经想到了，这里用到的是“责任链模式”。本文我们将以DefaultChannelPipeline
为例看一下Pipeline
的构造以及其中重要的数据结构。
1 和Pipeline相关的其他组件
1.1 ChannelHandler
这是ChannelHandler
中的注释，翻译过来就是“处理IO事件或者拦截IO操作，并且将其向ChannelPipeline
中的下一个handler
传递”，说白了就是在责任链中注册的一系列回调方法。
Handles an I/O event or intercepts an I/O operation, and forwards it to its next handler in its ChannelPipeline
这里的I/O event
就是很多书中提到的“入站事件”，而I/O operation
就是很多书中提到的“出站事件”，前面我说过，这里我并不准备这么叫，按我的理解我习惯把这两者称之为“事件”和“命令”。很显然这里event
和operation
的含义是不一样的，event
更多地多表示事件发生了，我们被动地收到，而operation
则表示我们主动地发起一个动作或者命令。
1.2 ChannelHandlerContext
每一个ChannelHandler
在被添加进ChannelPipeline
时会被包装进一个ChannelHandlerContext
。有两个特殊的ChannelHandlerContext
除外，分别是HeadContext
和TailContext
，HeadContext
继承了ChannelInBoundHandler
和ChannelOutBoundHandler
，而TailContext
继承了ChannelInBoundHandler
。每个ChannelHandlerContext
中有两个指针next
和prev
，这是用来将多个ChannelHandlerContext
构成双向链表的。
2 Pipeline的构造方法
我们以DefaultChannelPipeline
为例，从它的构造方法开始。这里首先将Channel
保存到Pipeline
的属性中，又初始化了两个属性succeedFuture
和voidPromise
。这是两个特殊的可以共享的Promise
，这两个Promise
不是重点，不理解也没关系。
接下来的tail
和head
是两个特殊的ChannelHandlerContext
，这两个是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在执行完构造方法以后的结构如下图所示，head
和tail
构成了最简单的双向链表。
图中蓝色填充的就是ChannelHandlerContext
，目前只有HeadContext
和TailContext
，ChannelHandlerContext
中的较窄的矩形表示ChannelHandler
，由于HeadContext
和TailContext
并没有包含ChannelHandler
，而是继承ChannelHandler
，所以这里我们用虚线表示。上下贯通的ChannelHandler
表示既是ChannelInBoundHandler
又是ChannelOutBoundHandler
，只有上半部分的表示是ChannelInBoundHandler
，只有下半部分的表示是ChannelOutBoundHandler
。
3 添加ChannelHandler
在ChannelPipeline
中有很多以add
开头的方法，这些方法就是向ChannelPipeline
中添加ChannelHandler
的方法。
addAfter
：向某个ChannelHandler
后边添加addBefore
：向某个ChannelHandler
前面添加addFirst
：添加到头部，不能在head
的前面，而是紧挨着head
，在head
的后面addLast
：添加到尾部，不能在tail
的后面，而是紧挨着tail
，在tail
的前面
我们以最常用的的addLast
方法为例来分析一下Pipeline
中添加ChannelHandler
的操作。这里所贴出的addLast
方法其实我们已经在“服务端启动流程”这篇文章中打过照面了。方法参数中的EventExecutorGroup
意味着我们可以为这个ChannelHandler
单独设置Executor
而不使用Channel
所绑定的EventLoop
，一般情况下我们不这么做，所以group
参数为null
。
这里先把ChannelHandler
包装成ChannelHandlerContext
，再添加到尾部，随后调用ChannelHandler
的handlerAdded
方法。
在调用handlerAdded
方法时有一点问题，添加ChannelHandler
的操作不需要在EventLoop
线程中进行，而handlerAdded
方法则必须在EventLoop
线程中进行。也就是说存在添加Handler
时还未绑定EventLoop
的情况，此时则调用newCtx.setAddPending()
将当前HandlerContext
设置为ADD_PENDING
状态，并且调用callHandlerCallbackLater(newCtx, true)
将一个执行handlerAdded方法的异步任务添加到一个单向链表中，即pendingHandlerCallbackHead
这个链表，这个链表里的异步任务什么时候会执行呢，咱们接着看下去。
如果当前已经绑定了EventLoop
，则看当前调用线程是否为EventLoop
线程，如果不是则向EventLoop
提交一个异步任务调用callHandlerAdded0
方法，否则直接调用callHandlerAdded0
方法。
下面咱们依次分析一下newContext
，callHandlerCallbackLater
和callHandlerAdd0
方法。
@Override
public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
final AbstractChannelHandlerContext newCtx;
synchronized (this) {
//先把`handler`包装成`HandlerContext`
newCtx = newContext(group, filterName(name, handler), handler);
//添加到尾部
addLast0(newCtx);
//如果还未绑定`EventLoop`则稍后再发起对`HandlerAdded`方法的调用。
if (!registered) {
newCtx.setAddPending();
callHandlerCallbackLater(newCtx, true);
return this;
}
//如果已经绑定了EventLoop，并且当前线程非EventLoop线程的话就提交一个异步任务，就发起一个异步任务去调用HandlerAdded方法。
EventExecutor executor = newCtx.executor();
if (!executor.inEventLoop()) {
newCtx.setAddPending();
executor.execute(new Runnable() {
@Override
public void run() {
callHandlerAdded0(newCtx);
}
});
return this;
}
}
//如果当前线程是EventLoop线程，就直接调用HandlerAdded方法。
callHandlerAdded0(newCtx);
return this;
}
3.1 newContext
先看来一下newContext
方法，这里直接调用了DefaultChannelHandlerContext
的构造方法，咱们跟进去看看。
private AbstractChannelHandlerContext newContext(EventExecutorGroup group, String name, ChannelHandler handler) {
return new DefaultChannelHandlerContext(this, childExecutor(group), name, handler);
}
在DefaultChannelHandlerContext
的构造方法中又调用了父类AbstractChannelHandlerContext
的构造方法，保存了handler
属性。在调用父类构造方法之前调用了isInboud
和isOutbound
方法判断当前的Handler
是否为ChannelInBoundHandler
或者ChannelOutBoundHandler
，这两个方法很简单，不再展开。
DefaultChannelHandlerContext(
DefaultChannelPipeline pipeline, EventExecutor executor, String name, ChannelHandler handler) {
super(pipeline, executor, name, isInbound(handler), isOutbound(handler));
if (handler == null) {
throw new NullPointerException("handler");
}
this.handler = handler;
}
接下来看AbstractChannelHandlerContext
的构造方法，这里非常简单，保存了几个属性，咱们看一下ordered
这个属性。ordered
表示EventExecutor
在执行异步任务时是否按添加顺序执行，这里一般情况下executor
为null
，表示使用Channel
所绑定的EventLoop
线程，而EventLoop
线程都是OrderedEventExecutor
的实现类。所以这里我们不考虑ordered
为false
的情况。
AbstractChannelHandlerContext(DefaultChannelPipeline pipeline, EventExecutor executor, String name,
boolean inbound, boolean outbound) {
this.name = ObjectUtil.checkNotNull(name, "name");
this.pipeline = pipeline;
this.executor = executor;
this.inbound = inbound;
this.outbound = outbound;
// Its ordered if its driven by the EventLoop or the given Executor is an instanceof OrderedEventExecutor.
ordered = executor == null || executor instanceof OrderedEventExecutor;
}
上面提到了ChannelHandlerContext
可以在构造方法里单独指定EventExecutor
，如果没有单独指定的话就使用Channel
所绑定的EventLoop
，代码在哪里呢，就在AbstractChannelHandlerContext#executor
方法，非常简单，如果没有为当前ChannelHandler
指定excutor
则返回Channel
所绑定的EventLoop
。
@Override
public EventExecutor executor() {
if (executor == null) {
return channel().eventLoop();
} else {
return executor;
}
}
3.2 callHandlerCallbackLater
在添加完ChannelHandler
之后将调用ChannledHandler
的handlerAdded
方法，但是此时有可能还未绑定EventLoop
，而handlerAdded
方法的调用必须在EventLoop
线程内执行，此时就需要调用callHandlerCallbackLater
方法在pendingHandlerCallbackHead
链表中添加一个PendingHandlerAddedTask
。
private void callHandlerCallbackLater(AbstractChannelHandlerContext ctx, boolean added) {
assert !registered;
PendingHandlerCallback task = added ? new PendingHandlerAddedTask(ctx) : new PendingHandlerRemovedTask(ctx);
PendingHandlerCallback pending = pendingHandlerCallbackHead;
if (pending == null) {
pendingHandlerCallbackHead = task;
} else {
// Find the tail of the linked-list.
while (pending.next != null) {
pending = pending.next;
}
pending.next = task;
}
}
接下来咱们看一下PendingHandlerAddedTask
的代码，逻辑在execute
方法里，这里直接调用了callHandlerAdded0
。
private final class PendingHandlerAddedTask extends PendingHandlerCallback {
PendingHandlerAddedTask(AbstractChannelHandlerContext ctx) {
super(ctx);
}
@Override
public void run() {
callHandlerAdded0(ctx);
}
@Override
void execute() {
EventExecutor executor = ctx.executor();
if (executor.inEventLoop()) {
callHandlerAdded0(ctx);
} else {
try {
executor.execute(this);
} catch (RejectedExecutionException e) {
}
}
}
}
3.3 callHandlerAdded0
不管是在未绑定EventLoop
的情况下延迟调用handlerAdded
还是在已经绑定了EventLoop
的情况下立即调用HandlerAdded
，最终都会调用到callHandlerAdded0
方法。这里干了两件事，一是调用ChannelHandler
的handlerAdded
方法，二是将HandlerContext
的状态设置为ADD_COMPLETE
状态。
private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
try {
ctx.handler().handlerAdded(ctx);
ctx.setAddComplete();
} catch (Throwable t) {
}
3.4 添加多个ChannelHandler后的Pipeline
还记得咱们的“Netty整体架构图”吗，在这里咱们把Pipeline
部分单独放大拿出来看一下，在添加完多个ChannelHandler
之后，Pipeline
的结构是这样的。
4 删除ChannelHandler
在Pipeline
中有几个以remove
开头的方法，这些方法的作用就是删除ChannelHandler
。
remove(ChannelHandler handler)
：从head
向tail
查找，用==
判断是否为同一实例，只删除第1个。remove(Class<T> handlerType)
：从head
向tail
查找，用isAssignableFrom
方法判断是否为符合条件的类型，只删除第1个。remove(String name)
：从head
向tail
查找，用name
精确匹配查找，只删除第1个，因为name
不能重复，所以这里删除第1个也是唯一的1个。removeFirst
：删除head
的后一个，不能删除tail
。removeLast
：删除tail
的前一个，不能删除head
。
上述无论哪种删除方式在查找到对应的HandlerContext
后都会调用到remove(final AbstractChannelHandlerContext ctx)
方法，查找过程比较简单，咱们不再展开，直接看remove(final AbstractChannelHandlerContext ctx)
方法。
看看这个方法的实现，是不是和addLast(EventExecutorGroup group, String name, ChannelHandler handler)
很相似，非常相似。首先从双向链表中删除ChannelHandlerContext
，再调用callHandlerRemoved0
方法，callHandlerRemoved0
方法内会调用handlerRemoved
方法，这个调用必须在EventLoop
线程内进行。如果删除时还未绑定EventLoop
则添加一个异步任务到链表pendingHandlerCallbackHead
中。
如果已经绑定了EventLoop
并且当前线程非EventLoop
线程则向EventLoop
提交一个异步任务，否则直接调用callHandlerRemoved0
方法。
private AbstractChannelHandlerContext remove(final AbstractChannelHandlerContext ctx) {
synchronized (this) {
//从双向链表中删除`ChannelHandlerContext`
remove0(ctx);
//如果还未绑定`EventLoop`，则稍后调用`handlerRemoved`方法
if (!registered) {
callHandlerCallbackLater(ctx, false);
return ctx;
}
//如果已经绑定了`EventLoop`，但是当前线程非`EventLoop`线程的话，就发起一个异步任务调用callHandlerRemoved0方法
EventExecutor executor = ctx.executor();
if (!executor.inEventLoop()) {
executor.execute(new Runnable() {
@Override
public void run() {
callHandlerRemoved0(ctx);
}
});
return ctx;
}
}
//如果当前线程就是`EventLoop`线程，则直接调用callHandlerRemoved0方法。
callHandlerRemoved0(ctx);
return ctx;
}
callHandlerCallbackLater
方法咱们前面已经分析过，和添加ChannelHandler
时不同的是，这里向链表添加的是PendingHandlerRemovedTask
，这个类也很简单，不再展开。
这里咱们只看一下callHandlerRemoved0
方法。这个方法很简单，调用handlerRemoved
方法，再把ChannelHandlerContext
的状态设置为REMOVE_COMPLETE
。
private void callHandlerRemoved0(final AbstractChannelHandlerContext ctx) {
// Notify the complete removal.
try {
try {
ctx.handler().handlerRemoved(ctx);
} finally {
ctx.setRemoved();
}
} catch (Throwable t) {
fireExceptionCaught(new ChannelPipelineException(
ctx.handler().getClass().getName() + ".handlerRemoved() has thrown an exception.", ));
}
}
5 pendingHandlerCallbackHead链表中的任务什么时候调用
在AbstractUnsafe
的register0
方法中，在绑定EventLoop
以后，会调用pipeline.invokeHandlerAddedIfNeeded()
方法，我们看一下pipeline.invokeHandlerAddedIfNeeded()
方法。
private void register0(ChannelPromise promise) {
try {
// 去完成那些在绑定EventLoop之前触发的添加handler操作，这些操作被放在pipeline中的pendingHandlerCallbackHead中，是个链表
pipeline.invokeHandlerAddedIfNeeded();
}
invokeHandlerAddedIfNeeded
方法调用了callHandlerAddedForAllHandlers
方法，咱们接着看下去。
final void invokeHandlerAddedIfNeeded() {
assert channel.eventLoop().inEventLoop();
if (firstRegistration) {
firstRegistration = false;
callHandlerAddedForAllHandlers();
}
}
callHandlerAddedForAllHandlers
方法的逻辑咱就不再展开来说了，非常简单，就是遍历pendingHandlerCallbackHead
这个单向链表，依次调用每个元素的execute
方法，并且清空这个单向链表。
private void callHandlerAddedForAllHandlers() {
final PendingHandlerCallback pendingHandlerCallbackHead;
synchronized (this) {
assert !registered;
registered = true;
pendingHandlerCallbackHead = this.pendingHandlerCallbackHead;
this.pendingHandlerCallbackHead = null;
}
PendingHandlerCallback task = pendingHandlerCallbackHead;
while (task != null) {
task.execute();
task = task.next;
}
}
6 总结
Pipeline
中最重要的数据结构就是由多个ChannelHandlerContext
组成的双向链表，而每个ChannelHandlerContext
中包含一个ChannelHandler
，ChannelHandler
既可以添加也可以删除。在Pipeline
中有两个特殊的ChannelHandlerContext
分别是HeadContext
及TailContext
，这两个ChannelHandlerContext
中不包含ChannelHandler
，而是采用继承的方式。HeadContext
实现了ChannelOutBoundHandler
和ChannelInBoundHandler
，而TailContext
实现了ChannelInBoundHandler
。
关于作者
王建新，转转架构部资深Java工程师，主要负责服务治理、RPC框架、分布式调用跟踪、监控系统等。爱技术、爱学习，欢迎联系交流。
长按识别二维码关注公众号“种代码”，收听最新文章，获取作者微信深入交流，原创文章，码字不易，别忘了点右下角“在看”。
