Netty学习之五—— 粘包/拆包之分隔符解码器DelimiterBasedFrameDecoder的应用

r囧r小猫 2022-02-21 10:38 321阅读 0赞

# DelimiterBasedFrameDecoder 应用开发 #

通过对 DelimiterBasedFrameDecoder 的使用，可以自动完成以分隔符作为码流结束标识的消息的解码。

演示程序以经典的 Echo 服务为例。EchoServer 接收到 EchoClient 的请求消息后，将其打印出来，然后将原始消息返回给客户端，消息以“$\_” 作为分隔符。

服务端代码：

package com.netty.echo;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    
    public class EchoServer {
        public void bind(int port) throws Exception {
            // 配置服务端的 NIO 线程组
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 100)
                        .handler(new LoggingHandler(LogLevel.INFO))
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
                                /**
                                 * 第一个参数：1024 表示单条记录的最大长度，当达到最大长度仍然没有找到
                                 *              分隔符，就抛出 TooLongFrameException 异常，防止异常码流缺失
                                 *              分隔符导致内存溢出，这是Netty解码器的可靠性保证。
                                 * 第二个参数：是分隔符
                                 */
                                ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
                                ch.pipeline().addLast(new StringDecoder());
                                ch.pipeline().addLast(new EchoServerHandler());
                            }
                        });
                // 绑定端口，同步等待成功
                ChannelFuture f = b.bind(port).sync();
    
                // 等待服务端监听端口关闭
                f.channel().closeFuture().sync();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
    
        public static void main(String[] args) throws Exception {
            int port = 8080;
            new EchoServer().bind(port);
        }
    }

package com.netty.echo;
    
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    import java.util.Date;
    
    public class EchoServerHandler extends ChannelHandlerAdapter {
        private int counter;
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            String body = (String) msg;
            System.out.println("This is " + ++counter + " times receive client : [" + body + "]");
            body += "$_";
            ByteBuf resp = Unpooled.copiedBuffer(body.getBytes());
            ctx.writeAndFlush(resp);
        }
    }

客户端代码：

package com.netty.echo;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    
    public class EchoClient {
        public void connect(int port, String host) throws Exception {
            EventLoopGroup group = new NioEventLoopGroup();
            try{
                Bootstrap b = new Bootstrap();
                b.group(group).channel(NioSocketChannel.class)
                        .option(ChannelOption.TCP_NODELAY, true)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
                                /**
                                 * 第一个参数：1024 表示单条记录的最大长度，当达到最大长度仍然没有找到
                                 *              分隔符，就抛出 TooLongFrameException 异常，防止异常码流缺失
                                 *              分隔符导致内存溢出，这是Netty解码器的可靠性保证。
                                 * 第二个参数：是分隔符
                                 */
                                ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
                                ch.pipeline().addLast(new StringDecoder());
                                ch.pipeline().addLast(new EchoClientHandler());
                            }
                        });
                // 发起异步连接操作
                ChannelFuture f = b.connect(host, port).sync();
    
                // 等待客户端链路关闭
                f.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws Exception {
            int port = 8080;
            new EchoClient().connect(port, "127.0.0.1");
        }
    }

package com.netty.echo;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    public class EchoClientHandler extends ChannelHandlerAdapter {
    
        private int counter;
        static final String ECHO_REQ = "Hi , Netty Server .$_";
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            for (int i = 0; i < 10; i++) {
                ctx.writeAndFlush(Unpooled.copiedBuffer(ECHO_REQ.getBytes()));
            }
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            System.out.println("This is " + ++counter + " times receive server : [" + msg + "]");
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }
    }

运行结果：

服务端：![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz_size_16_color_FFFFFF_t_70][]客户端：![20190411093416405.png][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20190411093401982.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1YW5oYW8xMjAz,size_16,color_FFFFFF,t_70
[20190411093416405.png]: /images/20220221/d794c6b66f824b918c8c9bbdea1b18ae.png