7.2 源码分析
7.2.1 ChannelPipeline
首先看ChannelPipeline接口的关键方法,相似方法只列出一个:
ChannelPipeline addLast(String name, ChannelHandler handler);
ChannelPipeline remove(ChannelHandler handler);
ChannelHandler first();
ChannelHandlerContext firstContext();
ChannelHandler get(String name);
ChannelHandlerContext context(ChannelHandler handler);
Channel channel();
ChannelPipeline fireChannelRegistered();
ChannelFuture bind(SocketAddress localAddress);
DefaultChannelPipeline是ChannelPipeline的一个子类,回忆ChannelHandler的事件处理顺序,与双向链表的正向遍历和反向遍历顺序相同,可推知DefaultChannelPipeline使用了双向链表。事实上如此,所不同的是:链表中的节点并不是ChannelHandler而是ChannelHandlerContext。明白了这些,先看其中的字段:
final AbstractChannelHandlerContext head; // 双向链表头
final AbstractChannelHandlerContext tail; // 双向链表尾
private final Channel channel; // 对应Channel
// 线程池中的线程映射,记住这个映射是为了保证执行任务时使用同一个线程
private Map<EventExecutorGroup, EventExecutor> childExecutors;
private MessageSizeEstimator.Handle estimatorHandle; // 消息大小估算器,内部没有使用
private boolean firstRegistration = true; // 对应Channel首次注册到EventLoop
// ChannelHandler添加任务队列链表头部
private PendingHandlerCallback pendingHandlerCallbackHead;
// 注册到EventLoop标记,该值一旦设置为true后不再改变
private boolean registered;
此外还需要注意一个static字段:
private static final FastThreadLocal<Map<Class<?>, String>> nameCaches =
initialValue() -> { return new WeakHashMap<Class<?>, String>(); };
这是一个Netty内部定义的FastThreadLocal变量,以后会分析它的实现,现在先了解这样的事实:nameCaches是一个线程本地(局部)变量,也就是说每个线程都存有一份该变量,该变量是一个WeakHashMap,其中存放的是ChannelHandler的Class与字符串名称的映射关系。简单说就是每个线程都有一份Handler的Class与字符串名称的映射关系,之所以这样是为了避免使用复杂的CurrentHashMap也能实现并发安全。
首先看我们常用的addLast()方法:
public final ChannelPipeline addLast(EventExecutorGroup group, String name,
ChannelHandler handler) {
final AbstractChannelHandlerContext newCtx;
synchronized (this) {
// 检查Handler是否重复添加
checkMultiplicity(handler);
// 新建一个Context
newCtx = newContext(group, filterName(name, handler), handler);
// 实际的双向链表插入操作
addLast0(newCtx);
if (!registered) {
// 此时Channel还没注册的EventLoop中,而Netty的原则是事件在同一个EventLoop执行,
// 所以新增一个任务用于注册后添加
newCtx.setAddPending();
callHandlerCallbackLater(newCtx, true);
return this;
}
EventExecutor executor = newCtx.executor();
if (!executor.inEventLoop()) {
// 当前线程不是EventLoop线程
newCtx.setAddPending();
executor.execute( () -> { callHandlerAdded0(newCtx); } );
return this;
}
}
// 当前线程为EventLoop线程且已注册则直接触发HandlerAdd事件
callHandlerAdded0(newCtx);
return this;
}
public final ChannelPipeline addLast(String name, ChannelHandler handler) {
// 线程池为null则使用channel注册到的EventLoop
return addLast(null, name, handler);
}
先看其中的checkMultiplicity()方法,功能是保证ChannelPipeline中至多只有一个同一类型的非共享Handler,代码如下:
private static void checkMultiplicity(ChannelHandler handler) {
if (handler instanceof ChannelHandlerAdapter) { // 为什么只对Adapter?
ChannelHandlerAdapter h = (ChannelHandlerAdapter) handler;
if (!h.isSharable() && h.added) {
// 非共享且已被添加到pipeline中
throw new ChannelPipelineException("...");
}
h.added = true;
}
}
filterName()方法对Handler名称进行重复检查,generateName()生成形如:HandlerClassName#0、HandlerClassName#1的Handler字符串名称,checkDuplicateName()检查名称是否已使用,也就是说pipeline中Handler的名称也要求满足唯一性。代码如下:
private String filterName(String name, ChannelHandler handler) {
if (name == null) {
return generateName(handler);
}
checkDuplicateName(name);
return name;
}
checkDuplicateName()代码如下:
private void checkDuplicateName(String name) {
if (context0(name) != null) {
throw new IllegalArgumentException("Duplicate handler name: " + name);
}
}
// 双向链表中查找是否已有该名称的context
private AbstractChannelHandlerContext context0(String name) {
AbstractChannelHandlerContext context = head.next;
while (context != tail) {
if (context.name().equals(name)) {
return context;
}
context = context.next;
}
return null;
}
generateName()代码如下:
private String generateName(ChannelHandler handler) {
Map<Class<?>, String> cache = nameCaches.get(); // 获得ThreadLocal变量
Class<?> handlerType = handler.getClass();
String name = cache.get(handlerType);
if (name == null) {
name = generateName0(handlerType); // 生成HandlerClassName#0
cache.put(handlerType, name);
}
// HandlerClassName#0已有,则末尾编号加1
if (context0(name) != null) {
String baseName = name.substring(0, name.length() - 1);
for (int i = 1;; i ++) {
String newName = baseName + i;
if (context0(newName) == null) {
name = newName;
break;
}
}
}
return name;
}
// 生成HandlerClassName#0
private static String generateName0(Class<?> handlerType) {
return StringUtil.simpleClassName(handlerType) + "#0";
}
再看newContext()方法,返回一个默认Contenxt,其构造方法需要传入一个EventExecutor用于执行Handler中事件处理代码。childExecutor()正是用来从线程池中分配这个EventExecutor,代码如下:
private AbstractChannelHandlerContext newContext(EventExecutorGroup group, String name, ChannelHandler handler) {
return new DefaultChannelHandlerContext(this, childExecutor(group), name, handler);
}
private EventExecutor childExecutor(EventExecutorGroup group) {
if (group == null) {
return null;
}
Boolean pinEventExecutor = channel.config().getOption(ChannelOption.SINGLE_EVENTEXECUTOR_PER_GROUP);
if (pinEventExecutor != null && !pinEventExecutor) {
// Channel参数配置非同一个线程处理,不建议开启
return group.next();
}
Map<EventExecutorGroup, EventExecutor> childExecutors = this.childExecutors;
if (childExecutors == null) {
childExecutors = this.childExecutors = new IdentityHashMap<EventExecutorGroup, EventExecutor>(4);
}
// 保证pipeline中的事件为同一个EventExecutor处理,可视为将EventExecutor绑定到pipeline
EventExecutor childExecutor = childExecutors.get(group);
if (childExecutor == null) {
childExecutor = group.next();
childExecutors.put(group, childExecutor);
}
return childExecutor;
}
接着看实际的双向链表插入操作addLast0()操作:
private void addLast0(AbstractChannelHandlerContext newCtx) {
AbstractChannelHandlerContext prev = tail.prev;
newCtx.prev = prev;
newCtx.next = tail;
prev.next = newCtx;
tail.prev = newCtx;
}
DefaultChannelPipeline的双向链表初始化持有头部和尾部节点,这两个节点对用户不可见,也就是说,用户addLast只是将节点插入尾部节点之前,addFirst将节点插入头部节点之后。明白了这些,代码便易于理解。
接着看callHandlerCallbackLater()方法,当我们在Channel注册到之前添加或删除Handler时,此时没有EventExecutor可执行HandlerAdd或HandlerRemove事件,所以Netty为此事件生成一个相应任务等注册完成后在调用执行任务。添加或删除任务可能有很多个,DefaultChannelPipeline使用一个链表存储,链表头部为先前的字段pendingHandlerCallbackHead
,代码如下:
// 参数added为True表示HandlerAdd任务,False表示HandlerRemove任务
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 { // 插入到链表尾部
while (pending.next != null) {
pending = pending.next;
}
pending.next = task;
}
}
以HandlerAdd任务为例分析任务部分的代码(HandlerRemove可类比):
private abstract static class PendingHandlerCallback implements Runnable {
final AbstractChannelHandlerContext ctx;
PendingHandlerCallback next;
PendingHandlerCallback(AbstractChannelHandlerContext ctx) { this.ctx = ctx;}
abstract void execute();
}
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()) {
// 当前线程为EventLoop线程,调用HandlerAdd事件
callHandlerAdded0(ctx);
} else {
try {
executor.execute(this); // 否则提交一个任务,任务执行run()方法
} catch (RejectedExecutionException e) {
logger.warn("...");
remove0(ctx); // 异常时,将已添加的Handler删除
ctx.setRemoved();
}
}
}
}
callHandlerAdded0()方法执行实际的调用事件操作,作为addLast()的最后一个方法,代码如下:
private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
try {
ctx.handler().handlerAdded(ctx); // 调用事件处理
ctx.setAddComplete();
} catch (Throwable t) {
boolean removed = false; // 异常时删除Context,尽量恢复现场
try {
remove0(ctx); // 实际双向链表删除操作
try {
ctx.handler().handlerRemoved(ctx); // 调用事件处理
} finally {
ctx.setRemoved();
}
removed = true;
} catch (Throwable t2) {
logger.warn("Failed to remove a handler: " + ctx.name(), t2);
}
if (removed) {
fireExceptionCaught(new ChannelPipelineException("handlerAdded() has thrown an exception; removed."));
} else {
fireExceptionCaught(new ChannelPipelineException("handlerAdded() has thrown an exception; also failed to remove."));
}
}
}
终于分析完addLast()方法,我们经常使用的不起眼一行代码,背后的流程却很长。分析完这个方法的代码,其他方法的代码,我们可推断:remove()作为addXXX()的逆方法,其处理过程可推断为:找到对应Context节点,执行实际的双向链表删除操作,如果非注册则新增一个HandlerRemove任务并链接到任务链表尾部,如果已注册但Context需求线程非EventLoop,提交一个调用任务到需求线程,如果已注册且需求线程为EventLoop直接调用事件处理。first()返回的是链表头部的下一个Handler即用户可见的首个Handler,last()返回链表尾部的前一个Handler即用户可见的最后一个Handler,firstContext()和lastContext()同理,只是返回Context。get(handlerName)返回相应名称的Handler,context(handerlNmae)返回相应名称的Context,操作都是从双向链表头部进行遍历查找。
再看一下fireXXX方法和bind等事件触发方法的代码:
@Override
public final ChannelPipeline fireChannelRegistered() {
// 入站事件从双向链表头部处理
AbstractChannelHandlerContext.invokeChannelRegistered(head);
return this;
}
@Override
public final ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
// 出站事件从双向链表尾部处理
return tail.bind(localAddress, promise);
}
由于头部和尾部节点都是ChannelHandlerContext,具体的事件触发处理都委托给head和tail处理,将在之后一节进行分析。至此,接口中的方法已分析完毕,是不是还差点什么?仔细回想一下,在addXXX()方法中有待执行的HandlerAdd和HandlerRemove任务,它们怎么执行的呢?DefaultChannelPipeline提供了invokeHandlerAddedIfNeeded()方法:
final void invokeHandlerAddedIfNeeded() {
assert channel.eventLoop().inEventLoop();
if (firstRegistration) {
firstRegistration = false;
// 至此,channel已注册到EventLoop,可以执行任务
callHandlerAddedForAllHandlers();
}
}
private void callHandlerAddedForAllHandlers() {
final PendingHandlerCallback pendingHandlerCallbackHead;
synchronized (this) {
assert !registered; // 必须为非注册
registered = true; // 至此则说明已注册
pendingHandlerCallbackHead = this.pendingHandlerCallbackHead;
this.pendingHandlerCallbackHead = null; // 帮助垃圾回收
}
// 用一个局部变量保存任务链表头部是因为以下代码如果在synchronized块内,则当用户在
// 非EventLoop中执行HandlerAdd()方法而该方法中又新增一个handler时不会发生死锁
PendingHandlerCallback task = pendingHandlerCallbackHead;
while (task != null) {
task.execute(); // 遍历链表依次执行
task = task.next;
}
}
invokeHandlerAddedIfNeeded()方法在以下两种情况被调用:(1).AbstractUnsafe的register事件框架,当Channel注册到EventLoop之前会被调用,确保异步注册操作一旦完成就触发HandlerAdd事件;(2).双向链表头部节点的channelRegistered()方法(为什么此时调用,双重保护?)。
DefaultChannelPipeline还有最后一个方法destroy(),将pipeline中的所有节点销毁,顺序由尾部向头部并触发HandlerRemove事件,代码如下:
private synchronized void destroy() {
destroyUp(head.next, false);
}
// 参数inEventLoop应理解为是否直接执行本段代码的for循环部分,也就是说为true时不需要提交
// 一个destroyUp任务,为False时则需要判断Handler的执行线程是否为EventLoop线程
private void destroyUp(AbstractChannelHandlerContext ctx, boolean inEventLoop) {
final Thread currentThread = Thread.currentThread();
final AbstractChannelHandlerContext tail = this.tail;
for (;;) {
if (ctx == tail) {
destroyDown(currentThread, tail.prev, inEventLoop);
break;
}
final EventExecutor executor = ctx.executor();
if (!inEventLoop && !executor.inEventLoop(currentThread)) {
final AbstractChannelHandlerContext finalCtx = ctx;
// destroyUp()的for循环部分需在executor内执行,所以置True
executor.execute( () -> { destroyUp(finalCtx, true); } );
break;
}
ctx = ctx.next;
inEventLoop = false; // 每次都悲观的认为下一个Handler的处理线程会是另外一个线程
}
}
private void destroyDown(Thread currentThread, AbstractChannelHandlerContext ctx, boolean inEventLoop) {
// 至此,已经到达双向链表尾部,可确定入站事件已在删除操作进行之前传播完毕
final AbstractChannelHandlerContext head = this.head;
for (;;) {
if (ctx == head) {
break;
}
// 这部分代码实质与up部分一致,采用两种表现形式容易引起困惑
// 本质上 (!a && !b) == (a || b)
final EventExecutor executor = ctx.executor();
if (inEventLoop || executor.inEventLoop(currentThread)) {
synchronized (this) {
remove0(ctx);
}
callHandlerRemoved0(ctx);
} else {
final AbstractChannelHandlerContext finalCtx = ctx;
executor.execute(() -> { destroyDown(Thread.currentThread(), finalCtx, true); });
break;
}
ctx = ctx.prev;
inEventLoop = false;
}
}
这部分代码晦涩难懂,考虑这样一种情况,当我们由尾部向头部删除节点时,有一个入站事件正从头部向尾部传播,由于从尾部开始删除了某些节点,入站事件的处理流程被破坏。这部分代码正是为了处理这种情况,所以首先从头部向尾部遍历,确保没有入站事件,此时才从尾部向头部进行删除销毁操作。
这部分代码还为了保证事件在正确的线程中执行,假设有如下pipeline:
HEAD --> [E1] H1 --> [E2] H2 --> TAIL
其中E1和E2为两个线程,则必须保证Handler1中的事件在E1执行,Handler2中的事件在E2执行,而Head和Tail的事件在Channel注册到的EventLoop中执行。
网友评论