20-Netty TCP 粘包和拆包及解决方案

TCP粘包和拆包的基本介绍TCP是面向连接的, 面向流的, 提供可靠性服务, 收发两端(客户端和服务器端) 都有一一成对的Socket,因此发送端为了将多个发给接收端的包, 更有效的发给对方, 使用了优化算法(Nagle算法),将多次间隔较小且数据量小的数据,合并成一个大的数据块,然后进行封包, 这样做虽然提高了效率,但是接收端就难于分辨出完整的数据包了,因为面向流的通信是无消息保护边界的由于TCP无消息保护边界, 需要在接收端处理消息边界问题, 也就是我们所说的粘包,拆包问题,看一张图示意图TCP粘包,拆包图解

对图的说明

假设客户端分别发送了两个数据包D1和D2给服务端, 由于服务端一次读取到字节数是不确定的,故有可能存在以下四种情况

服务端分别两次读取到了两个独立的数据包, 分别是D1 和 D2, 没有粘包和拆包服务端一次接收到了两个数据包D1和D2粘在了一起,称之为TCP粘包服务端分两次读取到了数据包, 第一次读取到了完整的D1包和D2包的部分内容, 第二次读取到了D2包的剩余部分, 称之为TCP拆包服务器分两次读取到了数据包, 第一次读取到了D1包的部分内容D1_1, 第二次读取到了D1包的剩余部分D1_2, 和完整的D2包TCP粘包和拆包现象实例

在编写Netty程序时, 如果没有做处理,就会发生粘包和拆包问题

看一个具体的实例

NettyServer

代码语言：javascript代码运行次数：0运行复制

package com.dance.netty.netty.tcp;import io.netty.bootstrap.ServerBootstrap;import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.*;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioServerSocketChannel;import java.nio.charset.StandardCharsets;import java.util.UUID;public class NettyServer {    public static void main(String[] args) {        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);        NioEventLoopGroup workerGroup = new NioEventLoopGroup();        try {            ServerBootstrap serverBootstrap = new ServerBootstrap();            serverBootstrap.group(bossGroup,workerGroup)                    .channel(NioServerSocketChannel.class)                    .option(ChannelOption.SO_BACKLOG, 128)                    .childOption(ChannelOption.SO_KEEPALIVE, true)                    .childHandler(new ChannelInitializer() {                        @Override                        protected void initChannel(SocketChannel ch) throws Exception {                            ChannelPipeline pipeline = ch.pipeline();                            pipeline.addLast(new NettyServerHandler());                        }                    });            ChannelFuture sync = serverBootstrap.bind("127.0.0.1", 7000).sync();            System.out.println("server is ready ......");            sync.channel().closeFuture().sync();        } catch (Exception e) {            e.printStackTrace();        } finally {            bossGroup.shutdownGracefully();            workerGroup.shutdownGracefully();        }    }    static class NettyServerHandler extends SimpleChannelInboundHandler {        private int count = 0;        @Override        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {            byte[] bytes = new byte[msg.readableBytes()];            msg.readBytes(bytes);            count++;            System.out.println("服务器第"+count+"次接收到来自客户端的数据:" + new String(bytes, StandardCharsets.UTF_8));            // 服务器回送数据给客户端 回送随机的UUID给客户端            ctx.writeAndFlush(Unpooled.copiedBuffer(UUID.randomUUID().toString(),StandardCharsets.UTF_8));        }        @Override        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {            ctx.close();            cause.printStackTrace();        }    }}

NettyClient

代码语言：javascript代码运行次数：0运行复制

package com.dance.netty.netty.tcp;import io.netty.bootstrap.Bootstrap;import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.*;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioSocketChannel;import java.nio.charset.StandardCharsets;public class NettyClient {    public static void main(String[] args) {        NioEventLoopGroup eventExecutors = new NioEventLoopGroup();        try {            Bootstrap bootstrap = new Bootstrap();            bootstrap.group(eventExecutors)                    .channel(NioSocketChannel.class)                    .handler(new ChannelInitializer() {                        @Override                        protected void initChannel(SocketChannel ch) throws Exception {                            ChannelPipeline pipeline = ch.pipeline();                            pipeline.addLast(new NettyClientHandler());                        }                    });            ChannelFuture sync = bootstrap.connect("127.0.0.1", 7000).sync();            sync.channel().closeFuture().sync();        } catch (Exception e) {            e.printStackTrace();        } finally {            eventExecutors.shutdownGracefully();        }    }    static class NettyClientHandler extends SimpleChannelInboundHandler {        private int count = 0;        @Override        public void channelActive(ChannelHandlerContext ctx) throws Exception {            // 连续发送10条数据            for (int i = 0; i < 10; i++) {                ctx.writeAndFlush(Unpooled.copiedBuffer("hello,server!" + i, StandardCharsets.UTF_8));            }        }        @Override        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {            byte[] bytes = new byte[msg.readableBytes()];            msg.readBytes(bytes);            // 接收服务器的返回            count++;            System.out.println("客户端第"+count+"次接收服务端的回送:" + new String(bytes, StandardCharsets.UTF_8));        }        @Override        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {            ctx.close();            cause.printStackTrace();        }    }}

执行结果

Server

代码语言：javascript代码运行次数：0运行复制

log4j:WARN No appenders could be found for logger (io.netty.util.internal.logging.InternalLoggerFactory).log4j:WARN Please initialize the log4j system properly.log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.server is ready ......服务器第1次接收到来自客户端的数据:hello,server!0hello,server!1hello,server!2hello,server!3hello,server!4hello,server!5hello,server!6hello,server!7hello,server!8hello,server!9服务器第1次接收到来自客户端的数据:hello,server!0服务器第2次接收到来自客户端的数据:hello,server!1服务器第3次接收到来自客户端的数据:hello,server!2hello,server!3hello,server!4服务器第4次接收到来自客户端的数据:hello,server!5hello,server!6服务器第5次接收到来自客户端的数据:hello,server!7hello,server!8hello,server!9

Client1

代码语言：javascript代码运行次数：0运行复制

客户端第1次接收服务端的回送:84653e99-0e7f-431d-a897-c215af959a3b

Client2

代码语言：javascript代码运行次数：0运行复制

客户端第1次接收服务端的回送:6f3b0e79-2f40-4066-bb6b-80f988ecec116b6bbd94-b345-46d6-8d36-a114534331a850628e04-ece1-4f58-b684-d30189f6cf26b2139027-6bda-4d40-9238-9fc0e59bc7a64b568ffe-f616-4f48-8f1c-05ecf3e817ee

分析:

服务器启动后到server is ready ……

第一个客户端启动后 TCP将10次发送直接封包成一次直接发送,所以导致了服务器一次就收到了所有的数据,产生了TCP粘包,拆包的问题

第二客户端启动后 TCP将10次发送分别封装成了5次请求,产生粘包,拆包问题

TCP粘包和拆包解决方案使用自定义协议 + 编解码器来解决关键就是要解决 服务器每次读取数据长度的问题, 这个问题解决, 就不会出现服务器多读或少读数据的问题,从而避免TCP粘包和拆包TCP粘包, 拆包解决方案实现要求客户端发送5个Message对象, 客户端每次发送一个Message对象服务器端每次接收一个Message, 分5次进行解码, 每读到一个Message, 会回复一个Message对象给客户端

新建协议MessageProtocol代码语言：javascript代码运行次数：0运行复制

package com.dance.netty.netty.protocoltcp;/** * 消息协议 */public class MessageProtocol {    private int length;    private byte[] content;    public MessageProtocol() {    }    public MessageProtocol(int length, byte[] content) {        this.length = length;        this.content = content;    }    public int getLength() {        return length;    }    public void setLength(int length) {        this.length = length;    }    public byte[] getContent() {        return content;    }    public void setContent(byte[] content) {        this.content = content;    }}

新建编码器代码语言：javascript代码运行次数：0运行复制

package com.dance.netty.netty.protocoltcp;import io.netty.buffer.ByteBuf;import io.netty.channel.ChannelHandlerContext;import io.netty.handler.codec.MessageToByteEncoder;/** * 自定义协议编码器 */public class MyMessageProtocolEncoder extends MessageToByteEncoder {    @Override    protected void encode(ChannelHandlerContext ctx, MessageProtocol msg, ByteBuf out) throws Exception {//        System.out.println("自定义协议-->开始编码");        // 开始发送数据        out.writeInt(msg.getLength()); // 优先发送长度,定义边界        out.writeBytes(msg.getContent());//        System.out.println("自定义协议---->编码完成");    }}

新建解码器代码语言：javascript代码运行次数：0运行复制

package com.dance.netty.netty.protocoltcp;import io.netty.buffer.ByteBuf;import io.netty.channel.ChannelHandlerContext;import io.netty.handler.codec.ByteToMessageDecoder;import java.util.List;public class MyMessageProtocolDecoder extends ByteToMessageDecoder {    @Override    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List