Pipeline添加与删除ChannelHandler

古城微笑少年丶 2023-06-29 10:59 37阅读 0赞

### 添加ChannelHandler ###

ChannelHandler是在服务端启动过程中，Channel初始化时通过addLast方法添加的。

public static void main(String[] args) throws Exception {
            // 配置服务端的 NIO线程组
            // boss线程组用于网络事件的监听
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            // worker线程组用于SocketChannel的网络读写
            EventLoopGroup workerGroup = new NioEventLoopGroup();
    
            try {
                // NIO服务端的辅助启动类，目的是降低服务端开发的复杂度
                ServerBootstrap b = new ServerBootstrap();
                // 配置两大线程组
                b.group(bossGroup, workerGroup)
                        // 配置服务端channel，在服务启动时通过反射创建channel实例
                        .channel(NioServerSocketChannel.class)
                        // 配置TCP基本属性
                        .childOption(ChannelOption.TCP_NODELAY, true)
                        // 客户端创建连接时绑定基本属性
                        .childAttr(AttributeKey.newInstance("childAttr"), "childAttrValue")
                        // 配置服务端启动过程逻辑处理器
                        .handler(new ServerHandler())
                        // 配置业务处理链 handler pipeline
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            public void initChannel(SocketChannel ch) {
                                // 配置业务处理handler，对数据流进行读写等处理
                                ch.pipeline().addLast(new ChannelInboundHandlerAdapter());
                                ch.pipeline().addLast(new ChannelOutboundHandlerAdapter());
                            }
                        });
    
                // 前面都是一些属性配置的逻辑，真正的服务端启动在此处开始
                // 绑定端口，正式启动server端服务
                ChannelFuture f = b.bind(8888).sync();
                // 同步等待，直至服务端监听端口关闭
                f.channel().closeFuture().sync();
            } finally {
                // 优雅退出，释放线程池资源
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }

@Override
        public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
            final AbstractChannelHandlerContext newCtx;
            synchronized (this) {
                // 判断是否重复添加，共享的除外（被@Shareable注解修饰）
                checkMultiplicity(handler);
    
                // filterName方法用来检查重复名称
                // newContext把ChannelHandler包装成ChannelContext节点
                newCtx = newContext(group, filterName(name, handler), handler);
    
                // 将节点添加到链表中tail节点前一个元素的位置
                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()) {
                    newCtx.setAddPending();
                    executor.execute(new Runnable() {
                        @Override
                        public void run() {
                            // 回调添加完成事件（用户自定义的handlerAdded方法）
                            callHandlerAdded0(newCtx);
                        }
                    });
                    return this;
                }
            }
            callHandlerAdded0(newCtx);
            return this;
        }

总结一下addLast方法的主要逻辑：

1.  判断节点是否重复添加，共享节点除外（被@Shareable注解修饰）。
2.  创建节点并添加至链表。
3.  回调添加完成事件。

--------------------

### 删除ChannelHandler ###

什么样的场景下需要删除ChannelHandler节点呢？

例如当我们添加一个类似于权限校验的节点，在节点内实现一些验证逻辑，当验证通过时，删除节点。

// 配置业务处理链 handler pipeline
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            public void initChannel(SocketChannel ch) {
                                 // 添加权限验证节点，验证通过后删除节点
                                 ch.pipeline().addLast(new AuthHandler());
                            }
                        });

public class AuthHandler extends SimpleChannelInboundHandler<ByteBuf> {
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf password) throws Exception {
            if (paas(password)) {
                // 验证通过调用删除节点逻辑
                ctx.pipeline().remove(this);
            } else {
                ctx.close();
            }
        }
    
        private boolean paas(ByteBuf password) {
            // 定义具体的验证逻辑
            return true;
        }
    }

remove

@Override
        public final ChannelPipeline remove(ChannelHandler handler) {
            // getContextOrDie(handler)：遍历链表，拿到节点
            remove(getContextOrDie(handler));
            return this;
        }
    
        @Override
        public final ChannelHandler remove(String name) {
            return remove(getContextOrDie(name)).handler();
        }
    
        @SuppressWarnings("unchecked")
        @Override
        public final <T extends ChannelHandler> T remove(Class<T> handlerType) {
            return (T) remove(getContextOrDie(handlerType)).handler();
        }
    
        private AbstractChannelHandlerContext remove(final AbstractChannelHandlerContext ctx) {
            assert ctx != head && ctx != tail;
    
            synchronized (this) {
                // 从双向链表中删除节点
                remove0(ctx);
    
                // 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.
                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;
                }
            }
            callHandlerRemoved0(ctx);
            return ctx;
        }
    
        private static void remove0(AbstractChannelHandlerContext ctx) {
            AbstractChannelHandlerContext prev = ctx.prev;
            AbstractChannelHandlerContext next = ctx.next;
            prev.next = next;
            next.prev = prev;
        }

总结一下remove方法的主要逻辑：

1.  遍历链表，找到节点。 
2.  使用双向链表的常规删除方式删除节点，head与tail节点不能删除。
3.  回调删除Handler事件。