Netty权威指南之TCP粘包和拆包
本章节需要学习的内容如下:
1、TCP粘包和拆包基础知识。
2、TCP粘包和拆包异常案例
3、Netty 解决半包问题。
第一节:TCP粘包和拆包基础知识
TCP粘包和拆包问题产生原因?
TCP协议本质上就是个“流”协议,就是没有界限的一串数据。大家可以想象河里的水,是连成一片,期间没有分界线。TCP底层并不了解上层业务数据的具体含义,它会根据TCP缓冲区的实际情况进行包的划分,所以在业务上认为,一个完整的包可能会被TCP拆分成多个包进行发送,也有可能把大大小小的数据包封装成一个大的数据包进行发送,这就是所谓的TCP粘包和拆包问题。
TCP 粘包和拆包通信模型图:
假设客户端分别发送两个数据包D1和D2给服务端,由于服务端一次读取的字节数是不确定的,故可能存在以下四种情况:
1、服务端分两次读取到两个独立的数据包,分别是D1和D2。没有粘包和拆包。
2、服务端一次接受两个数据包,D1和D2粘合在一起,称之为粘包。
3、服务端分两次读取到两个数据包,第一次读取到了完整D1包和D2包的部分内容,第二次读取到D2包剩余的内容,称之为拆包。
4、服务端分两次读取到两个数据包。第一此读取到的D1包部分内容,第二次读取D1剩余部分和完整D2包。
如果此时服务端TCP接受划窗非常小,而数据包D1和D2都比较大,很可能发送第五种情况:及服务端多次才能将D1和D2接受完全,期间发生多次拆包。
TCP粘包和拆包发送的原因:
问题产生的原因有三个,分别如下:
1、应用程序write写入的字节大小大于套接字发送缓存大小
2、进行MSS大小的TCP分段
3、以太网帧的playload大于MTU进行IP分片。
粘包问题解决策略:
由于底层TCP无法理解上层的业务数据,所以底层是无法保证数据包不被拆分和重组的,这个问题只能通过上层的应用协议栈设计来解决,根据业界主流协议的解决方案,可以归类如下:
1、消息定长,列如:每个报文的大小固定长度为200个字节,如果不够,空位补空格。
2、在包尾增加回车换行符进行分割,列如:FTP协议
3、将消息分为消息头和消息体,消息头中包含表示消息的总长度(或消息体长度)的字段,通常的设计思路为消息头的第一个字段用int32来表示消息的总长度。
4、更复杂的应用协议。
介绍完TCP的粘包和拆包的基础知识,下面我们通过实践教程看看如何使用Netty提供的半包解码器来解决TCP粘包和拆包问题。
第二节:TCP 粘包和拆包异常案例
在前面的NettyServer服务器代码中,我们多次强调没有考虑半包问题,在功能测试时往往没有什么问题,但是一旦压力上来,或者发送大报文之后,就存在粘包和拆包的问题。如果代码没有考虑,往往会出现解码错位会错误,导致程序不能正常的工作。我们已下面的代码为例,模拟故障现场,然后看看如何正确使用Netty半包解码器,来解决TCP的粘包和拆包问题。
Netty 服务端代码:
NettyServer.java
package com.nio.server;
import com.nio.handler.NettyServerHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
/**
* Created by vixuan-008 on 2015/6/19.
*/
public class NettyServer {
public static void main(String[] args)throws Exception{
int port=17666;
new NettyServer().bind(port);
}
public void bind(int port)throws Exception{
//配置服务端的NIO线程池
EventLoopGroup bossGroup=new NioEventLoopGroup();
EventLoopGroup workGroup=new NioEventLoopGroup();
try{
ServerBootstrap b=new ServerBootstrap();
b.group(bossGroup,workGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG,1024).childHandler(new NettyServerHandler());
//绑定端口,等待同步成功
ChannelFuture f=b.bind(port).sync();
//等待服务端关闭监听端口
f.channel().closeFuture().sync();
}finally {
//释放线程池资源
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
//添加内部类
private class ChildChannelHandler extends ChannelInitializer<SocketChannel>{
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new NettyServerHandler());
}
}
}
NettyServerHandler.java
package com.nio.handler;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
/**
* Created by vixuan-008 on 2015/6/19.
*/
public class NettyServerHandler extends ChannelHandlerAdapter {
private int counter;
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf=(ByteBuf)msg;
byte[] req=new byte[buf.readableBytes()];
buf.readBytes(req);
String body=new String(req,"UTF-8").substring(0,req.length - System.getProperty("line.separator").length());
System.out.println("the server receiver data is:"+body+"; the counter is:"+ ++counter);
String currentTime="time".equalsIgnoreCase(body) ? new java.util.Date(System.currentTimeMillis()).toString():"no zuo no die";
currentTime=currentTime+ System.getProperty("line.separator");
ByteBuf resp= Unpooled.copiedBuffer(currentTime.getBytes());
ctx.writeAndFlush(resp);
}
}
Netty客户端代码:
NettyClient.java
package com.nio.client;
import com.nio.handler.NettyClientHandler;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
/**
* Created by vixuan-008 on 2015/6/19.
*/
public class NettyClient {
public static void main(String[] args)throws Exception{
int port=17666;
new NettyClient().connect(port,"127.0.0.1");
}
public void connect(int port,String host)throws Exception{
//配置客户端NIO线程池
EventLoopGroup workGroup=new NioEventLoopGroup();
try{
io.netty.bootstrap.Bootstrap b=new io.netty.bootstrap.Bootstrap();
b.group(workGroup).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY,true).handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new NettyClientHandler());
}
});
//发起异步连接操作
ChannelFuture f=b.connect(host,port).sync();
//等待客户端链路关闭
f.channel().closeFuture().sync();
}finally {
//释放NIO 线程组
workGroup.shutdownGracefully();
}
}
}
NettyClientHandler.java
package com.nio.handler;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
/**
* Created by vixuan-008 on 2015/6/19.
*/
public class NettyClientHandler extends ChannelHandlerAdapter {
private int counter;
private byte[] req;
public NettyClientHandler() {
req=("time"+System.getProperty("line.separator")).getBytes();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
@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 buf=(ByteBuf)msg;
byte[] req=new byte[buf.readableBytes()];
buf.readBytes(req);
String body=new String(req,"UTF-8");
System.out.println("this client receiver data is:"+body+"; this is counter is"+ ++counter);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
}
}
NettyServer 服务端执行效果截图:
按照设计初衷,服务端应该接收到100条消息,但实际上服务端只接收到了一条请求信息。这说明发生了TCP粘包情况。
NettyClient 客户端执行效果截图:
按照设计初衷,客户端应该接收到100条当前系统时间,但实际上只收到一条数据信息。这个不难理解,客户端发送的请求只有一次,所以服务器应答也只有一次。
由于上面的程序没有考虑到TCP 粘包和拆包的问题,所以导致我们的程序无法正常运行。
第三节:Netty解决半包问题
为了解决TCP的粘包和拆包导致的半包读写问题,Netty 提供多种编解码器用于处理半包,只有熟练使用这些类库,TCP的粘包和拆包就变得很容易哒。你甚至不需要关心它们,这也是其他框架和Java NIO 原生框架无法匹敌的。
Netty服务端代码
NettuServer.java
package com.nio.server;
import com.nio.handler.NettyServerHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
/**
* Created by vixuan-008 on 2015/6/19.
*/
public class NettyServer {
public static void main(String[] args)throws Exception{
int port=17666;
new NettyServer().bind(port);
}
public void bind(int port)throws Exception{
//配置服务端的NIO线程池
EventLoopGroup bossGroup=new NioEventLoopGroup();
EventLoopGroup workGroup=new NioEventLoopGroup();
try{
ServerBootstrap b=new ServerBootstrap();
b.group(bossGroup,workGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG,1024).childHandler(new ChildChannelHandler());
//绑定端口,等待同步成功
ChannelFuture f=b.bind(port).sync();
//等待服务端关闭监听端口
f.channel().closeFuture().sync();
}finally {
//释放线程池资源
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
//添加内部类
private class ChildChannelHandler extends 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 NettyServerHandler());
}
}
}
NettyServerHandler.java
package com.nio.handler;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
/**
* Created by vixuan-008 on 2015/6/19.
*/
public class NettyServerHandler extends ChannelHandlerAdapter {
private int counter;
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String body=(String)msg;
System.out.println("the server receiver data is:"+body+"; the counter is:"+ ++counter);
String currentTime="time".equalsIgnoreCase(body) ? new java.util.Date(System.currentTimeMillis()).toString():"no zuo no die";
currentTime=currentTime+ System.getProperty("line.separator");
ByteBuf resp= Unpooled.copiedBuffer(currentTime.getBytes());
ctx.writeAndFlush(resp);
}
}
Netty客户端代码
NettyClient.java
package com.nio.client;
import com.nio.handler.NettyClientHandler;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
/**
* Created by vixuan-008 on 2015/6/19.
*/
public class NettyClient {
public static void main(String[] args)throws Exception{
int port=17666;
new NettyClient().connect(port,"127.0.0.1");
}
public void connect(int port,String host)throws Exception{
//配置客户端NIO线程池
EventLoopGroup workGroup=new NioEventLoopGroup();
try{
io.netty.bootstrap.Bootstrap b=new io.netty.bootstrap.Bootstrap();
b.group(workGroup).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 NettyClientHandler());
}
});
//发起异步连接操作
ChannelFuture f=b.connect(host,port).sync();
//等待客户端链路关闭
f.channel().closeFuture().sync();
}finally {
//释放NIO 线程组
workGroup.shutdownGracefully();
}
}
}
NettyClientHandler.java
package com.nio.handler;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
/**
* Created by vixuan-008 on 2015/6/19.
*/
public class NettyClientHandler extends ChannelHandlerAdapter {
private int counter;
private byte[] req;
public NettyClientHandler() {
req=("time"+System.getProperty("line.separator")).getBytes();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
@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 {
String body=(String)msg;
System.out.println("this client receiver data is:"+body+"; this is counter is"+ ++counter);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
}
}
NettyServer 服务端运行截图:
NettyClient 客户端运行截图:
程序的运行结果,完全符合预期。说明通过使用LineBaseFrameDecode和StringDecoder成功解决TCP粘包和拆包导致的半包问题。对于使用者来说,只有将支持半包解的handler添加到ChannlePipline中即可,不需要编写而外的代码,用户使用起来很简单。
1.本站遵循行业规范,任何转载的稿件都会明确标注作者和来源;2.本站的原创文章,请转载时务必注明文章作者和来源,不尊重原创的行为我们将追究责任;3.作者投稿可能会经我们编辑修改或补充。