Netty粘包拆包
1、什么是粘包拆包？
粘包拆包一般是发生在TCP协议层的，TCP作为传输层一般不知道上层（这里指的是我们的代码业务层）数据的具体含义，它会根据TCP缓冲区 的实际情况进行数据包的划分，所以在业务上认为是一个完整的包，可能会被TCP拆分成多个包进行发送，也有可能把多个小的包封装成 一个大的数据包发送，这就是所谓的TCP粘包和拆包问题。面向流的通信是无消息保护边界的。
client发了两个数据包D1和D2，但是server端可能会收到如下几种情况的数据。如下图所示：
2、解决方案
1、给每个消息一个固定的长度，比如说 100 字节，就是当这条消息不够一百个字节时就用空格补齐。
这样的话会产生两个问题，第一个呢就是如果发送的数据长度很小，但是要补齐 100 个字节，就会浪费很大的内存空间，第二个呢，一次数据的长度最大只能为100个字节。
2、在数据包尾部添加特殊分隔符，比如下划线，中划线等，这种方法简单易行，但选择分隔符的时候一定要注意每条数据的内部一定不 能出现分隔符。
3、发送长度：发送每条数据的时候，将数据的长度一并发送，比如可以选择每条数据的前4位是数据的长度，应用层处理时可以根据长度 来判断每条数据的开始和结束。
// （回车换行分包）
LineBasedFrameDecoder
// （特殊分隔符分包）
DelimiterBasedFrameDecoder
// （固定长度报文来分包）
FixedLengthFrameDecoder
3、代码示例
这里我们示例一个第三种解决方案的代码。
涉及到下面 7 个类。
3.1、MyClient
public class MyClient {
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 ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new MyMessageEncoder());
pipeline.addLast(new MyClientHandler());
}
});
System.out.println("netty client start。。");
ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 9000).sync();
channelFuture.channel().closeFuture().sync();
}finally {
group.shutdownGracefully();
}
}
}
3.2、MyClientHandler
public class MyClientHandler extends SimpleChannelInboundHandler<MyMessageProtocol> {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
for(int i = 0; i< 10; i++) {
String msg = "你好，我是张三！";
//创建协议包对象
MyMessageProtocol messageProtocol = new MyMessageProtocol();
messageProtocol.setLen(msg.getBytes(CharsetUtil.UTF_8).length);
messageProtocol.setContent(msg.getBytes(CharsetUtil.UTF_8));
ctx.writeAndFlush(messageProtocol);
}
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, MyMessageProtocol msg) throws Exception {
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
3.3、MyServer
public class MyServer {
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 ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new MyMessageDecoder());
pipeline.addLast(new MyServerHandler());
}
});
System.out.println("netty server start。。");
ChannelFuture channelFuture = serverBootstrap.bind(9000).sync();
channelFuture.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
3.4、MyServerHandler
public class MyServerHandler extends SimpleChannelInboundHandler<MyMessageProtocol> {
private int count;
@Override
protected void channelRead0(ChannelHandlerContext ctx, MyMessageProtocol msg) throws Exception {
System.out.println("====服务端接收到消息如下====");
System.out.println("长度=" + msg.getLen());
System.out.println("内容=" + new String(msg.getContent(), CharsetUtil.UTF_8));
System.out.println("服务端接收到消息包数量=" + (++this.count));
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
3.5、MyMessageDecoder
public class MyMessageDecoder extends ByteToMessageDecoder {
int length = 0;
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
System.out.println();
System.out.println("MyMessageDecoder decode 被调用");
//需要将得到二进制字节码-> MyMessageProtocol 数据包(对象)
System.out.println(in);
if(in.readableBytes() >= 4) {
if (length == 0){
length = in.readInt();
}
if (in.readableBytes() < length) {
System.out.println("当前可读数据不够，继续等待。。");
return;
}
byte[] content = new byte[length];
if (in.readableBytes() >= length){
in.readBytes(content);
//封装成MyMessageProtocol对象，传递到下一个handler业务处理
MyMessageProtocol messageProtocol = new MyMessageProtocol();
messageProtocol.setLen(length);
messageProtocol.setContent(content);
out.add(messageProtocol);
}
length = 0;
}
}
}
3.6、MyMessageEncoder
public class MyMessageEncoder extends MessageToByteEncoder<MyMessageProtocol> {
@Override
protected void encode(ChannelHandlerContext ctx, MyMessageProtocol msg, ByteBuf out) throws Exception {
System.out.println("MyMessageEncoder encode 方法被调用");
out.writeInt(msg.getLen());
out.writeBytes(msg.getContent());
}
}
3.7、MyMessageProtocol
public class MyMessageProtocol {
/**
* 定义一次发送包体长度
*/
private int len;
/**
* 一次发送包体内容
*/
private byte[] content;
public int getLen() {
return len;
}
public void setLen(int len) {
this.len = len;
}
public byte[] getContent() {
return content;
}
public void setContent(byte[] content) {
this.content = content;
}
}