一、什么是粘包拆包
举个例子
客户端和服务器建立一个连接，客户端发送一条消息，客户端关闭与服务端的连接。
客户端和服务器简历一个连接，客户端连续发送两条消息，客户端关闭与服务端的连接。
对于第一种情况，服务端的处理流程可以是这样的：当客户端与服务端的连接建立成功之后，服务端不断读取客户端发送过来的数据，当客户端与服务端连接断开之后，服务端知道已经读完了一条消息，然后进行解码和后续处理...。对于第二种情况，如果按照上面相同的处理逻辑来处理，那就有问题了，我们来看看第二种情况下客户端发送的两条消息递交到服务端有可能出现的情况：
第一种情况：
服务端一共读到两个数据包，第一个包包含客户端发出的第一条消息的完整信息，第二个包包含客户端发出的第二条消息，那这种情况比较好处理，服务器只需要简单的从网络缓冲区去读就好了，第一次读到第一条消息的完整信息，消费完再从网络缓冲区将第二条完整消息读出来消费。
第二种情况：
服务端一共就读到一个数据包，这个数据包包含客户端发出的两条消息的完整信息，这个时候基于之前逻辑实现的服务端就蒙了，因为服务端不知道第一条消息从哪儿结束和第二条消息从哪儿开始，这种情况其实是发生了TCP粘包。
第三种情况：
服务端一共收到了两个数据包，第一个数据包只包含了第一条消息的一部分，第一条消息的后半部分和第二条消息都在第二个数据包中，或者是第一个数据包包含了第一条消息的完整信息和第二条消息的一部分信息，第二个数据包包含了第二条消息的剩下部分，这种情况其实是发送了TCP拆，因为发生了一条消息被拆分在两个包里面发送了，同样上面的服务器逻辑对于这种情况是不好处理的。
二、如何解决拆包粘包
既然知道了tcp是无界的数据流，且协议本身无法避免粘包，拆包的发生，那我们只能在应用层数据协议上，加以控制。通常在制定传输数据时，可以使用如下方法：
1、使用带消息头的协议、消息头存储消息开始标识及消息长度信息，服务端获取消息头的时候解析出消息长度，然后向后读取该长度的内容。
2、设置定长消息，服务端每次读取既定长度的内容作为一条完整消息。
3、设置消息边界，服务端从网络流中按消息编辑分离出消息内容。
三、netty解决拆包粘包的三种方案
这里只贴出核心代码，源码在文章后面会给出地址
1、使用FixedLengthFrameDecoder固定长度解码器
服务端：
public void bind(int port) throws InterruptedException {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
//BACKLOG用于构造服务端套接字ServerSocket对象，标识当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度。如果未设置或所设置的值小于1，Java将使用默认值50。
.option(ChannelOption.SO_BACKLOG, 10024)
//可以对入站\出站事件进行日志记录，从而方便我们进行问题排查。比如REGISTERED、ACTIVE、CLOSE、INACTIVE、UNREGISTERED事件都有日志
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChildFixedLengthChannelHandler());
ChannelFuture future = bootstrap.bind(port).sync();
future.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws InterruptedException {
new NettyServer().bind(8080);
}private class ChildFixedLengthChannelHandler extends ChannelInitializer<SocketChannel>{
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 这里将FixedLengthFrameDecoder解码指定长度为100
ch.pipeline().addLast(new FixedLengthFrameDecoder(100));
// 将前一步解码的数据转码为字符串
ch.pipeline().addLast(new StringDecoder());
// 这里FixedLengthMessageEncoder是我们自定义的，用于将长度不足16的消息进行补全空格
ch.pipeline().addLast(new FixedLengthMessageEncoder(100));
// 最终的数据处理
ch.pipeline().addLast(new NettyServerHandler());
}
}FixedLengthMessageEncoder是自定义编码器，长度不足的空格补齐，代码如下:
package com.zhouzy.base.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
public class FixedLengthMessageEncoder extends MessageToByteEncoder<String> {
private int length;
public FixedLengthMessageEncoder(int length) {
this.length = length;
}
@Override
protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out)
throws Exception {
// 对于超过指定长度的消息，这里直接抛出异常
if (msg.length() > length) {
throw new UnsupportedOperationException("字符串长度超出长度了，指定长度为:"+length);
}
// 如果长度不足，则进行补全
if (msg.length() < length) {
msg = addSpace(msg);
}
//Unpooled.wrappedBuffer实现零拷贝，将字符串转为ByteBuf
ctx.writeAndFlush(Unpooled.wrappedBuffer(msg.getBytes()));
}
// 如果没有达到指定长度进行空格补全
private String addSpace(String msg) {
StringBuilder builder = new StringBuilder(msg);
for (int i = 0; i < length - msg.length(); i++) {
builder.append(" ");
}
return builder.toString();
}
}
package com.zhouzy.base.netty.server;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSONObject;
import com.zhouzy.base.netty.UserInfo;
public class NettyServerHandler extends SimpleChannelInboundHandler<String> {
Logger log = LoggerFactory.getLogger(NettyServerHandler.class);
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
log.info("收到客户端的消息:{}",msg);
ctx.writeAndFlush(msg);
}
}
客户端:
Bootstrap bootstrap = null;
EventLoopGroup group = new NioEventLoopGroup();
public void connect(final String host, final int port) throws Exception {
ChannelFuture future = null;
try {
if (bootstrap == null) {
bootstrap = new Bootstrap();
}
bootstrap.group(group)
.channel(NioSocketChannel.class)
//TCP_NODELAY就是用于启用或关于Nagle算法。如果要求高实时性，有数据发送时就马上发送，就将该选项设置为true关闭Nagle算法；如果要减少发送次数减少网络交互，就设置为false等累积一定大小后再发送。默认为false。
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChildFixedLengthChannelHandler());
future = bootstrap.connect(host, port).sync();
future.channel().closeFuture().sync();
//加一个监听器，断线重连
future.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture channelFuture) throws Exception {
if(channelFuture.isSuccess()){
System.out.println("连接服务端成功");
}else{
System.out.println("每隔2s重连....");
channelFuture.channel().eventLoop().schedule(new Runnable() {
@Override
public void run() {
try {
connect(host,port);
} catch (Exception e) {
log.error("重连报错:{}",e.getMessage(),e);
}
}
},2,TimeUnit.SECONDS);
}
}
});
} finally {
group.shutdownGracefully();
}
}
public static void main(String[] args) throws InterruptedException {
try {
new NettyClient().connect("localhost", 8080);
} catch (Exception e) {
log.error("连接服务器异常:{}",e.getMessage(),e);
}
}/**
* 固定长度
* @author Administrator
*
*/
private class ChildFixedLengthChannelHandler extends ChannelInitializer<SocketChannel>{
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 对服务端发送的消息进行粘包和拆包处理，由于服务端发送的消息已经进行了空格补全，
// 并且长度为100，因而这里指定的长度也为100
ch.pipeline().addLast(new FixedLengthFrameDecoder(100));
// 将粘包和拆包处理得到的消息转换为字符串
ch.pipeline().addLast(new StringDecoder());
// 对客户端发送的消息进行空格补全，保证其长度为100
ch.pipeline().addLast(new FixedLengthMessageEncoder(100));
// 客户端发送消息给服务端，并且处理服务端响应的消息
ch.pipeline().addLast(new NettyClientHandler());
}
}
2、使用DelimiterBasedFrameDecoder自定义字符进行分隔
大部分代码同上，只是处理类childhandler不一样
private class ChildDelimiterChannelHandler extends ChannelInitializer<SocketChannel>{
@Override
protected void initChannel(SocketChannel ch) throws Exception {
String delimiter = "_$";
//对服务端返回的消息通过_$进行分隔，并且每次查找的最大大小为1024字节
ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,Unpooled.wrappedBuffer(delimiter.getBytes())));
// 将前一步解码的数据转码为字符串
ch.pipeline().addLast(new StringDecoder());
// 对客户端发送的数据进行编码，这里主要是在客户端发送的数据最后添加分隔符
ch.pipeline().addLast(new DelimiterBasedMessageEncoder(delimiter));
// 最终的数据处理
ch.pipeline().addLast(new NettyServerHandler());
}
}DelimiterBasedMessageEncoder这个是自定义编码器，发送时自动加上字符，代码如下:
package com.zhouzy.base.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
public class DelimiterBasedMessageEncoder extends MessageToByteEncoder<String>{
//特殊字符-分隔符
private String delimiter;
public DelimiterBasedMessageEncoder(String delimiter) {
this.delimiter = delimiter;
}
@Override
protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out)
throws Exception {
// 在响应的数据后面添加分隔符
ctx.writeAndFlush(Unpooled.wrappedBuffer((msg + delimiter).getBytes()));
}
}
客户端跟服务端代码类似，就不贴代码了
3、LengthFieldBasedFrameDecoder和LengthFieldPrepender结合
进行长度字段解码，就是消息体传了消息的长度
handler代码：
/**
* 理粘拆包的主要思想是在生成的数据包中添加一个长度字段,用于记录当前数据包的长度,LengthFieldBasedFrameDecoder会按照参数指定的包长度偏移量数据对接收到的数据进行解码，从而得到目标消息体数据；而LengthFieldPrepender则会在响应的数据前面添加指定的字节数据，这个字节数据中保存了当前消息体的整体字节数据长度。
* @author zhouzhiyao
*
*/
private class ChildLengthFieldChannelHandler extends ChannelInitializer<SocketChannel>{
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// 这里将LengthFieldBasedFrameDecoder添加到pipeline的首位，因为其需要对接收到的数据
// 进行长度字段解码，这里也会对数据进行粘包和拆包处理
/*
* maxFrameLength：指定了每个包所能传递的最大数据包大小；
lengthFieldOffset：指定了长度字段在字节码中的偏移量；
lengthFieldLength：指定了长度字段所占用的字节长度；
lengthAdjustment：对一些不仅包含有消息头和消息体的数据进行消息头的长度的调整，这样就可以只得到消息体的数据，这里的lengthAdjustment指定的就是消息头的长度；
initialBytesToStrip：对于长度字段在消息头中间的情况，可以通过initialBytesToStrip忽略掉消息头以及长度字段占用的字节。
*/
ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 0, 2, 0, 2));
// LengthFieldPrepender是一个编码器，主要是在响应字节数据前面添加字节长度字段
ch.pipeline().addLast(new LengthFieldPrepender(2));
// 对经过粘包和拆包处理之后的数据进行json反序列化，从而得到UserInfo对象
ch.pipeline().addLast(new JsonDecoder());
// 对响应数据进行编码，主要是将UserInfo对象序列化为json
ch.pipeline().addLast(new JsonEncoder());
// 最终的数据处理
ch.pipeline().addLast(new NettyServerHandler());
}
}JsonDecoder和JsonEncoder对象序列化编码、解码器
package com.zhouzy.base.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import com.alibaba.fastjson.JSON;
public class JsonEncoder extends MessageToByteEncoder<UserInfo> {
@Override
protected void encode(ChannelHandlerContext ctx, UserInfo user, ByteBuf buf)
throws Exception {
if(null != user){
String json = JSON.toJSONString(user);
ctx.writeAndFlush(Unpooled.wrappedBuffer(json.getBytes()));
}
}
}
package com.zhouzy.base.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageDecoder;
import io.netty.util.CharsetUtil;
import java.util.List;
import com.alibaba.fastjson.JSON;
public class JsonDecoder extends MessageToMessageDecoder<ByteBuf> {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out)
throws Exception {
byte[] bytes = new byte[buf.readableBytes()];
buf.readBytes(bytes);
UserInfo user = JSON.parseObject(new String(bytes, CharsetUtil.UTF_8), UserInfo.class);
if(null != user){
out.add(user);
}
}
}
对象
package com.zhouzy.base.netty;
public class UserInfo {
private String name;
private int age;
private String address;
private String sex;
private String phone;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
public String getAddress() {
return address;
}
public void setAddress(String address) {
this.address = address;
}
public String getSex() {
return sex;
}
public void setSex(String sex) {
this.sex = sex;
}
public String getPhone() {
return phone;
}
public void setPhone(String phone) {
this.phone = phone;
}
}
package com.zhouzy.base.netty.server;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSONObject;
import com.zhouzy.base.netty.UserInfo;
public class NettyServerHandler extends SimpleChannelInboundHandler<UserInfo> {
Logger log = LoggerFactory.getLogger(NettyServerHandler.class);
@Override
protected void channelRead0(ChannelHandlerContext ctx, UserInfo user) throws Exception {
log.info("收到客户端的消息:{}",JSONObject.toJSONString(user));
ctx.writeAndFlush(user);
}
}
测试结果：
?