Netty代码基本使用及调用流程

╰+攻爆jí腚メ 2022-11-15 03:41 203阅读 0赞

# 为什么要使用netty？ #

1.  NIO 的类库和 API 繁杂， 使用麻烦；
2.  开发工作量和难度都非常大，异常处理繁琐
3.  Netty对NIO进行了良好的封装，解决了NIO中繁琐的工作。

# netty使用场景 #

Dubbo，RocketMQ等分布式系统中，各个节点之间需要远程服务调用。

# 代码 #

## 服务端 ##

public class NettyServer {
    
        public static void main(String[] args) throws Exception {
    
            // bossGroup只是处理连接请求 ,真正的和客户端业务处理，会交给workerGroup完成
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            //workerGroup个数默认为cpu核数的两倍
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                //创建服务器端的启动对象
                ServerBootstrap bootstrap = new ServerBootstrap();
                //使用链式编程来配置参数
                bootstrap.group(bossGroup, workerGroup) //设置两个线程组
                        .channel(NioServerSocketChannel.class) //使用NioServerSocketChannel作为服务器的通道实现
                        // 初始化服务器连接队列大小，服务端处理客户端连接请求是顺序处理的,所以同一时间只能处理一个客户端连接。
                        // 多个客户端同时来的时候,服务端将不能处理的客户端连接请求放在队列中等待处理
                        .option(ChannelOption.SO_BACKLOG, 1024)
                        .childHandler(new ChannelInitializer<SocketChannel>() {//创建通道初始化对象，设置初始化参数
    
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                //对workerGroup的SocketChannel设置处理器
                                ch.pipeline().addLast(new NettyServerHandler());
                            }
                        });
                System.out.println("netty server start。。");
                //绑定一个端口并且同步, 生成了一个ChannelFuture异步对象，通过isDone()等方法可以判断异步事件的执行情况
                //启动服务器(并绑定端口)，bind是异步操作，sync方法是等待异步操作执行完毕
                ChannelFuture cf = bootstrap.bind(9000).sync();
                //对通道关闭进行监听，closeFuture是异步操作，监听通道关闭
                // 通过sync方法同步等待通道关闭处理完毕，这里会阻塞等待通道关闭完成
                cf.channel().closeFuture().sync();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
    }
    
    public class NettyServerHandler extends ChannelInboundHandlerAdapter {
        /**
         * 当客户端连接服务器完成就会触发该方法
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            System.out.println("客户端连接通道建立完成");
        }
    
        /**
         * 读取客户端发送的数据
         *
         * @param ctx 上下文对象, 含有通道channel，管道pipeline
         * @param msg 就是客户端发送的数据
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            //Channel channel = ctx.channel();
            //ChannelPipeline pipeline = ctx.pipeline(); //本质是一个双向链接, 出站入站
            //将 msg 转成一个 ByteBuf，类似NIO 的 ByteBuffer
            ByteBuf buf = (ByteBuf) msg;
            System.out.println("收到客户端的消息:" + buf.toString(CharsetUtil.UTF_8));
        }
    
        /**
         * 数据读取完毕处理方法
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) {
            System.out.println("数据读取完毕");
            ByteBuf buf = Unpooled.copiedBuffer("HelloClient".getBytes(CharsetUtil.UTF_8));
            ctx.writeAndFlush(buf);
        }
    
        /**
         * 处理异常, 一般是需要关闭通道
         *
         * @param ctx
         * @param cause
         * @throws Exception
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            System.out.println("出现异常");
            ctx.close();
        }
    }

## 客户端 ##

public class NettyClient {
        public static void main(String[] args) throws Exception {
            //客户端需要一个事件循环组
            EventLoopGroup group = new NioEventLoopGroup();
            try {
                //创建客户端启动对象
                Bootstrap bootstrap = new Bootstrap();
                //设置相关参数
                bootstrap.group(group) //设置线程组
                        .channel(NioSocketChannel.class) // 使用NioSocketChannel作为客户端的通道实现
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                //加入处理器
                                ch.pipeline().addLast(new NettyClientHandler());
                            }
                        });
    
                System.out.println("netty client start。。");
                //启动客户端去连接服务器端
                ChannelFuture cf = bootstrap.connect("127.0.0.1", 9000).sync();
                //对通道关闭进行监听
                cf.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully();
            }
        }
    }
    
    public class NettyClientHandler extends ChannelInboundHandlerAdapter {
        /**
         * 当客户端连接服务器完成就会触发该方法
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            System.out.println("已经连接客户端");
            ByteBuf buf = Unpooled.copiedBuffer("HelloServer".getBytes(CharsetUtil.UTF_8));
            ctx.writeAndFlush(buf);
        }
    
        //当通道有读取事件时会触发，即服务端发送数据给客户端
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            System.out.println("读取客户端数据");
            ByteBuf buf = (ByteBuf) msg;
            System.out.println("收到服务端的消息:" + buf.toString(CharsetUtil.UTF_8));
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            System.out.println("出现异常");
            cause.printStackTrace();
            ctx.close();
        }
    
    }

## 流程 ##

1.  启动服务端，绑定9000端口
2.  启动客户端执行bootstrap.connect()进行连接
3.  客户端接连接，客户端和服务端执行initChannel中Handler的channelActive方法，进行连接
4.  服务端执行ServerHandler的channelRead接受客户端发过来的信息
5.  服务端执行ServerHandler的channelReadComplete向客户端相应信息
6.  客户端执行ClientHandler的channelRead接受服务端发过来的信息