粘包拆包
概述
TCP是面向流的协议,TCP在网络上传输的数据就是一连串的数据,完全没有分界线。
TCP协议的底层并不了解上层业务的具体定义,它会根据TCP缓冲区的实际情况进行包的划分。
在业务层面认为一个完整的包可能会被TCP拆分成多个小包进行发送,也可能把多个小的包封装成一个大的数据包进行发送,这就是所谓的TCP粘包拆包问题，如上图会存在多种情况
原因分析
TCP数据流最终发到目的地,必须通过以太网协议封装成一个个的以太网帧发送出去,以太网数据帧大小最小64字节,最大1518字节,除去header部分,其数据payload为46到1500字节。所以如果以太网帧的payload大于MTU(默认1500字节)就需要进行拆包
解决方案
由于TCP协议底层无法理解上层的业务数据，所以在底层是无法保证数据包不被拆分和重组的，所以该问题只能通过上层的应用层协议设计来解决,常见方案如下：
- 消息定长,发送方和接收方规定固定大小的消息长度,例如每个报文大小固定为200字节，如果不够，空位补空格
- 增加特殊字符进行分割,例如FTP协议
- 自定义协议,将消息分为消息头和消息体，消息头中包含消息总长度，这样服务端就可以知道每个数据包的具体长度了，知道了发送数据包的具体边界后，就可以解决粘包和拆包问题了
Netty解决方案
基于JDK原生的socket或者nio解决粘包拆包问题比较麻烦;作为一款非常强大的网络通信框架,Netty提供了多种编码器用于解决粘包拆包问题,只要掌握这些类库的使用即可解决粘包拆包问题,常见的类库包括：
- LineBasedFrameDecoder：基于行的解码器,遇到 “\n”、"\r\n"会被作为行分隔符
- FixedLengthFrameDecoder：基于固定长度的解码器
- DelimiterBasedFrameDecoder：基于分隔符的帧解码器
- ByteToMessageDecoder：处理自定义消息协议的解码器
- ReplayingDecoder：继承ByteToMessageDecoder,网络缓慢并且消息格式复杂时可能慢于ByteToMessageDecoder
自定义消息协议+编码/解码器解决粘包拆包问题实战
Netty中使用自定义协议(本质就是解决服务器端每次读取数据长度的问题) + 编码解码器来解决
客户端
package com.bierce.io.netty.stickingAndunpacking;
import io.netty.bootstrap.Bootstrap;
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.Charset;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
public class MyStickingUnpackingClientTest {
private String host;
private int port;
public MyStickingUnpackingClientTest(String host, int port){
this.host = host;
this.port = port;
}
public void run(){
NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(eventLoopGroup)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new MyMessageEncoder()); //自定义编码器
pipeline.addLast(new MyMessageDecoder()); //自定义解码器
pipeline.addLast(new MyStickingUnpackingClientHandler()); //自定义业务处理器
}
});
ChannelFuture cf = bootstrap.connect(host, port).sync();
Channel channel = cf.channel();
System.out.println("北京时间-" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy/MM/dd HH:mm:ss")) + ": " + "Client"+ channel.localAddress()+ " start Successful!!!");
cf.channel().closeFuture().sync();
} catch (InterruptedException e) {
throw new RuntimeException(e);
} finally {
eventLoopGroup.shutdownGracefully();
}
}
public static void main(String[] args) {
new MyStickingUnpackingClientTest("127.0.0.1", 9999).run();
}
}
class MyStickingUnpackingClientHandler extends SimpleChannelInboundHandler<MessageProtocal> {
private int count;
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
for (int i = 0; i < 5; i++) {
String msg = "Hello Bierce";
byte[] data = msg.getBytes(Charset.forName("UTF-8"));
int length = msg.getBytes(Charset.forName("UTF-8")).length;
MessageProtocal msgProtocal = new MessageProtocal();
msgProtocal.setData(data);
msgProtocal.setLength(length);
ctx.writeAndFlush(msgProtocal);
}
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, MessageProtocal msg) throws Exception {
byte[] data = msg.getData();
int length = msg.getLength();
System.out.println("客户端接收到服务端响应信息包括：");
System.out.println("长度=" + length + ";内容=" + new String(data, Charset.forName("UTF-8")) + ";消息包数量=" + (++this.count));
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("异常消息： " + cause.getMessage());
ctx.close();
}
}服务端
package com.bierce.io.netty.stickingAndunpacking;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import java.nio.charset.Charset;
import java.util.UUID;
public class MyStickingUnpackingServerTest {
private int port;
public MyStickingUnpackingServerTest(int port){
this.port = port;
}
public void run(){
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup,workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new MyMessageDecoder()); //自定义解码器
pipeline.addLast(new MyMessageEncoder()); //自定义编码器
pipeline.addLast(new MyStickingUnpackingServerHandler());
}
});
System.out.println("服务器启动成功!");
serverBootstrap.bind(port).sync().channel().closeFuture().sync();
} catch (InterruptedException e) {
throw new RuntimeException(e);
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) {
new MyStickingUnpackingServerTest(9999).run();
}
}
class MyStickingUnpackingServerHandler extends SimpleChannelInboundHandler<MessageProtocal> {
private int count;
@Override
protected void channelRead0(ChannelHandlerContext ctx, MessageProtocal msg) throws Exception {
int length = msg.getLength();
byte[] data = msg.getData();
System.out.println();
System.out.println("服务器接收到的信息包括：");
System.out.println("长度=" + length + ";内容=" + new String(data, Charset.forName("UTF-8")) + ";消息包数量=" + (++this.count));
System.out.println("------------服务端读取成功------------");
//回复给客户端信息
String responseContent = UUID.randomUUID().toString();
int responseLen = responseContent.getBytes("UTF-8").length;
MessageProtocal msgProtocal = new MessageProtocal();
msgProtocal.setLength(responseLen);
msgProtocal.setData(responseContent.getBytes("UTF-8"));
ctx.writeAndFlush(msgProtocal);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("异常消息：" + cause.getMessage());
ctx.close();
}
}自定义消息协议类
package com.bierce.io.netty.stickingAndunpacking;
/**
* 自定义消息服协议
*/
public class MessageProtocal {
private int length;
private byte[] data;
public int getLength() {
return length;
}
public void setLength(int length) {
this.length = length;
}
public byte[] getData() {
return data;
}
public void setData(byte[] data) {
this.data = data;
}
}
自定义消息解码编码器类
package com.bierce.io.netty.stickingAndunpacking;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import io.netty.handler.codec.ReplayingDecoder;
import java.util.List;
public class MyMessageDecoder extends ReplayingDecoder<Void> {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
System.out.println("MyMessageDecoder method called");
int length = in.readInt();
byte[] data = new byte[length];
in.readBytes(data);
MessageProtocal messageProtocal = new MessageProtocal();
messageProtocal.setLength(length);
messageProtocal.setData(data);
out.add(messageProtocal);
}
}
class MyMessageEncoder extends MessageToByteEncoder<MessageProtocal> {
@Override
protected void encode(ChannelHandlerContext ctx, MessageProtocal msg, ByteBuf out) throws Exception {
System.out.println("MyMessageEncoder method called");
out.writeInt(msg.getLength());
out.writeBytes(msg.getData());
}
}
效果图
