使用Netty提供的解码器解决TCP粘包和拆包问题

本是古典何须时尚 2022-08-28 12:40 212阅读 0赞

解决粘包和拆包问题：

发送端需要使用tcp\_nodelay来关闭nagle算法，有数据就发送；

接收端，需要在应用层来解决问题，可用的方式：

固定消息长度，比如不足的位数补空格,接收端使用FixedLengthFrameDecoder；

使用换行符、自定义分隔符给消息添加边界；

将消息拆分为消息头和消息体，消息头中指定了消息长度等信息,接收端使用LengthFieldBaseFrameDecoder；

# 一 使用换行符给消息添加边界 #

可以使用Netty提供的解码器LineBasedFrameDecoder,这个解码器是以换行符为结束标志进行解码，在发送消息时，我们给消息结束追加一个换行符，当接收端收到后按换行符进行解码。

package com.tech.netty.ed;
    
    import com.tech.netty.echo.InboundleHandlerA;
    import com.tech.netty.echo.InboundleHandlerB;
    import com.tech.netty.echo.OutboundleHandlerA;
    import com.tech.netty.echo.OutboundleHandlerB;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.LineBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    
    /**
     * @author lw
     * @since 2021/9/17
     */
    public class EchoServer {
    
        private int port;
    
        public EchoServer(int port) {
            this.port = port;
        }
    
        /**
         * 启动流程
         * @throws InterruptedException
         */
        public void run() throws InterruptedException {
            //配置服务端线程组
            NioEventLoopGroup bossGroup = new NioEventLoopGroup();
            NioEventLoopGroup workerGroup = new NioEventLoopGroup();
            try{
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(bossGroup,workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
                                ch.pipeline().addLast(new StringDecoder());
                                ch.pipeline().addLast(new EchoServerHandler());
                            }
                        });
                System.out.println("Echo 服务器启动ing");
                //绑定端口 同步等待成功
                ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
                //等待服务端监听端口关闭
                channelFuture.channel().closeFuture().sync();
            }finally {
                //优雅退出释放线程池
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            int port=8080;
            if(args.length>0){
                port=Integer.parseInt(args[0]);
            }
            new EchoServer(port).run();
        }
    }

package com.tech.netty.ed;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.CharsetUtil;
    
    /**
     * @author lw
     * @since 2021/9/17
     */
    public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    
        private int count;
    
    //    @Override
    //    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    //        ByteBuf byteBuf = (ByteBuf) msg;
    //        byte[] bytes=new byte[byteBuf.readableBytes()];
    //        byteBuf.readBytes(bytes);
    //        String body=new String(bytes,"UTF-8").substring(0,bytes.length-System.getProperty("line.separator").length());
    //        System.out.println("服务端收到消息内容为："+body+"，收到消息次数："+ ++count);
    //    }
    
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            String body = (String) msg;
            System.out.println("服务端收到消息内容为：" + body + "，收到消息次数：" + ++count);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

package com.tech.netty.ed;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    
    /**
     * @author lw
     * @since 2021/9/17
     */
    public class EchoClient {
        private int port;
        private String host;
    
        public EchoClient(int port, String host) {
            this.port = port;
            this.host = host;
        }
    
        private void start() throws InterruptedException {
            NioEventLoopGroup group = new NioEventLoopGroup();
            try{
                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 EchoClientHandler());
                            }
                        });
                //连接到服务端，connect是异步连接，在调用同步等待sync，等待连接成功
                ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
                //阻塞直到客户端通道关闭
                channelFuture.channel().closeFuture().sync();
            }finally {
                //优雅退出，释放NIO线程组
                group.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            new EchoClient(8080,"127.0.0.1").start();;
        }
    }

package com.tech.netty.ed;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.util.CharsetUtil;
    
    /**
     * @author lw
     * @since 2021/9/17
     */
    public class EchoClientHandler extends ChannelInboundHandlerAdapter {
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ByteBuf mes=null;
            byte[] req = ("hello word"+System.getProperty("line.separator")).getBytes("UTF-8");
            //连续发送
            for(int i=0;i<100;i++){
                mes=Unpooled.buffer(req.length);
                mes.writeBytes(req);
                ctx.writeAndFlush(mes);
            }
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

正常收到消息：

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_20_color_FFFFFF_t_70_g_se_x_16][]

#  #

# 二 使用自定义分隔符给消息添加边界 #

可以使用Netty提供的解码器DelimiterBasedFrameDecoder，按分割符分割消息进行解码，发送端发送消息时在消息末尾追加分隔符，接收端收到后使用DelimiterBasedFrameDecoder按分割符解码消息。

package com.tech.netty.ed.delimiter;
    
    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.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.LineBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.util.CharsetUtil;
    
    /**
     * @author lw
     * @since 2021/9/17
     */
    public class EchoServer {
    
        private int port;
    
        public EchoServer(int port) {
            this.port = port;
        }
    
        /**
         * 启动流程
         * @throws InterruptedException
         */
        public void run() throws InterruptedException {
            //配置服务端线程组
            NioEventLoopGroup bossGroup = new NioEventLoopGroup();
            NioEventLoopGroup workerGroup = new NioEventLoopGroup();
            try{
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(bossGroup,workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ByteBuf delimiter= Unpooled.copiedBuffer("&_", CharsetUtil.UTF_8);
                                ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,delimiter));
                                ch.pipeline().addLast(new StringDecoder());
                                ch.pipeline().addLast(new EchoServerHandler());
                            }
                        });
                System.out.println("Echo 服务器启动ing");
                //绑定端口 同步等待成功
                ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
                //等待服务端监听端口关闭
                channelFuture.channel().closeFuture().sync();
            }finally {
                //优雅退出释放线程池
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            int port=8080;
            if(args.length>0){
                port=Integer.parseInt(args[0]);
            }
            new EchoServer(port).run();
        }
    }

package com.tech.netty.ed.delimiter;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    /**
     * @author lw
     * @since 2021/9/17
     */
    public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    
        private int count;
    
    //    @Override
    //    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    //        ByteBuf byteBuf = (ByteBuf) msg;
    //        byte[] bytes=new byte[byteBuf.readableBytes()];
    //        byteBuf.readBytes(bytes);
    //        String body=new String(bytes,"UTF-8").substring(0,bytes.length-System.getProperty("line.separator").length());
    //        System.out.println("服务端收到消息内容为："+body+"，收到消息次数："+ ++count);
    //    }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            String body = (String) msg;
            System.out.println("服务端收到消息内容为：" + body + "，收到消息次数：" + ++count);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    
    }

package com.tech.netty.ed.delimiter;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    
    /**
     * @author lw
     * @since 2021/9/17
     */
    public class EchoClient {
        private int port;
        private String host;
    
        public EchoClient(int port, String host) {
            this.port = port;
            this.host = host;
        }
    
        private void start() throws InterruptedException {
            NioEventLoopGroup group = new NioEventLoopGroup();
            try{
                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 EchoClientHandler());
                            }
                        });
                //连接到服务端，connect是异步连接，在调用同步等待sync，等待连接成功
                ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
                //阻塞直到客户端通道关闭
                channelFuture.channel().closeFuture().sync();
            }finally {
                //优雅退出，释放NIO线程组
                group.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            new EchoClient(8080,"127.0.0.1").start();;
        }
    }

package com.tech.netty.ed.delimiter;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    /**
     * @author lw
     * @since 2021/9/17
     */
    public class EchoClientHandler extends ChannelInboundHandlerAdapter {
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            String msg="Netty is a NIO client server framework which enables quick&_" +
                    "and easy development of network applications such as protocol servers and clients.&_" +
                    "It greatly simplifies and streamlines network programming such as TCP and UDP socket server.&_";
            byte[] bytes = msg.getBytes();
            ByteBuf byteBuf=Unpooled.buffer(bytes.length);
            byteBuf.writeBytes(bytes);
            ctx.writeAndFlush(byteBuf);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/90b041409b0943549fdd2d968a0a0c3d.png