TCP传输协议是基于数据流传输的,而基于流化的数据是没有界限的,当客户端向服务端发送数据时,可能会把一个完整的数据报文拆分成多个小报文进行发送,也可能将多个报文合并成一个大报文进行发送。
在这样的情况下,有可能会出现图3-1所示的情况。
图3-1 粘包和拆包问题
由于存在拆包/粘包问题,接收方很难界定数据包的边界在哪里,所以可能会读取到不完整的数据导致数据解析出现问题。
下面演示一个拆包粘包问题
public class PackageNettyServer {
public static void main(String[] args) {
EventLoopGroup bossGroup=new NioEventLoopGroup();
EventLoopGroup workGroup=new NioEventLoopGroup();
try{
ServerBootstrap serverBootstrap=new ServerBootstrap();
serverBootstrap.group(bossGroup,workGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new SimpleServerHandler());
}
});
ChannelFuture channelFuture=serverBootstrap.bind(8080).sync(); //绑定端口
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
}
public class SimpleServerHandler extends ChannelInboundHandlerAdapter {
private int count;
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf in=(ByteBuf) msg;
byte[] buffer=new byte[in.readableBytes()]; //长度为可读的字节数
in.readBytes(buffer); //读取到字节数组中
String message=new String (buffer,"UTF-8");
System.out.println("服务端收到的消息内容:"+message+"\n服务端收到的消息数量"+(++count));
ByteBuf resBB= Unpooled.copiedBuffer(UUID.randomUUID().toString(), Charset.forName("utf-8"));
ctx.writeAndFlush(resBB);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();//关闭连接
}
}
public class PackageNettyClient {
public static void main(String[] args) {
EventLoopGroup 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 {
ch.pipeline().addLast(new SimpleClientHandler());
}
});
ChannelFuture channelFuture=bootstrap.connect("localhost",8080).sync();
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
eventLoopGroup.shutdownGracefully();
}
}
}
public class SimpleClientHandler extends ChannelInboundHandlerAdapter {
private int count;
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("客户端和服务端成功建立连接");
//客户端和服务端建立连接后,发送十次消息给服务端
for (int i = 0; i < 10; i++) {
ByteBuf buf= Unpooled.copiedBuffer("客户端消息"+i, Charset.forName("utf-8"));
ctx.writeAndFlush(buf);
}
super.channelActive(ctx);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//接收服务端发过来的消息
System.out.println("接收到服务端返回的信息");
ByteBuf buf=(ByteBuf)msg;
byte[] buffer=new byte[buf.readableBytes()];
buf.readBytes(buffer);
String message=new String(buffer,Charset.forName("utf-8"));
System.out.println("客户端收到的消息内容为:"+message);
System.out.println("客户端收到的消息数量为:"+(++count));
super.channelRead(ctx, msg);
}
}
运行上述案例后,会出现粘包和拆包问题。
如何解决拆包和粘包问题呢?
一般我们会在应用层定义通信协议。其实思想也很简单,就是通信双方约定一个通信报文协议,服务端收到报文之后,按照约定的协议进行解码,从而避免出现粘包和拆包问题。
其实大家把这个问题往深度思考一下就不难发现,之所以在拆包粘包之后导致收到消息端的内容解析出现错误,是因为程序无法识别一个完整消息,也就是不知道如何把拆包之后的消息组合成一个完整消息,以及将粘包的数据按照某个规则拆分形成多个完整消息。所以基于这个角度思考,我们只需要针对消息做一个通信双方约定的识别规则即可。
每个数据报文都需要一个固定的长度,当接收方累计读取到固定长度的报文后,就认为已经获得了一个完整的消息,当发送方的数据小于固定长度时,则需要空位补齐.
如图3-2所示,假设我们固定消息长度是4,那么没有达到长度的报文,需要通过一个空位来补齐,从而使得消息能够形成一个整体。
图3-2
这种方式很简单,但是缺点也很明显,对于没有固定长度的消息,不清楚如何设置长度,而且如果长度设置过大会造成字节浪费,长度太小又会影响消息传输,所以一般情况下不会采用这种方式。
既然没办法通过固定长度来分割消息,那能不能在消息报文中增加一个分割符呢?然后接收方根据特定的分隔符来进行消息拆分。比如我们采用\r\n来进行分割,如图3-3所示。
图3-3
对于特定分隔符的使用场景中,需要注意分隔符和消息体中的字符不要存在冲突,否则会出现消息拆分错误的问题。
基于消息长度+消息内容+分隔符的方式进行数据通信,这个之前大家在Redis中学习过,redis的报文协议定义如下。
*3\r\n$3\r\nSET\r\n$4\r\nname\r\n$3\r\nmic
可以发现消息报文包含三个维度
这种方式在项目中是非常常见的协议,首先通过消息头中的总长度来判断当前一个完整消息所携带的参数个数。然后在消息体中,再通过消息内容长度以及消息体作为一个组合,最后通过\r\n进行分割。服务端收到这个消息后,就可以按照该规则进行解析得到一个完整的命令进行执行。
在Zookeeper中使用了Jute协议,这是zookeeper自定义消息协议,请求协议定义如图3-4所示。
xid用于记录客户端请求发起的先后序号,用来确保单个客户端请求的响应顺序。type代表请求的操作类型,常见的包括创建节点、删除节点和获取节点数据等。
协议的请求体部分是指请求的主体内容部分,包含了请求的所有操作内容。不同的请求类型,其请求体部分的结构是不同的。
图3-4
响应协议定义如图3-5所示。
协议的响应头中的xid和上文中提到的请求头中的xid是一致的,响应中只是将请求中的xid原值返回。zxid代表ZooKeeper服务器上当前最新的事务ID。err则是一个错误码,当请求处理过程中出现异常情况时,会在这个错误码中标识出来。协议的响应体部分是指响应的主体内容部分,包含了响应的所有返回数据。不同的响应类型,其响应体部分的结构是不同的。
图3-5
在Netty中,默认帮我们提供了一些常用的编解码器用来解决拆包粘包的问题。下面简单演示几种解码器的使用。
固定长度解码器FixedLengthFrameDecoder的原理很简单,就是通过构造方法设置一个固定消息大小frameLength,无论接收方一次收到多大的数据,都会严格按照frameLength进行解码。
如果累计读取的长度大小为frameLength的消息,那么解码器会认为已经获取到了一个完整的消息,如果消息长度小于frameLength,那么该解码器会一直等待后续数据包的达到,知道获得指定长度后返回。
使用方法如下,在3.3节中演示的代码的Server端,增加一个FixedLengthFrameDecoder,长度为10。
ServerBootstrap serverBootstrap=new ServerBootstrap();
serverBootstrap.group(bossGroup,workGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline()
.addLast(new FixedLengthFrameDecoder(10)) //增加解码器
.addLast(new SimpleServerHandler());
}
});
特殊分隔符解码器: DelimiterBasedFrameDecoder,它有以下几个属性
比如接收方收到的消息体为
hello\nworld\r\n
此时指定多个分隔符\n
和\r\n
,那么最终会选择最短的分隔符解码,得到如下数据
hello | world |
\n
,那么数据解码的方式如下。当stripDelimiter=false时,解码后得到
hello\n | world\r\n
下面演示一下DelimiterBasedFrameDecoder的用法。
public class DecoderNettyServer {
public static void main(String[] args) {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workGroup = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ByteBuf delimiter= Unpooled.copiedBuffer("&".getBytes());
ch.pipeline()
.addLast(new DelimiterBasedFrameDecoder(10,true,true,delimiter))
.addLast(new PrintServerHandler());
}
});
ChannelFuture channelFuture = serverBootstrap.bind(8080).sync(); //绑定端口
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
}
}
}
定义一个普通的Inbound,打印接收到的数据。
public class PrintServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf=(ByteBuf)msg;
System.out.println("Receive client Msg:"+buf.toString(CharsetUtil.UTF_8));
}
}
Ctrl+]
组合键,进入到一个telnet界面LengthFieldBasedFrameDecoder是长度域解码器,它是解决拆包粘包最常用的解码器,基本上能覆盖大部分基于长度拆包的场景。其中开源的消息中间件RocketMQ就是使用该解码器进行解码的。
首先来说明一下该解码器的核心参数
上面这些参数理解起来比较难,我们通过几个案例来说明一下。
假设存在图3-6所示的由长度和消息内容组成的数据包,其中length表示报文长度,用16进制表示,共占用2个字节,那么该协议对应的编解码器参数设置如下。
图3-6
如果我们希望解码后的结果中只包含消息内容,其他部分不变,如图3-7所示。对应解码器参数组合如下
图3-7
如图3-8所示,如果Length字段中包含Length字段自身的长度以及Content字段所占用的字节数,那么Length的值为0x00d(2+11=13字节),在这种情况下解码器的参数组合如下
图3-8
如图3-9所示,Length字段已经不再是报文的起始位置,Length字段的值是0x000b,表示content字段占11个字节,那么此时解码器的参数配置如下:
图3-9
如图3-10所示,Length字段前后分别有hdr1和hdr2字段,各占据1个字节,所以需要做长度字段的便宜,还需要做lengthAdjustment的修正,相关参数配置如下。
图3-10
比如我们定义如下消息头,客户端通过该消息协议发送数据,服务端收到该消息后需要进行解码
先定义客户端,其中Length部分,可以使用Netty自带的LengthFieldPrepender来实现,它可以计算当前发送消息的二进制字节长度,然后把该长度添加到ByteBuf的缓冲区头中。
public class LengthFieldBasedFrameDecoderClient {
public static void main(String[] args) {
EventLoopGroup workGroup=new NioEventLoopGroup();
Bootstrap b=new Bootstrap();
b.group(workGroup)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline()
//如果协议中的第一个字段为长度字段,
// netty提供了LengthFieldPrepender编码器,
// 它可以计算当前待发送消息的二进制字节长度,将该长度添加到ByteBuf的缓冲区头中
.addLast(new LengthFieldPrepender(2,0,false))
//使用StringEncoder,在通过writeAndFlush时,不需要自己转化成ByteBuf
//StringEncoder会自动做这个事情
.addLast(new StringEncoder())
.addLast(new ChannelInboundHandlerAdapter(){
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ctx.writeAndFlush("i am request!");
ctx.writeAndFlush("i am a another request!");
}
});
}
});
try {
ChannelFuture channelFuture=b.connect("localhost",8080).sync();
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
workGroup.shutdownGracefully();
}
}
}
上述代码运行时,会得到两个报文。
下面是Server端的代码,增加了LengthFieldBasedFrameDecoder解码器,其中有两个参数的值如下
public class LengthFieldBasedFrameDecoderServer {
public static void main(String[] args) {
EventLoopGroup bossGroup=new NioEventLoopGroup();
EventLoopGroup workGroup=new NioEventLoopGroup();
try{
ServerBootstrap serverBootstrap=new ServerBootstrap();
serverBootstrap.group(bossGroup,workGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline()
.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,2,0,2))
.addLast(new StringDecoder())
.addLast(new ChannelInboundHandlerAdapter(){
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("receive message:"+msg);
}
});
}
});
ChannelFuture channelFuture=serverBootstrap.bind(8080).sync(); //绑定端口
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
}
前面我们分析的几个常用解码器,只是帮我们解决了半包和粘包的问题,最终会让接受者收到一个完整有效的请求报文并且封装到ByteBuf中, 而这个报文内容是否有其他的编码方式,比如序列化等,还需要单独进行解析处理。
另外,很多的中间件,都会定义自己的报文协议,这些报文协议除了本身解决粘包半包问题以外,还会传递一些其他有意义的数据,比如zookeeper的jute、dubbo框架的dubbo协议等。
版权声明:本博客所有文章除特别声明外,均采用 CC BY-NC-SA 4.0 许可协议。转载请注明来自 Mic带你学架构
!
如果本篇文章对您有帮助,还请帮忙点个关注和赞,您的坚持是我不断创作的动力。欢迎关注「跟着Mic学架构」公众号公众号获取更多技术干货!
版权说明 : 本文为转载文章, 版权归原作者所有 版权申明
原文链接 : https://www.cnblogs.com/mic112/p/15551997.html
内容来源于网络,如有侵权,请联系作者删除!