netty学习笔记（二）针对基础实践

我就是我 2023-05-21 13:58 55阅读 0赞

### 文章目录 ###

*  前言
 *  coder
 *   *  ToLongDecoder
 *  心跳
 *   *  总结
 *  下一篇

# 前言 #

在上一篇介绍了netty基础概念，这一篇主要针对这些概念进行实践，加深印象

*  [3种IO区别以及netty基础学习笔记][3_IO_netty]

# coder #

测试下coder，在channelpipeline链上一开始进来就是需要对byte进行解码，在输出的时候进去编码。

**解码器**

在这里插入代码片public class MyDecoder extends ByteToMessageDecoder {
    
        private int length;
    
        public MyDecoder(int length) {
            this.length = length;
        }
    
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
            int bytes = in.readableBytes();
            if (bytes >= length) {
                int readLength = length;
                while (bytes >= 0) {
                    if (bytes < length) {
                        readLength = bytes;
                    }
                    ByteBuf byteBuf = in.readBytes(readLength);
                    out.add(byteBuf);
                    bytes -= length;
                }
            }
        }
    
    }

这个解码器主要是为了将传入的数据切成固定的长度，其余的放下一个。

**Test**

@Test
        public void encoder() {
            ByteBuf byteBuf = Unpooled.buffer();
            for (int i = 0; i < 10; i++) {
                byteBuf.writeByte(i);
            }
            //read会索引移动
            /*System.out.println(byteBuf.readBytes(5));
            System.out.println(byteBuf.readerIndex());
            System.out.println(byteBuf.readBytes(5));
            System.out.println(byteBuf.readerIndex());*/
    
            EmbeddedChannel embeddedChannel = new EmbeddedChannel(new MyDecoder(3));
            //System.out.println(byteBuf.readBytes(10));
           
            System.out.println(embeddedChannel.writeInbound(byteBuf.readBytes(5)));
            ByteBuf byteBuf2 = embeddedChannel.readInbound();
            System.out.println("可读字数：" + byteBuf2.readableBytes());
            while (byteBuf2.isReadable()) {
                System.out.println(byteBuf2.readByte());
            }
            byteBuf2.release();
            ByteBuf byteBuf3 = embeddedChannel.readInbound();
            System.out.println("可读字数：" + byteBuf3.readableBytes());
            while (byteBuf3.isReadable()) {
                System.out.println(byteBuf3.readByte());
            }
            byteBuf3.release();
        }

**输出结果**

true
    可读字数：2
    3
    4
    可读字数：3
    5
    6
    7

从上面注释掉的代码可以看到，read这个ByteBuf的话索引是会移动的。

## ToLongDecoder ##

当输入的长度太长的时候会抛出异常。

@Test
        public void tooLong() {
            ByteBuf byteBuf = Unpooled.buffer();
            for (int i = 0; i < 10; i++) {
                byteBuf.writeByte(i);
            }
            EmbeddedChannel embeddedChannel = new EmbeddedChannel(new ToLongDecoder(8));
            try {
                System.out.println(embeddedChannel.writeInbound(byteBuf.readBytes(10)));
            } catch (TooLongFrameException e) {
                System.out.println("太长...");
            }
            ByteBuf byteBuf1 = embeddedChannel.readInbound();
            System.out.println("可读字数：" + byteBuf1.readableBytes());
            while (byteBuf1.isReadable()) {
                System.out.println(byteBuf1.readByte());
            }
            byteBuf1.release();
        }

# 心跳 #

这个我基本是试了一天，差点没有吐血，哈哈哈。（可能我还是比较菜吧）

EmbeddedChannel这个类是不能用来测试心跳的，一直报UnSupportException，最后只能实现一个服务端跟客户端了。

**服务端**

public static void main(String[] args) throws InterruptedException {
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workGroup);
                b.channel(NioServerSocketChannel.class);
    //            b.childHandler(idleStateHandler);
    //            b.childHandler(new MyIdleHandler());
                b.childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast("idle",new IdleStateHandler(
                                6L,0L,0L, TimeUnit.SECONDS))
                                .addLast("MyIdleHandler",new MyIdleHandler());
                    }
                });
                System.out.println("服务端开启等待客户端连接....");
                Channel ch = b.bind(8888).sync().channel();
                ch.closeFuture().sync();
            } catch (Exception e) {
                e.printStackTrace();
            }finally{
                //优雅的退出程序
                bossGroup.shutdownGracefully().sync();
                workGroup.shutdownGracefully().sync();
            }
        }

注意一下细节，b.childHandler(idleStateHandler);我使用之后一直不行，后面统一使用ChannelInitializer来添加ChannelPipeline.

**针对超时引发IdleStateEvent做处理，如果超过太多次就直接断开连接。**

@ChannelHandler.Sharable
    public class MyIdleHandler extends ChannelInboundHandlerAdapter {
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("had connection");
            super.channelActive(ctx);
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            System.out.println("get data:"+msg);
            super.channelRead(ctx, msg);
        }
    
        private int lossConnectCount = 0;
    
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            lossConnectCount++;
            if(lossConnectCount==3){
                ctx.channel().close();
            }
            if (evt instanceof IdleStateEvent) {
                System.out.println("too long close connection");
                ctx.writeAndFlush(Unpooled.copiedBuffer("closing", CharsetUtil.UTF_8)).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            }
            super.userEventTriggered(ctx, evt);
        }
    
    }

**addListener(ChannelFutureListener.CLOSE\_ON\_FAILURE);**

目前是如果发送不成功，那么客户端绝壁挂了，直接关闭channel~

**new IdleStateHandler(6L,0L,0L, TimeUnit.SECONDS)**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70]  
服务端使用read超时，客户端使用write超时

**客户端**

EventLoopGroup group = new NioEventLoopGroup();
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
    
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new SimpleChannelInboundHandler<ByteBuf>() {
    
                                @Override
                                public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                    System.out.println(msg);
                                    if (msg instanceof ByteBuf) {
                                        byte[] bytes = new byte[((ByteBuf) msg).readableBytes()];
                                        ((ByteBuf) msg).readBytes(bytes);
                                        System.out.println(new String(bytes));
                                    }
                                    super.channelRead(ctx, msg);
                                }
    
                                @Override
                                protected void messageReceived(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
    
                                }
    
                            });
                        }
                    });
            Channel channel = bootstrap.connect("127.0.0.1", 8888).channel();
            int i = 0;
            while (i < 3) {
                i++;
                channel.writeAndFlush(Unpooled.copiedBuffer(new Date() + ": hello world!", CharsetUtil.UTF_8)).addListener(future -> {
                    if (future.isSuccess()) {
                        System.out.println("send...");
                    } else {
                        System.out.println("send fail...");
                        future.cause().printStackTrace();
                    }
                });
                Thread.sleep(2000);
            }

**输出结果**

1.服务端

14:54:37.753 [nioEventLoopGroup-3-3] INFO com.example.demo.test.MyIdleHandler - had connection
    14:54:37.763 [nioEventLoopGroup-3-3] INFO com.example.demo.test.MyIdleHandler - get data:UnpooledUnsafeDirectByteBuf(ridx: 0, widx: 42, cap: 1024)
    14:54:37.763 [nioEventLoopGroup-3-3] DEBUG io.netty.channel.DefaultChannelPipeline - Discarded inbound message UnpooledUnsafeDirectByteBuf(ridx: 0, widx: 42, cap: 1024) that reached at the tail of the pipeline. Please check your pipeline configuration.
    14:54:39.763 [nioEventLoopGroup-3-3] INFO com.example.demo.test.MyIdleHandler - get data:UnpooledUnsafeDirectByteBuf(ridx: 0, widx: 42, cap: 1024)
    14:54:39.763 [nioEventLoopGroup-3-3] DEBUG io.netty.channel.DefaultChannelPipeline - Discarded inbound message UnpooledUnsafeDirectByteBuf(ridx: 0, widx: 42, cap: 1024) that reached at the tail of the pipeline. Please check your pipeline configuration.
    14:54:41.763 [nioEventLoopGroup-3-3] INFO com.example.demo.test.MyIdleHandler - get data:UnpooledUnsafeDirectByteBuf(ridx: 0, widx: 42, cap: 512)
    14:54:41.764 [nioEventLoopGroup-3-3] DEBUG io.netty.channel.DefaultChannelPipeline - Discarded inbound message UnpooledUnsafeDirectByteBuf(ridx: 0, widx: 42, cap: 512) that reached at the tail of the pipeline. Please check your pipeline configuration.
    14:54:47.763 [nioEventLoopGroup-3-3] INFO com.example.demo.test.MyIdleHandler - too long close connection
    14:54:53.764 [nioEventLoopGroup-3-3] INFO com.example.demo.test.MyIdleHandler - too long close connection
    14:54:59.765 [nioEventLoopGroup-3-3] INFO com.example.demo.test.MyIdleHandler - too long close connection

每6秒心跳问下客户端，你还活着吗？可以看到14:54:41.763的时候收到客户端的发送，在14:54:47.763的时候没有回复了就开始心跳，咚咚咚~

2.客户端

send...
    send...
    send...
    UnpooledUnsafeDirectByteBuf(ridx: 0, widx: 7, cap: 1024)
    closing
    UnpooledUnsafeDirectByteBuf(ridx: 0, widx: 7, cap: 1024)
    closing

一开始发送消息，证明自己是活着的。后面没有发送消息之后，收到了来自服务端closing这个心跳。

## 总结 ##

上面看到服务端是正常的超时进行心跳，同样客户端也能配置超时，就是写超时，当超时的时候给服务端心跳访问，当然这个时间可以短些，这样服务端写就不会超时了。

*  [Netty学习（五）—IdleStateHandler心跳机制][Netty_IdleStateHandler]

当超时channel关闭时，可以试下重连~

# 下一篇 #

下一篇讲讲解我大学做的一个小项目（基于netty图文直播），当然是抄抄网络大佬们的代码，因为我现在回头看这些代码都有些看不太懂咯~

[3_IO_netty]: https://blog.csdn.net/weixin_38336658/article/details/105585912
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODMzNjY1OA_size_16_color_FFFFFF_t_70]: /images/20230521/7c85cf9197c04798b41da43594a91936.png
[Netty_IdleStateHandler]: https://blog.csdn.net/u013967175/article/details/78591810