Netty源码分析之ChannelPipeline(二)—ChannelHa…

2019-12-07 16:00:52来源：博客园阅读 ()

Netty源码分析之ChannelPipeline(二)—ChannelHandler的添加与删除

上篇文章中，我们对Netty中ChannelPipeline的构造与初始化进行了分析与总结，本篇文章我们将对ChannelHandler的添加与删除操作进行具体的的代码分析；

一、ChannelHandler的添加

下面是Netty官方的一段demo源码，可以看到在服务端初始化时执行了向ChannelPipeline中添加自定义channelHandler的操作。

            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 100)
                    .handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline p = ch.pipeline();
                            if (sslCtx != null) {
                                p.addLast(sslCtx.newHandler(ch.alloc()));
                            }
                            // p.addLast(new LoggingHandler(LogLevel.INFO));
                            // 向ChannelPipeline中添加自定义channelHandler
                            p.addLast(serverHandler);
                        }
                    });

我们可以看到上面的代码中调用ChannelPipeline的addLast方法实现了channelHandler的添加，下面我们就分析下addLast方法的具体源码实现

首先看下addLast方方法的具体源码实现

    public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
        final AbstractChannelHandlerContext newCtx;
        synchronized (this) {
            //判断handler是否被重复添加
            checkMultiplicity(handler);

            //创建ChannelHandlerContext节点  filterName检查名称是否重复
            newCtx = newContext(group, filterName(name, handler), handler);

            //双向链表中增加ChannelHandlerContext
            addLast0(newCtx);

            // If the registered is false it means that the channel was not registered on an eventLoop yet.
            // In this case we add the context to the pipeline and add a task that will call
            // ChannelHandler.handlerAdded(...) once the channel is registered.
            if (!registered) {
                newCtx.setAddPending();
                callHandlerCallbackLater(newCtx, true);
                return this;
            }

            EventExecutor executor = newCtx.executor();
            if (!executor.inEventLoop()) {//判断是否在同一线程中
                callHandlerAddedInEventLoop(newCtx, executor);
                return this;
            }
        }
        callHandlerAdded0(newCtx);
        return this;
    }

分析addLast方法代码可以看到，ChannelHandler的添加基本可以分为四步

1、验证ChannelHandler是否重复添加

我们看下checkMultiplicity方法的具体实现

    private static void checkMultiplicity(ChannelHandler handler) {
        if (handler instanceof ChannelHandlerAdapter) {
            ChannelHandlerAdapter h = (ChannelHandlerAdapter) handler;
            if (!h.isSharable() && h.added) {//如果该handler非共享且已经被添加
                throw new ChannelPipelineException(
                        h.getClass().getName() +
                        " is not a @Sharable handler, so can't be added or removed multiple times.");
            }
            h.added = true;//添加过之后,修改标识
        }
    }

2、创建一个HandlerContext对象

我们之前说过netty会把一个channelhandler封装成一个ChannelHandlerContext对象，如下面代码所示

newCtx = newContext(group, filterName(name, handler), handler);

封装对象时，我们可以给要添加的channelhandler起一个名字，filterName方法可以判断该handler的命名是否重复

    private String filterName(String name, ChannelHandler handler) {
        if (name == null) {
            return generateName(handler);//返回一个默认名称
        }
        checkDuplicateName(name);
        return name;
    }

checkDuplicateName方法会遍历链表中节点如果查询到有重复的name则会抛出异常

    private void checkDuplicateName(String name) {
        if (context0(name) != null) { //遍历节点,查找是否有重复name
            throw new IllegalArgumentException("Duplicate handler name: " + name);
        }
    }

    private AbstractChannelHandlerContext context0(String name) {
        AbstractChannelHandlerContext context = head.next;
        while (context != tail) {
            if (context.name().equals(name)) {
                return context;
            }
            context = context.next;
        }
        return null;
    }

3、向链表中添加添加context

前面进行了一系列判断后，通过addLast0方法我们把ChannelHandlerContext添加到pipeline中的双向链表中

    //相当于在tail节点前面插入一个节点，也就是addLast
    private void addLast0(AbstractChannelHandlerContext newCtx) {
        AbstractChannelHandlerContext prev = tail.prev;//拿到tail节点的前置节点
        newCtx.prev = prev;//把当前context的前置节点置为 prev
        newCtx.next = tail;//把当前context的后置节点置为tail
        prev.next = newCtx;//把prev节点的后置节点置为context
        tail.prev = newCtx;//把tail节点的前置节点置为context
    }

addLast0内部实现很简单，就是在tail节点前面插入一个节点，也就是把该ChannelHandlerContext放在链表的最后。

4、调用回调方法，通知添加成功

这一步就是当ChannelHandler添加到Pipeline中时调用，通过callHandlerAdded()回调方法通知ChannelHandler添加成功，执行handlerAdded()方法；

首先判断当前线程与eventloop线程是否一致，不一致的话封装成task提交给eventloop线程，是同一线程直接执行callHandlerAdded0方法，我们看下方法具体实现

    private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
        try {
            ctx.callHandlerAdded();//调用callHandlerAdded回调方法
        } catch (Throwable t) {
            boolean removed = false;
            try {
                remove0(ctx);//如果出现异常的话，把该ctx删除
                ctx.callHandlerRemoved();//调用callHandlerRemoved回调方法
                removed = true;
            } catch (Throwable t2) {
                if (logger.isWarnEnabled()) {
                    logger.warn("Failed to remove a handler: " + ctx.name(), t2);
                }
            }

            if (removed) {
                fireExceptionCaught(new ChannelPipelineException(
                        ctx.handler().getClass().getName() +
                        ".handlerAdded() has thrown an exception; removed.", t));
            } else {
                fireExceptionCaught(new ChannelPipelineException(
                        ctx.handler().getClass().getName() +
                        ".handlerAdded() has thrown an exception; also failed to remove.", t));
            }
        }
    }

    final void callHandlerAdded() throws Exception {
        // We must call setAddComplete before calling handlerAdded. Otherwise if the handlerAdded method generates
        // any pipeline events ctx.handler() will miss them because the state will not allow it.
        if (setAddComplete()) {//在添加handler之前，保证状态为可添加状态
            handler().handlerAdded(this);
        }
    }

通过上面代码实现，我们就可以通过重写ChannelHandler的handlerAdded方法，执行一些当ChannelHandler添加到Pipeline中时需要触发的功能逻辑。

二、ChannelHandler的删除

ChannelHandler的删除主要是通过ChannelPipeline的remove方法来实现的

我们先看下remove方法源码具体实现

    @Override
    public final ChannelPipeline remove(ChannelHandler handler) {
        remove(getContextOrDie(handler));//删除handler
        return this;
    }

private AbstractChannelHandlerContext remove(final AbstractChannelHandlerContext ctx) {
        //不能删除头节点和尾节点
        assert ctx != head && ctx != tail;

        //加锁，保证线程安全
        synchronized (this) {
            remove0(ctx);//在链表中删除context对象

            // If the registered is false it means that the channel was not registered on an eventloop yet.
            // In this case we remove the context from the pipeline and add a task that will call
            // ChannelHandler.handlerRemoved(...) once the channel is registered.
            //这里主要判断下该pipline对应channel是否已经注册到eventloop上
            if (!registered) {
                callHandlerCallbackLater(ctx, false);
                return ctx;
            }

            
            EventExecutor executor = ctx.executor();
            if (!executor.inEventLoop()) {//判断是否在同一线程中
                executor.execute(new Runnable() {
                    @Override
                    public void run() {
                        callHandlerRemoved0(ctx);
                    }
                });
                return ctx;
            }
        }
        //调用回调方法，通知handler已删除
        callHandlerRemoved0(ctx);
        return ctx;
    }

删除操作整个流程与添加类似

1、获取ChannelHandlerContext对象

    //根据传入的handler拿到其包装的ChannelHandlerContext对象
    private AbstractChannelHandlerContext getContextOrDie(ChannelHandler handler) {
        //根据context方法从链表中获取该handler的ChannelHandlerContext对象
        AbstractChannelHandlerContext ctx = (AbstractChannelHandlerContext) context(handler);
        if (ctx == null) {
            throw new NoSuchElementException(handler.getClass().getName());
        } else {
            return ctx;
        }
    }

    @Override
    public final ChannelHandlerContext context(ChannelHandler handler) {
        if (handler == null) {
            throw new NullPointerException("handler");
        }

        AbstractChannelHandlerContext ctx = head.next;
        //遍历链表拿到该handler封装的ChannelHandlerContext对象
        for (;;) {

            if (ctx == null) {
                return null;
            }

            if (ctx.handler() == handler) {
                return ctx;
            }

            ctx = ctx.next;
        }
    }

2、判断是否是首尾节点

首先判断删除的节点既不是头节点也不是尾节点

   //不能删除头节点和尾节点
   assert ctx != head && ctx != tail;

3、执行删除操作

然后通过remove0方法删除指定Context节点

    private static void remove0(AbstractChannelHandlerContext ctx) {
        AbstractChannelHandlerContext prev = ctx.prev;//获取当前节点的前置节点
        AbstractChannelHandlerContext next = ctx.next;//获取当前节点的后置节点
        prev.next = next;//把prev后置节点设置为next
        next.prev = prev;//把next前置节点设置为prev
    }

4、调用回调方法，通知删除成功

同样会判断当前线程与eventloop线程是否一致，不一致的话封装成task提交给eventloop线程

 EventExecutor executor = ctx.executor();
            if (!executor.inEventLoop()) {//判断是否在同一线程中
                executor.execute(new Runnable() {
                    @Override
                    public void run() {
                        callHandlerRemoved0(ctx);
                    }
                });
                return ctx;
            }

 //调用回调方法，通知handler已删除
 callHandlerRemoved0(ctx);

调用callHandlerRemoved()回调方法，通知触发handlerRemoved删除成功。

    private void callHandlerRemoved0(final AbstractChannelHandlerContext ctx) {
        // Notify the complete removal.
        try {
            ctx.callHandlerRemoved();//调用ctx中的回调方法
        } catch (Throwable t) {
            fireExceptionCaught(new ChannelPipelineException(
                    ctx.handler().getClass().getName() + ".handlerRemoved() has thrown an exception.", t));
        }
    }

    final void callHandlerRemoved() throws Exception {
        try {
            // Only call handlerRemoved(...) if we called handlerAdded(...) before.
            if (handlerState == ADD_COMPLETE) {
                handler().handlerRemoved(this);//通知handler删除成功事件
            }
        } finally {
            // Mark the handler as removed in any case.
            setRemoved();
        }
    }