前言
TCP是面向连接的,服务端和客户端通过socket进行数据传输,发送端为了更有效的发送数据,通常会使用Nagle算法把多个数据块合并成一个大的数据块,这样做虽然提高了效率,但是接收端就很难识别完整的数据包了(TCP无消息保护边界),可能会出现粘包拆包的问题。
粘包拆包理解
下面我用一个图来带大家理解什么是粘包和拆包 解释一下
- 第一次传输没有问题,数据1和数据2没有粘合,也没有拆分
- 第二次传输,数据1和数据2粘在一起传输了,出现了粘包
- 第三次传输,数据2被分为了2部分,数据2_1 第一份和数据1粘在一起,数据2_2第二份单独传输,这里即出现了拆包也出现了粘包
粘包拆包代码演示
这里写一个简单案例来演示粘包拆包,客户端发送10个数据包,观察服务端是否做了10次读取,如果不是,就出现粘包或者拆包的情况,这里我们使用byte类型来传输案例如下。
第一步:编写Netty服务端
public static void main(String[] args) {
NioEventLoopGroup bossGroup = new NioEventLoopGroup();
NioEventLoopGroup workGroup = new NioEventLoopGroup();
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workGroup);
bootstrap.channel(NioServerSocketChannel.class);
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new ServerHandler());
}
});
try {
ChannelFuture sync = bootstrap.bind(3000).sync();
sync.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
第二步:编写服务端handler
public class ServerHandler extends SimpleChannelInboundHandler<ByteBuf> {
private int num = 0;
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
System.out.println("接收消息,次数 = "+ num++);
byte[] bytes = new byte[msg.readableBytes()];
msg.readBytes(bytes);
System.out.println(new String(bytes, CharsetUtil.UTF_8));
}
}
这里定义了一个num来记录服务端数据读取次数。
第三步:定义Netty客户端
public static void main(String[] args) {
NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(eventLoopGroup);
bootstrap.channel(NioSocketChannel.class);
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new ClientHandler());
}
});
ChannelFuture sync = null;
try {
sync = bootstrap.connect("127.0.0.1", 3000).sync();
sync.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
eventLoopGroup.shutdownGracefully();
}
}
第四步:定义客户端的Handler
public class ClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
for (int i = 0; i < 10; i++) {
ctx.writeAndFlush(Unpooled.copiedBuffer("数据块"+i+";", CharsetUtil.UTF_8));
}
}
}
这里循环了10次,我发送了10个数据块
第五步:测试,启动服务端和客户端。观察控制台 问题比较明显,客户端发送了10次数据,服务端做了5次接收,第3次4次5次都出现了粘包的情况。
定义编码器解决粘包拆包问题
要解决粘包拆包的问题就要明确数据边界,尴尬的是面向流的通信是没有消息保护边界的。所以我们需要自定义传输协议来确定消息的边界,说的再直白一点就是我们如果能够明确服务端每次读取消息的长度,那就不会出现粘包拆包问题了。
如果要做到该效果,那么就需要自定义消息协议和编码解码器,我们先来处理客户端。
第一步:定义协议 , 指定消息长度和内容
public class MsgProtocol {
private int len;
private byte[] data;
public MsgProtocol(int len , byte[] data){
this.len = len;
this.data = data;
}
public MsgProtocol(){}
public int getLen() {
return len;
}
public void setLen(int len) {
this.len = len;
}
public byte[] getData() {
return data;
}
public void setData(byte[] data) {
this.data = data;
}
}
第二步:客户端的handler发送MsgProtocol对象
public class ClientHandler extends SimpleChannelInboundHandler<MsgProtocol> {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
for (int i = 0; i < 10; i++) {
String data = "数据块"+i;
byte[] bytes = data.getBytes(CharsetUtil.UTF_8);
int len = bytes.length;
ctx.writeAndFlush(new MsgProtocol(len,bytes));
}
}
}
第三步:继承MessageToByteEncoder,自定义编码器 ,把消息的长度和内容写出去
public class MessageEncoder extends MessageToByteEncoder<MsgProtocol> {
@Override
protected void encode(ChannelHandlerContext ctx, MsgProtocol msg, ByteBuf out) throws Exception {
out.writeInt(msg.getLen());
out.writeBytes(msg.getData());
}
}
第四步:客户端指定编码器
public static void main(String[] args) {
NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(eventLoopGroup);
bootstrap.channel(NioSocketChannel.class);
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new MessageEncoder());
pipeline.addLast(new ClientHandler());
}
});
ChannelFuture sync = null;
try {
sync = bootstrap.connect("127.0.0.1", 3000).sync();
sync.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
eventLoopGroup.shutdownGracefully();
}
}
客户端的工作完成了,接下来我们处理服务端
第一步:编写解码器,需要把byte数据封装成MsgProtocol
public class MessageDecoder extends ReplayingDecoder<Void> {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
int len = in.readInt();
byte[] bytes = new byte[len];
in.readBytes(bytes);
out.add(new MsgProtocol(len,bytes));
}
}
ReplayingDecoder就是对ByteToMessageDecoder的 扩展和简化
第二步:服务端handler,这里接收的是MsgProtocol消息对象
public class ServerHandler extends SimpleChannelInboundHandler<MsgProtocol> {
private int num = 0;
@Override
protected void channelRead0(ChannelHandlerContext ctx, MsgProtocol msg) throws Exception {
System.out.println("接收消息,次数 = "+ num++);
System.out.println(new String(msg.getData(), CharsetUtil.UTF_8));
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.channel().close();
}
}
第三步:服务端指定解码器
public static void main(String[] args) {
NioEventLoopGroup bossGroup = new NioEventLoopGroup();
NioEventLoopGroup workGroup = new NioEventLoopGroup();
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workGroup);
bootstrap.channel(NioServerSocketChannel.class);
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new MessageDecoder());
pipeline.addLast(new ServerHandler());
}
});
try {
ChannelFuture sync = bootstrap.bind(3000).sync();
sync.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
到这里服务端编写完成,接下来依次启动服务端和客户端进行测试,效果如下 可以看到,客户端发送了10次,服务器接收了10次,没有出现粘包拆包的情况了。所以问题的关键就是服务端解码器中需要明确消息的长度,就能够明确每次消息读取的边界,就不会出问题了。
好了文章结束,喜欢就给个好评吧
|