netty学习笔记06 - netty模型-蒲公英云

netty模型

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

工作示意图

netty工作示意图说明

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

Netty入门实例

实例要求：

使用IDEA创建Netty项目
Netty服务器在6668端口监听，客户端能发送消息给服务器 “hello,服务器”
服务器可以回复消息给客户端“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();
    }
}