Netty-TCP粘包拆包问题（二）

发表于 | 分类于

概念

无论是服务端还是客户端，当我们读取或者发送数据的时候，都需要考虑TCP底层的粘包/拆包机制。TCP是一个“流”协议，所谓流就是没有界限的遗传数据。大家可以想象下如果河流里的水就好比是数据，他们是连成一片的，没有界线，TCP底层并不了解上层的业务数据具体的含义，他会根据TCP缓冲区的实际情况进行包的划分，也就是说，在业务上，我们一个完整的包可能会被TCP分成多个包进行发送，也可能多个包封装成一个大的数据包发送出去，这就是所谓的TCP粘包、拆包问题。分析TCP粘包、拆包问题的产生原因：

  1. 应用程序write写入的字节大小大于套接字发送缓冲区的大小。
  2. 进行MSS大小的TCP分段。
  3. 以太网帧的payload大于NTU进行IP分片。

TCP粘包/拆包问题解决方案

粘包拆包问题的解决方案，根据业界主流协议有三种方案：

  1. 消息定长，例如每个报文的大小固定为200个字节，如果不够，空位补空格。
  2. 在包尾部增加特殊字符进行分割，例如增加回车等。
  3. 将消息分为消息头和消息体，在消息头中包含表示消息总长度的字段，然后进行业务逻辑处理。

Netty如何去解决粘包拆包问题

1. 分隔符类：DelimiterBasedFrameDecoder（自定义分隔符）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
public class Server {

	public static void main(String[] args) throws InterruptedException {
		//1.第一个线程组是用于接收Client连接的
		EventLoopGroup bossGroup = new NioEventLoopGroup(); 
		//2.第二个线程组是用于实际的业务处理操作的
		EventLoopGroup workerGroup = new NioEventLoopGroup();
		//3.声明一个启动NIO服务的辅助启动类，就是对我们的Server端进行一系列的配置
		ServerBootstrap b = new ServerBootstrap();
		//将两个工作线程组加入ServerBootstrap
		b.group(bossGroup, workerGroup)
		//使用NioServerSocketChannel这种类型的通道
		.channel(NioServerSocketChannel.class)
		//一定要使用childHandler绑定具体的时间处理器
		.childHandler(new ChannelInitializer<SocketChannel>() {
			@Override
			protected void initChannel(SocketChannel sc) throws Exception {
				//设置特殊分隔符
				ByteBuf buf = Unpooled.copiedBuffer("$_".getBytes());
				sc.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf));
				//设置字符串形式的解码，将buffer类型转成字符串类型处理
				sc.pipeline().addLast(new StringDecoder());
				sc.pipeline().addLast(new ServerHandler());
			}
		})
		//设置指定通道实现的配置参数，也就是设置tcp缓冲区
		.option(ChannelOption.SO_BACKLOG, 128)
		//保持连接
		.option(ChannelOption.SO_KEEPALIVE, true);

		//绑定指定端口，进行监听
		ChannelFuture f = b.bind(8765).sync();
		
		//Thread.sleep(Integer.MAX_VALUE);
		f.channel().closeFuture().sync();
		
		bossGroup.shutdownGracefully();
		workerGroup.shutdownGracefully();
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public class ServerHandler extends ChannelHandlerAdapter{

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		String request = (String) msg;
		System.out.println("Server: " + request);
		String response = "服务器响应$_";
		//ctx.writeAndFlush(response);
		ctx.writeAndFlush(Unpooled.copiedBuffer(response.getBytes()));
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		 cause.printStackTrace();
		 ctx.close();
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
public class Client {

	public static void main(String[] args) throws InterruptedException {
		EventLoopGroup workgroup = new NioEventLoopGroup();
		Bootstrap b = new Bootstrap();
		b.group(workgroup)
		.channel(NioSocketChannel.class)
		.handler(new ChannelInitializer<SocketChannel>() {
			@Override
			protected void initChannel(SocketChannel sc) throws Exception {
				//同服务端编码
				ByteBuf buf = Unpooled.copiedBuffer("$_".getBytes());
				sc.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf));
				sc.pipeline().addLast(new StringDecoder());
				sc.pipeline().addLast(new ClientHandler());
			}
		});
		
		ChannelFuture cf = b.connect("127.0.0.1", 8765).sync();
		
		//向服务端写数据
		cf.channel().writeAndFlush(Unpooled.copiedBuffer("aa$_".getBytes()));
		cf.channel().writeAndFlush(Unpooled.copiedBuffer("bbbb$_".getBytes()));
		cf.channel().writeAndFlush(Unpooled.copiedBuffer("ccccc$_".getBytes()));
		
		cf.channel().closeFuture().sync();
		workgroup.shutdownGracefully();
	}
}

/*
服务端响应：
Server: aa
Server: bbbb
Server: ccccc

客户端响应：
Client: 服务器响应
Client: 服务器响应
Client: 服务器响应
*/
1
2
3
4
5
6
7
8
9
10
11
12
13
14
public class ClientHandler extends ChannelHandlerAdapter{

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		String response = (String) msg;
		System.out.println("Client: " + response);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		 cause.printStackTrace();
		 ctx.close();
	}
}

2. FixedLengthFrameDecoder（定长）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
public class Server {

	public static void main(String[] args) throws InterruptedException {
		//1.第一个线程组是用于接收Client连接的
		EventLoopGroup bossGroup = new NioEventLoopGroup(); 
		//2.第二个线程组是用于实际的业务处理操作的
		EventLoopGroup workerGroup = new NioEventLoopGroup();
		//3.声明一个启动NIO服务的辅助启动类，就是对我们的Server端进行一系列的配置
		ServerBootstrap b = new ServerBootstrap();
		//将两个工作线程组加入ServerBootstrap
		b.group(bossGroup, workerGroup)
		//使用NioServerSocketChannel这种类型的通道
		.channel(NioServerSocketChannel.class)
		//一定要使用childHandler绑定具体的时间处理器
		.childHandler(new ChannelInitializer<SocketChannel>() {
			@Override
			protected void initChannel(SocketChannel sc) throws Exception {
				//设置定长字符串接收
				sc.pipeline().addLast(new FixedLengthFrameDecoder(5));
				//设置字符串形式的解码，将buffer类型转成字符串类型处理
				sc.pipeline().addLast(new StringDecoder());
				sc.pipeline().addLast(new ServerHandler());
			}
		})
		//设置指定通道实现的配置参数，也就是设置tcp缓冲区
		.option(ChannelOption.SO_BACKLOG, 128)
		//保持连接
		.option(ChannelOption.SO_KEEPALIVE, true);

		//绑定指定端口，进行监听
		ChannelFuture f = b.bind(8765).sync();
		
		//Thread.sleep(Integer.MAX_VALUE);
		f.channel().closeFuture().sync();
		
		bossGroup.shutdownGracefully();
		workerGroup.shutdownGracefully();
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public class ServerHandler extends ChannelHandlerAdapter{

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		String request = (String) msg;
		System.out.println("Server: " + request);
		String response = "服务器响应";
		//ctx.writeAndFlush(response);
		ctx.writeAndFlush(Unpooled.copiedBuffer(response.getBytes()));
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		 cause.printStackTrace();
		 ctx.close();
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
public class Client {

	public static void main(String[] args) throws InterruptedException {
		EventLoopGroup workgroup = new NioEventLoopGroup();
		Bootstrap b = new Bootstrap();
		b.group(workgroup)
		.channel(NioSocketChannel.class)
		.handler(new ChannelInitializer<SocketChannel>() {
			@Override
			protected void initChannel(SocketChannel sc) throws Exception {
				//设置定长字符串接收，这里设置10个字节是因为“服务器响应”每个字符占用两个字节
				sc.pipeline().addLast(new FixedLengthFrameDecoder(10));
				sc.pipeline().addLast(new StringDecoder());
				sc.pipeline().addLast(new ClientHandler());
			}
		});
		
		ChannelFuture cf = b.connect("127.0.0.1", 8765).sync();
		
		//向服务端写数据
		cf.channel().writeAndFlush(Unpooled.copiedBuffer("aaaaaccccc".getBytes()));
		cf.channel().writeAndFlush(Unpooled.copiedBuffer("bbbbbbbbbb".getBytes()));
		//由于服务端设置按5个字节接收数据，所以ddd将会丢失
		//cf.channel().writeAndFlush(Unpooled.copiedBuffer("ddd".getBytes()));
		//使用空格补位
		cf.channel().writeAndFlush(Unpooled.copiedBuffer("ddd  ".getBytes()));
		
		cf.channel().closeFuture().sync();
		workgroup.shutdownGracefully();
	}
}

/*
服务端响应：
Server: aaaaa
Server: ccccc
Server: bbbbb
Server: bbbbb
Server: ddd  

客户端响应：
Client: 服务器响应
Client: 服务器响应
Client: 服务器响应
Client: 服务器响应
Client: 服务器响应
*/
1
2
3
4
5
6
7
8
9
10
11
12
13
14
public class ClientHandler extends ChannelHandlerAdapter{

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		String response = (String) msg;
		System.out.println("Client: " + response);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		 cause.printStackTrace();
		 ctx.close();
	}
}
-------------本文结束感谢您的阅读-------------