美文网首页nettyNettyJava学习笔记
自顶向下深入分析Netty(七)--ChannelPipelin

自顶向下深入分析Netty(七)--ChannelPipelin

作者: Hypercube | 来源:发表于2016-11-02 15:45 被阅读942次

    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中执行。

    相关文章

      网友评论

        本文标题:自顶向下深入分析Netty(七)--ChannelPipelin

        本文链接:https://www.haomeiwen.com/subject/nwwvuttx.html