案例重现
首先我们通过具体的case重现一下TCP粘包的过程
我们模拟下故障场景，客户端循环一百次调用服务端传输报文，服务端接收报文并打印接收报文和计数，同时根据报文回应客户端
服务端代码
public class TimeServerHandler extends ChannelHandlerAdapter {
private int count;
@Override
public void channelRead(ChannelHandlerContext ctx,Object msg) throws Exception{
ByteBuf byteBuf = (ByteBuf)msg;
byte[] req = new byte[byteBuf.readableBytes()];
byteBuf.readBytes(req);
System.out.println("received msg length:"+req.length);
String body = new String(req,"UTF-8").substring(0,req.length-System.getProperty("line.separator").length());
System.out.println("the time server receive order:"+body + ";the counter is:" + ++count);
String currentTIme = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new java.util.Date(
System.currentTimeMillis()
).toString():"BAD ORDER";
currentTIme = currentTIme + System.getProperty("line.separator");
ByteBuf resp = Unpooled.copiedBuffer(currentTIme.getBytes());
ctx.writeAndFlush(resp);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
}
public class TimeServer {
public void bind(int port) throws Exception {
//创建两个线程组 一个用于服务端接收客户端的连接
EventLoopGroup bossGroup = new NioEventLoopGroup();
//一个用于网络读写
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG,1024)
.childHandler(new ChildChannelHander());
ChannelFuture future = b.bind(port).sync();
future.channel().closeFuture().sync();
}finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
private class ChildChannelHander extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
// ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
// ch.pipeline().addLast(new StringDecoder());
ch.pipeline().addLast(new TimeServerHandler());
}
}
public static void main(String[] args) {
int port = 8080;
if (args != null && args.length >0) {
try {
port = Integer.valueOf(args[0]);
}catch (NumberFormatException e) {
}
}
try {
new TimeServer().bind(port);
} catch (Exception e) {
e.printStackTrace();
}
}
}
客户端代码
public class TimeClientHandler extends ChannelHandlerAdapter {
private int count;
private byte[] req;
public TimeClientHandler() {
req = ("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes();
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ByteBuf message = null;
for (int i = 0; i< 100; i++) {
message = Unpooled.buffer(req.length);
message.writeBytes(req);
ctx.writeAndFlush(message);
}
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf byteBuf = (ByteBuf)msg;
byte[] req = new byte[byteBuf.readableBytes()];
byteBuf.readBytes(req);
String body = new String(req,"UTF-8");
System.out.println("Now is : "+body + "the counter is :"+ ++count);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("Unexpected exception from downstream :"+cause.getMessage());
ctx.close();
}
}
public class TimeClient {
public void connect(int port, String host) throws Exception {
//创建读写io线程组
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group).channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
//socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
//socketChannel.pipeline().addLast(new StringDecoder());
socketChannel.pipeline().addLast(new TimeClientHandler());
}
});
ChannelFuture f = b.connect(host, port).sync();
f.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
public static void main(String[] args) {
int port = 8080;
if (args != null && args.length > 0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
}
}
try {
new TimeClient().connect(port, "127.0.0.1");
} catch (Exception e) {
e.printStackTrace();
}
}
}
程序运行结果
服务端
image.png
image.png
客户端
image.png
按照设计初衷，客户端应该发送了100次数据给服务端，服务端每接收一次数据就回应一次客户端，那么客户端应该收到一百次消息，但是实际上客户端就收到2次消息，服务端也只收到两次消息。说明服务端和客户端都发生了粘包现象。
产生粘包拆包的原因
TCP协议是基于流的协议，是没有边界的一串数据，它会根据TCP缓冲区的实际情况进行包的拆分，上述例子中默认TCP缓存区的大小是1024个字节，服务端第一次收到的数据大小正好是1024个字节，也就是说多个小的报文可能封装出一个大的数据进行传送，而一个大的报文可能会被拆分成多个小包进行传送。
解决办法
由于TCP是底层通讯协议，它不关心上层业务，无法保证数据包不会拆包或者粘包，那么这个问题只能通过上层协议来解决，通常的解决办法有以下几点：
1、消息定长，例如每个报文都是500个字节，如果报文不够500个字节，那么就填充
2、报文尾部增加特殊分隔符
3、消息分为消息头和消息体，消息头定义消息的长度，消息体还是真实传送的报文（类型UDP协议）
Netty解决粘包拆包问题
Netty提供了多种编码解码器处理上述问题，其中可以使用LineBasedFrameDecoder解决粘包问题
服务端代码只要上述TimeServer加一个LineBasedFrameDecoder的ChannelHandler
private class ChildChannelHander extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
ch.pipeline().addLast(new TimeServerHandler());
}
}
客户端代码只要上述TimeClient中加一个LineBasedFrameDecoder的ChannelHandler
public void connect(int port, String host) throws Exception {
//创建读写io线程组
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group).channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
//socketChannel.pipeline().addLast(new StringDecoder());
socketChannel.pipeline().addLast(new TimeClientHandler());
}
});
ChannelFuture f = b.connect(host, port).sync();
f.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
代码运行结果：
[图片上传中...(image.png-bdb177-1553093851212-0)]
image.png
客户端运行结果
image.png
由此可见，增加LineBasedFrameDecoder之后解决了粘包问题
LineBasedFrameDecoder的工作原理是遍历缓冲区的可读字节，判断是否是“\n”或者"\r\n"，如果有，那么就以该位置作为结束位置，从缓冲区可读区域到结束位置作为一个完整报文，他是以标识符作为解码器。
LineBasedFrameDecoder 源码分析：
image.png
我们先看下LineBasedFrameDecoder的类继承图，继承了ByteToMessageDecoder方法，ByteToMessageDecoder是将ByteBuf字节解码成其他消息类型的抽象类，它有一个关键的方法：
callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
作用就是从Channel读到的字节数据转换成对应的具体消息类型List<Object>输出,而具体怎么解码是由继承它的子类实现
protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;
LineBasedFrameDecoder继承ByteToMessageDecoder并实现了具体的decode方法。
当LineBasedFrameDecoder加入到ChannelPipeLine管道后，当Channel缓冲区中有数据时会调用channelRead方法,ByteToMessageDecoder的channelRead源码：
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//1、判断是否是字节数据
if (msg instanceof ByteBuf) {
//2、初始化输出数据
RecyclableArrayList out = RecyclableArrayList.newInstance();
try {
ByteBuf data = (ByteBuf) msg;
//3、判断cumulation是否为空，初始化后cumulation为空,first为true
first = cumulation == null;
if (first) {
cumulation = data;
} else {
if (cumulation.writerIndex() > cumulation.maxCapacity() - data.readableBytes()) {
expandCumulation(ctx, data.readableBytes());
}
cumulation.writeBytes(data);
data.release();
}
//调用解析程序，将字节流转换传out集合对象
callDecode(ctx, cumulation, out);
} catch (DecoderException e) {
throw e;
} catch (Throwable t) {
throw new DecoderException(t);
} finally {
//将缓冲区回收防止内存溢出
if (cumulation != null && !cumulation.isReadable()) {
cumulation.release();
cumulation = null;
}
//获得返回数据结果的大小
int size = out.size();
decodeWasNull = size == 0;
//依次调用ChannelPipeLine的下一个ChannelRead方法，并将编码后的结果传递下去
for (int i = 0; i < size; i ++) {
ctx.fireChannelRead(out.get(i));
}
out.recycle();
}
} else {
ctx.fireChannelRead(msg);
}
}
protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
try {
//1、判断Bytebuf是否还有数据可读
while (in.isReadable()) {
int outSize = out.size();
int oldInputLength = in.readableBytes();
//2、调用实际的解码方法，如果能读到一行数据，会将数据添加到out中
decode(ctx, in, out);
// Check if this handler was removed before continuing the loop.
// If it was removed, it is not safe to continue to operate on the buffer.
//
// See https://github.com/netty/netty/issues/1664
if (ctx.isRemoved()) {
break;
}
//3、如果步骤2没有添加成功，代码无数据可读，跳出循环
if (outSize == out.size()) {
if (oldInputLength == in.readableBytes()) {
break;
} else {
continue;
}
}
//4、如果还是无数据可读，直接抛异常
if (oldInputLength == in.readableBytes()) {
throw new DecoderException(
StringUtil.simpleClassName(getClass()) +
".decode() did not read anything but decoded a message.");
}
if (isSingleDecode()) {
break;
}
}
} catch (DecoderException e) {
throw e;
} catch (Throwable cause) {
throw new DecoderException(cause);
}
}
LineBasedFrameDecoder的Decode方法如下：
protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
//调用实际解码函数，返回结果
Object decoded = decode(ctx, in);
if (decoded != null) {
//添加到返回结果集List中
out.add(decoded);
}
}
protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
//1、找到一行数据的结束标识的位置
final int eol = findEndOfLine(buffer);
if (!discarding) {
if (eol >= 0) {
final ByteBuf frame;
//2、结束位置-上次已经读取的位置=当前一行数据的长度
final int length = eol - buffer.readerIndex();
//3、下次读取需要跳过结束字符 所以将结束字符的长度记录下
final int delimLength = buffer.getByte(eol) == '\r'? 2 : 1;
//4、如果当前一行数据长度已经大于最大定义的可解码数据的长度，代表已经解析结束
if (length > maxLength) {
buffer.readerIndex(eol + delimLength);
fail(ctx, length);
return null;
}
if (stripDelimiter) {
//5、读取当前一行数据到临时缓存区
frame = buffer.readBytes(length);
//6、下次读取需要跳过结束字符
buffer.skipBytes(delimLength);
} else {
frame = buffer.readBytes(length + delimLength);
}
//返回本次读取的一行数据
return frame;
} else {
final int length = buffer.readableBytes();
if (length > maxLength) {
discardedBytes = length;
buffer.readerIndex(buffer.writerIndex());
discarding = true;
if (failFast) {
fail(ctx, "over " + discardedBytes);
}
}
return null;
}
} else {
if (eol >= 0) {
final int length = discardedBytes + eol - buffer.readerIndex();
final int delimLength = buffer.getByte(eol) == '\r'? 2 : 1;
buffer.readerIndex(eol + delimLength);
discardedBytes = 0;
discarding = false;
if (!failFast) {
fail(ctx, length);
}
} else {
discardedBytes = buffer.readableBytes();
buffer.readerIndex(buffer.writerIndex());
}
return null;
}
}
private static int findEndOfLine(final ByteBuf buffer) {
//获得当前ByteBuf写的位置
final int n = buffer.writerIndex();
//从ByteBuf可读位置开始循环，一直遍历到ByteBuf写的位置，如果有\n 或者\r\n,则意味找到一行数据结束的位置 并返回结束位置
for (int i = buffer.readerIndex(); i < n; i ++) {
final byte b = buffer.getByte(i);
if (b == '\n') {
return i;
} else if (b == '\r' && i < n - 1 && buffer.getByte(i + 1) == '\n') {
return i; // \r\n
}
}
return -1; // Not found.
}
由上源码分析可知，LineBasedFrameDecoder通过'\r'或者'\r\n'作为分割界限作为一个完整的报文输出，如果需要其他定制的分隔符作为界限则可以继承ByteToMessageDecoder 重写decode方法实现。
