Netty4之处理链

Dear 丶 2023-10-11 02:47 51阅读 0赞

本文是基于Netty4.1.x,Handler在Netty占据着很重要的位置，跟Servlet中的filter很像，通过Handler可以完成消息的编解码、拦截指定的消息、统一对日志错误进行处理、统一对请求进行计数。所有的Handler都实现ChannelHandler接口，分为两大类，ChannelInboundHandler与ChannelOutboundHandler,ChannelInboundHandler对从客户端发往服务器的报文进行处理，一般用来执行解码、读取客户端数据、进行业务处理等；ChannelOutboundHandler对从服务器发往客户端的报文进行处理，一般用来进行编码、发送报文到客户端。Netty中，可以注册多个handler。ChannelInboundHandler按照注册的先后顺序执行；ChannelOutboundHandler按照注册的先后顺序逆序执行，如下图所示，按照注册的先后顺序对Handler进行排序，request进入Netty后的执行顺序为：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FuZ2p1bnFpYW5n_size_16_color_FFFFFF_t_70][]

从上图可以看出，在入站时，request会依次经过相关的Inbound处理器，然后出站时response也经过相应的Outbound处理器。从上图中可以看出Handler组成了一个链表放在ChannelPipeplie中，下面看看ChannelPipe中的Handler如何组成链表的。下面是一段NettyServer初始化的代码：

public class NettyServer {
        public static void main(String[] args) throws Exception {
            new NettyServer().start("127.0.0.1", 8081);
        }
    
        public void start(String host, int port) throws Exception {
            ExecutorService executorService = Executors.newCachedThreadPool();
            EventLoopGroup bossGroup = new NioEventLoopGroup(0, executorService);//Boss I/O线程池,用于处理客户端连接,连接建立之后交给work I/O处理
            EventLoopGroup workerGroup = new NioEventLoopGroup(0, executorService);//Work I/O线程池
            EventExecutorGroup businessGroup = new DefaultEventExecutorGroup(2);//业务线程池
            ServerBootstrap server = new ServerBootstrap();//启动类
            server.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024).childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();
                    pipeline.addLast("decoder", new StringDecoder());
                    pipeline.addLast("encoder", new StringEncoder());
                    pipeline.addLast(businessGroup, new ServerHandler());
                }
            });
            server.childOption(ChannelOption.TCP_NODELAY, true);
            server.childOption(ChannelOption.SO_RCVBUF, 32 * 1024);
            server.childOption(ChannelOption.SO_SNDBUF, 32 * 1024);
            InetSocketAddress addr = new InetSocketAddress(host, port);
            server.bind(addr).sync().channel();//启动服务
        }
    }

从上面看出在initChannel时会调用 pipeline.addLast("decoder", new StringDecoder());逐个将Handler加入到Pipeline,下在看看 pipeline.addLast()这个方法：

@Override
    public final ChannelPipeline addLast(String name, ChannelHandler handler) {
        return addLast(null, name, handler);
    }
    
    @Override
    public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
        final AbstractChannelHandlerContext newCtx;
        synchronized (this) {
            checkMultiplicity(handler);
    
            newCtx = newContext(group, filterName(name, handler), handler);
    
            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;
    }
    
    private void addLast0(AbstractChannelHandlerContext newCtx) {
        AbstractChannelHandlerContext prev = tail.prev;
        newCtx.prev = prev;
        newCtx.next = tail;
        prev.next = newCtx;
        tail.prev = newCtx;
    }

上面代码片中 addLast()方法中先调用 checkMultiplicity(handler)检查一下是否有重复，然后将handler包装成一个AbstractChannelHandlerContext然后再调用 addLast0(newCtx);将newCtx放入到链表中，从addLast0方法中可以看出，处理链是一个双向链表。到这里处理链的执行逻辑及组成方式就说完了总结一下：

*  处理链是一个有顺的双向链表放在ChannelPipeline(DefaultChannelPipeline)中，消息上行或下行时依次会被相关的处理器处理；
 *  可以通过 channel.pipeline().addLast()、 channel.pipeline().addFirst()、channel.pipeline().addBefore()等方法向链表的指定位置添加处理器，也可以在处理过程中动态添加或删除处理器；
 *  可以为Handler指定处理的业务线程池；
 *  可以使用相关的api将消息在处理链路中流转，如ctx.fireChannelRead()表示将消息传递到下一个处理器，ctx.channel().pipeline()表示将消息放到pipeline让处理器从第一个处理器开始处理；

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FuZ2p1bnFpYW5n_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20190923190229193.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FuZ2p1bnFpYW5n,size_16,color_FFFFFF,t_70