netty初试

netty官网:点击进入

学习netty之实现一个丢弃服务器

环境：

•     JDK1.8
•     netty5.0+

步骤：

1.  实现一个丢弃服务器
2.  实现一个客户端发送数据

丢弃服务器的创建

     

//用于接受客户端的的连接，将连接注册到worker中
    EventLoopGroup boos = new NioEventLoopGroup();

    //处理客户端的连接
    EventLoopGroup worker = new NioEventLoopGroup();

    //服务端启动类
    ServerBootstrap serverBootstrap = new ServerBootstrap();

    ChannelFuture sync = null;
        try {
            ServerBootstrap boot = serverBootstrap.group(boos, worker)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                        //自己继承了ChannelHandlerAdapter
                            ch.pipeline().addLast(new DiscardHandler());
                        }
                    });
            //绑定端口，接受连接
            sync = boot.bind(10980).sync();
            System.out.println("丢弃服务器启动...");
            //等待服务器关闭
            sync.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally{
            if(null != sync){
                try {
                    //管理连接
                    sync.channel().closeFuture().sync();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //处理完客户端的请求后在优雅的关闭
            boos.shutdownGracefully();
            worker.shutdownGracefully();
        }　

DiscardHandler实现：

    public class DiscardHandler extends ChannelHandlerAdapter {
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("客户端断开连接");
        ctx.writeAndFlush(Unpooled.copiedBuffer("有一个客户端断开连接".getBytes("UTF-8")));
        cause.printStackTrace();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        服务器端本地查看发送来的信息
        ByteBuf msg1 = (ByteBuf) msg;
        byte[] buf = new byte[msg1.readableBytes()];
        msg1.readBytes(buf);
        String s = new String(buf, "UTF-8");
        System.out.println("客户端发来的信息："+s);
        //直接丢弃信息
        ReferenceCountUtil.release(msg);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.copiedBuffer("---连接成功：",CharsetUtil.UTF_8));
    }
  }

客户端发送数据：

           NioEventLoopGroup clientGroup = null;
        try {
            clientGroup = new NioEventLoopGroup();
            Bootstrap bootstrap = new Bootstrap();
            ChannelFuture connect = bootstrap.group(clientGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                        //自定义客户端的发送逻辑
                            ch.pipeline().addLast(new ClientHandler());
                        }
                    })
                    .connect("localhost", 10980);
            while (true) {
                Scanner scanner = new Scanner(System.in);
                System.out.println("输入 exit 结束：");
                String s = scanner.nextLine();

                if (!Objects.isNull(s) && s.equals("exit")) {
                    connect.channel().writeAndFlush(Unpooled.copiedBuffer(s.getBytes("UTF-8")))
                   //关闭监听器，代表ChannelFuture执行返回后，关闭连接。
                            .addListener(ChannelFutureListener.CLOSE);
                    break;
                }
                connect.channel().writeAndFlush(Unpooled
                                        .copiedBuffer(s.getBytes("UTF-8")));
            }
        }catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        } finally {
            clientGroup.shutdownGracefully();
        }
    }

ClientHandler实现：

    public class ClientHandler extends ChannelHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf msg1 = (ByteBuf) msg;
        byte[] buf = new byte[msg1.readableBytes()];
        msg1.readBytes(buf);
        System.out.println("服务器返回的信息："+new String(buf,"UTF-8"));
        //释放内存
        ReferenceCountUtil.release(msg);


    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
        System.out.println(cause.getMessage());
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.copiedBuffer("HI 首次连接".getBytes("UTF-8")));
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        System.out.println("数据发送成功");
    }

    }

问题：

    开始是继承了SimpleChannleINBoundAdpater<ByteBuf>
    在channelReadComplete中 发送 ctx.writeAndFlush()中发送服务器，会造成死循环，客户端一直发送信息给服务器，造成资源浪费
    其次，在发发送数据的时候直接发送ByteBuf对象，客户端是无法接受到的需要转换成
    Unpooled.copiedBuffer(byte[]);    发送数据

PS：

    可以实现在线用户数量梳理：    
    在服务端初始化一个AtomicInteger类，每次客户端连接的时候将此数值原子性+1，
    在客户端断开的时候将其-1.每个用户连接后，将其信息发送出去。