netty学习笔记06 - netty模型

本是古典何须时尚 2023-07-06 04:48 17阅读 0赞

### netty模型 ###

Netty 主要基于主从 Reactors 多线程模型（如图）做了一定的改进，其中主从 Reactor 多线程模型有多个 Reactor

*  工作示意图  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Z5ajEzOTI1NDc1OTU3_size_16_color_FFFFFF_t_70]  
    netty工作示意图说明

1.  Netty抽象出两个线程池BossGruop专门负责接收客户端连接 ，WorkGroup专门负责网络的读写
2.  BossGroup和WorkGroup类型都是NioEventLoopGroup
3.  NioEventLoopGroup相当一个事件循环组，这个组含有多个事件循环，每个一个事件循环都是NioEventLoop
4.  NioEventLoop表示一个不断循环的执行处理任务的线程，每个NioEventLoop都有 一个selector，用去监听绑定在其上Socket的网络通讯
5.  NioEventLoopGroup可以用多个线程，即可以有多个NioEventLoop
6.  每个BossGroup执行的步骤有3步
    
    > 1.  轮询accept事件
    > 2.  处理accept事件，与client建立连接，生成NioSocketChannel，并将其注册到WorkGroup上的selector上
    > 3.  处理任务队列，即runAllTasks
7.  每个workGroup的NioEventLoop循环 执行的 步骤
    
    > 1.  轮询read,write事件
    > 2.  处理IO事件，即read,write事件，在对于的NioSocketChannel处理
    > 3.  处理任务队列的任务，即runAllTasks
8.  每个workGroup的NioEventLoop处理业务时，都 会使用pipeline(管道)，pipeline中包含了channel,即 通过pipeline可以 获取对于的channel,管道中维护了很多处理器

### Netty入门实例 ###

实例要求：

1.  使用IDEA创建Netty项目
2.  Netty服务器在6668端口监听，客户端能发送消息 给服务器 “hello,服务器”
3.  服务器可以回复消息给客户 端“hello,客户端”

代码入下：  
服务器：

public class NettyServer { 
    
        public static void main(String[] args) throws InterruptedException { 
            // 1.创建bossGroup和workGroup
            // boosGroup只处理连接请求，真正和客户端业务处理的交给workGroup完成
            // 两个都是无限轮询
            // boosGroup和WorkGroup含有线程（NioEventLoop）的个数默认是cpu核数*2
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
    
            try { 
                // 2.创建引导程序，服务器的启动对象配置参数，链式编程
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                // 设置2个线程组
                serverBootstrap.group(bossGroup, workerGroup)
                        // 设置服务器端的通道使用NioServerSocketChannle
                        .channel(NioServerSocketChannel.class)
                        // 设置线程队列得到的连接个数
                        .option(ChannelOption.SO_BACKLOG, 128)
                        // 设置活动连接的状态
                        .childOption(ChannelOption.SO_KEEPALIVE, true)
                        // 设置workGroup对应的Handle
                        .childHandler(new ChannelInitializer<SocketChannel>() { // 创建一个匿名的初始化对象
                            protected void initChannel(SocketChannel socketChannel) throws Exception { 
                                // 获取该连接通道pipeline，并添加处理器
                                socketChannel.pipeline().addLast(new NettyServerHandler());
                            }
                        });
                System.out.println("服务器准备完毕...");
                // 绑定一个端口并且同步, 生成了一个 ChannelFuture 对象
                // 启动服务器(并绑定端口)
                ChannelFuture channelFuture = serverBootstrap.bind(6669).sync();
                // 对关闭通道进行监听
                channelFuture.channel().closeFuture().sync();
            } finally { 
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

服务器的处理器和任务队列使用场景

public class NettyServerHandler extends ChannelInboundHandlerAdapter { 
    
        /** * 这个方法是读取客户端发送的消息 * @param ctx 上下文对象，含有pipeline，通道channel,地址 * @param msg 客户端发送的消息 * @throws Exception */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
    		// 任务队列使用 1.用户自定义的普通任务
            ctx.channel().eventLoop().execute(new Runnable() { 
                public void run() { 
                    ctx.writeAndFlush(Unpooled.copiedBuffer("hello,客户端", CharsetUtil.UTF_8));
                }
            });
    
            ByteBuf byteBuf = (ByteBuf) msg;
            System.out.println("收到客户端发送的消息："+ byteBuf.toString(CharsetUtil.UTF_8));
            System.out.println("客户端的地址是："+ ctx.channel().remoteAddress());
        }
    
        /** * 数据读取完毕 调用的方法 * @param ctx * @throws Exception */
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { 
            // 将数据写入到缓存，并刷新
            ByteBuf byteBuf = Unpooled.copiedBuffer("Hello,客户端", CharsetUtil.UTF_8);
            ctx.writeAndFlush(byteBuf);
        }
    
        /** * 发生异常调此方法 * @param ctx * @param cause * @throws Exception */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            // 关闭通道
            ctx.close();
        }
    }

客户端：

public class NettyClient { 
        public static void main(String[] args) throws InterruptedException { 
            // 客户端只需要一个事件循环组
            EventLoopGroup eventExecutors = new NioEventLoopGroup();
    
            // 创建客户端启动对象
            try { 
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(eventExecutors)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() { 
                            protected void initChannel(SocketChannel socketChannel) throws Exception { 
                                socketChannel.pipeline().addLast(new NettyClientHandler());
                            }
                        });
                System.out.println("客户端准备完毕...");
                ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6669);
                channelFuture.channel().closeFuture().sync();
            } finally { 
                eventExecutors.shutdownGracefully();
            }
        }
    }

客户端的处理器：

public class NettyClientHandler extends ChannelInboundHandlerAdapter { 
        /** * 当通道就绪就会触发该方法 * @param ctx * @throws Exception */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception { 
            ByteBuf byteBuf = Unpooled.copiedBuffer("hello,server: 我是客户端", CharsetUtil.UTF_8);
            ctx.writeAndFlush(byteBuf);
            System.out.println(byteBuf);
        }
    
        /** * 当通道有读取事件时，就会触发该方法 * @param ctx * @param msg * @throws Exception */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
            ByteBuf byteBuf = (ByteBuf) msg;
            System.out.println("服务器回复消息："+ byteBuf.toString(CharsetUtil.UTF_8));
            System.out.println("服务器的地址是：" + ctx.channel().remoteAddress());
        }
    
        /** * 有异常时 就会触发该方法 * @param ctx * @param cause * @throws Exception */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            ctx.close();
        }
    }

*  **Netty模型方案再说明**

1.  Netty抽象出2个线程池，BossGroup专门负责接收客户端 连接，WorkerGroup专门负责网络的读写操作 。
2.  NioEventLoop表示一个不断循环执行处理任务的线程，每个NioEventLoop都有一个selector，用于监听绑定在其上的Socket通道。
3.  NioEventLoop内部采用串行化设计，从消息的读取->解码->处理->编码->发送，始终由 IO 线程 NioEventLoop 负责
4.  NioEventLoopGroup 下包含多个 NioEventLoop
5.  每个 NioEventLoop 中包含有一个 Selector，一个 taskQueue
6.  每个 NioEventLoop 的 Selector 上可以注册监听多个 NioChannel
7.  每个 NioChannel 只会绑定在唯一的 NioEventLoop 上
8.  每个 NioChannel 都绑定有一个自己的 ChannelPipeline

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Z5ajEzOTI1NDc1OTU3_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200216100903149.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Z5ajEzOTI1NDc1OTU3,size_16,color_FFFFFF,t_70