[网络协议]Netty专栏——粘包与拆包

粘包与拆包

一、基本概念

1、TCP

所谓 TCP，就是Transport Control Protocol—传输控制协议，是 IP 协议族的重要组成，面向连接、提供可靠的传输服务，采用 Socket 技术。为了提稿传输效率，充分利用带宽，TCP 使用 Nagle 算法将多个较小的数据封装成一个较大的 TCP包，但是问题来了，TCP 缺乏消息边界机制，导致客户端难于将各个数据段进行正确拆分，这也是面向流的通信方式的局限。

2、TCP 粘包拆包示意图

• 1）服务端读取到两个独立的TCP包，不存在粘包和拆包问题
• 2）服务端读到了由两个数据段组合成的一个 TCP包，这种情况称之为粘包，因为两个本来独立的、互不相干的数据段被封装成了一个TCP包（形象地理解，就是粘在一起了）
• 3）服务端读取两个数据段时，如果发生了先读到了一个完整地数据段和另一个数据段地部分数据，或者先度到了一个数据段地部分数据然后再读到同个数据段剩下地数据和另一个完整的数据段，都称之为出现了 TCP 拆包。

二、TCP 粘包拆包示例

1、服务端

public class TcpServer {
    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try{
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup,workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new TcpServerInitializer());
            ChannelFuture channelFuture = serverBootstrap.bind(8001).sync();
            System.out.println("Service is ready!!!!!!!");
            channelFuture.channel().closeFuture().sync();
        } finally{
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }
}
public class TcpServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline().addLast(new TcpServerHandler());
    }
}
public class TcpServerHandler extends SimpleChannelInboundHandler<ByteBuf> {
    private int count;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        byte[] buffer = new byte[msg.readableBytes()];
        msg.readBytes(buffer);
        String str = new String(buffer, CharsetUtil.UTF_8);
        System.out.println("Received from client: " + str);
        System.out.println("The num of msg of server received is " + (++this.count));
        ByteBuf buf = Unpooled.copiedBuffer(UUID.randomUUID().toString()+"\n", CharsetUtil.UTF_8);
        ctx.writeAndFlush(buf);
    }

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

2、客户端

public class TcpClient {
    public static void main(String[] args) throws Exception {

        EventLoopGroup group = new NioEventLoopGroup();

        try{
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new TcpClientInitializer());

            ChannelFuture future = bootstrap.connect("localhost", 8001).sync();
            System.out.println("Client is ready~~~~~");
            future.channel().closeFuture().sync();

        } finally{
            group.shutdownGracefully();
        }

    }
}
public class TcpClientInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline().addLast(new TcpClientHandler());
    }
}
public class TcpClientHandler extends SimpleChannelInboundHandler<ByteBuf> {

    private int count;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        int len = 10;
        for (int i=0; i<len; i++){
            ByteBuf buf = Unpooled.copiedBuffer("Hello Server, time=" + i, CharsetUtil.UTF_8);
            ctx.writeAndFlush(buf);
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        byte[] buffer = new byte[msg.readableBytes()];
        msg.readBytes(buffer);
        String message = new String(buffer, CharsetUtil.UTF_8);
        System.out.println("The response of Server is " + message);
        System.out.println("The num of received msg from server is " + (++this.count));
    }

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

3、测试

Received from client: Hello Server, time=0.

The num of msg of server received is 1
Received from client: Hello Server, time=1.

The num of msg of server received is 2
Received from client: Hello Server, time=2.
Hello Server, time=3.
Hello Server, time=4.
Hello Server, time=5.

The num of msg of server received is 3
Received from client: Hello Server, time=6.
Hello Server, time=7.

The num of msg of server received is 4
Received from client: Hello Server, time=8.
Hello Server, time=9.

The num of msg of server received is 5


==========================================
The response of Server is 2ba766d8-442d-4c5d-bdc1-86d1528b663b
860708e3-5ad3-4fb3-bfce-2227e89567b7
11e789f9-fd8a-4f49-a918-3dcc7970d1a1
971f4f8c-d10e-44c4-a3e0-9c9a2ebcf96b
85a9011f-3804-44fd-a941-3acecf48a7e0

三、解决 TCP 粘包拆包

1、案例要求

• 1）使用自定义协议和编解码器解决
• 2）关键在于解决服务端每次读取数据的长度的问题
• 3）客户端发送5个message对象，每次发送一个
• 4）服务端每次接收一个message对象，分5次进行解码，每次读取到一个 Message，会回复一个 Message 给客户端。

2、自定义协议

@Data
public class MessageProtocol {
		private int len;
		private byte[] content;
}

3、编解码器

public class MessageEncoder extends MessageToByteEncoder<MessageProtocol> {
    @Override
    protected void encode(ChannelHandlerContext ctx, MessageProtocol msg, ByteBuf out)
        throws Exception {
        System.out.println("Starting encode message");
        out.writeInt(msg.getLen());
        out.writeBytes(msg.getContent());
    }
}
public class MessageDecoder extends ReplayingDecoder<Void> {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        System.out.println("Starting Decode message");
        int len = in.readInt();
        byte[] content = new byte[len];
        in.readBytes(content);
        MessageProtocol msg = new MessageProtocol();
        msg.setLen(len);
        msg.setContent(content);
        out.add(msg);
    }
}

然后将这两个编解码器都加入到Server和Client各自的 Initializer 类中

4、改进 handler

• 客户端的 Handler修改成如下：

public class TcpClientHandler extends SimpleChannelInboundHandler<MessageProtocol> {

    private int count;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        int len = 10;
        for (int i=0; i<len; i++){
            String msg = "Hello, Server, i am testing to use customize transport protocol to send message";
            byte[] content = msg.getBytes(StandardCharsets.UTF_8);
            int length = msg.getBytes(StandardCharsets.UTF_8).length;
            MessageProtocol message = new MessageProtocol();
            message.setLen(length);
            message.setContent(content);
            ctx.writeAndFlush(message);

        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {
        int len = msg.getLen();
        byte[] content = msg.getContent();
        System.out.println("Client received message: ");
        System.out.println("Length of message = " + len);
        System.out.println("Content of message = " + new String(content, CharsetUtil.UTF_8));
        System.out.println("Time = " + (++this.count));
    }

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

• 服务端也要作相应改变：

public class TcpServerHandler extends SimpleChannelInboundHandler<MessageProtocol> {
    private int count;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {
        int len = msg.getLen();
        byte[] content = msg.getContent();
        System.out.println("Server received message: ");
        System.out.println("Length of message = " + len);
        System.out.println("Content of message = " + new String(content, CharsetUtil.UTF_8));
        System.out.println("Time = " + (++this.count));
        MessageProtocol response = new MessageProtocol();
        byte[] responseMsg = "200, OK".getBytes(StandardCharsets.UTF_8);
        response.setLen(responseMsg.length);
        response.setContent(responseMsg);
        ctx.writeAndFlush(response);
    }

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

5、测试

测试效果如下：

======== client ======
Client is ready~~~~~
Starting encode message
Starting encode message
Starting encode message
Starting encode message
Starting encode message
Starting encode message
Starting encode message
Starting encode message
Starting encode message
Starting encode message
Starting Decode message
Client received message: 
Length of message = 7
Content of message = 200, OK
Time = 1
Starting Decode message
Client received message: 
Length of message = 7
Content of message = 200, OK
Time = 2
Starting Decode message
Client received message: 
Length of message = 7
Content of message = 200, OK
Time = 3
Starting Decode message
Client received message: 
Length of message = 7
Content of message = 200, OK
Time = 4
Starting Decode message
Client received message: 
Length of message = 7
Content of message = 200, OK
Time = 5
Starting Decode message
Client received message: 
Length of message = 7
Content of message = 200, OK
Time = 6
Starting Decode message
Client received message: 
Length of message = 7
Content of message = 200, OK
Time = 7
Starting Decode message
Client received message: 
Length of message = 7
Content of message = 200, OK
Time = 8
Starting Decode message
Client received message: 
Length of message = 7
Content of message = 200, OK
Time = 9
Starting Decode message
Client received message: 
Length of message = 7
Content of message = 200, OK
Time = 10

======= server =======
Service is ready!!!!!!!
Starting Decode message
Server received message: 
Length of message = 79
Content of message = Hello, Server, i am testing to use customize transport protocol to send message
Time = 1
Starting encode message
Starting Decode message
Server received message: 
Length of message = 79
Content of message = Hello, Server, i am testing to use customize transport protocol to send message
Time = 2
Starting encode message
Starting Decode message
Server received message: 
Length of message = 79
Content of message = Hello, Server, i am testing to use customize transport protocol to send message
Time = 3
Starting encode message
Starting Decode message
Server received message: 
Length of message = 79
Content of message = Hello, Server, i am testing to use customize transport protocol to send message
Time = 4
Starting encode message
Starting Decode message
Server received message: 
Length of message = 79
Content of message = Hello, Server, i am testing to use customize transport protocol to send message
Time = 5
Starting encode message
Starting Decode message
Server received message: 
Length of message = 79
Content of message = Hello, Server, i am testing to use customize transport protocol to send message
Time = 6
Starting encode message
Starting Decode message
Server received message: 
Length of message = 79
Content of message = Hello, Server, i am testing to use customize transport protocol to send message
Time = 7
Starting encode message
Starting Decode message
Server received message: 
Length of message = 79
Content of message = Hello, Server, i am testing to use customize transport protocol to send message
Time = 8
Starting encode message
Starting Decode message
Server received message: 
Length of message = 79
Content of message = Hello, Server, i am testing to use customize transport protocol to send message
Time = 9
Starting encode message
Starting Decode message
Server received message: 
Length of message = 79
Content of message = Hello, Server, i am testing to use customize transport protocol to send message
Time = 10
Starting encode message