一、是什么？
客户端通过socket给服务端发送数据，为了传输更有效率，会将多次间隔较小的且数据量小的数据，通过nagle算法，合并成一个大的数据块，然后进行封包。这样做提高了效率，缺点就是你发送到服务端的数据，服务端不知道是不是完整的，不知道哪几小块数据拼起来才是原来的数据。举个例子：客户端要发送原信息是A和B两个数据包，服务端接收到之后，可能出现如下情况：
- 拆包：读取了A数据包的一部分，A的另一部分和B数据包一起读取了；
由于TCP是没有消息保护边界的，也就是上面的消息，没有边界，服务端并不知道hello的o是一个边界，hello是一个单词，所以我们就得中服务端处理边界问题。这也就是粘包拆包问题。
二、Netty中粘拆包如何解决
- 使用自定义协议 + 编解码器来解决。说人话就是：服务端你不是不知道消息的长度吗？那我就让客户端发送的消息封装成一个对象，对象包括消息长度和消息内容，服务端读取的时候通过对象就可以拿到每次读取的长度了。
下面看具体案例：
- 封装消息对象 MessageProtocol.java：
@Data
public class MessageProtocol {
private int len; // 长度
private byte[] content; // 发送的内容
}
public class MessageDecoder extends ReplayingDecoder<Void> {
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
System.out.println("MessageDecoder.decode 被调用");
// 将byte转成MessageProtocol对象
MessageProtocol msg = new MessageProtocol();
int len = in.readInt();
byte[] content = new byte[len];
in.readBytes(content);
msg.setContent(content);
msg.setLen(len);
// 放入到out中传递给下一个handler处理
out.add(msg);
}
}
public class MessageEncoder extends MessageToByteEncoder<MessageProtocol> {
@Override
protected void encode(ChannelHandlerContext ctx, MessageProtocol msg, ByteBuf out) throws Exception {
System.out.println("MessageEncoder.encode被调用");
out.writeInt(msg.getLen());
out.writeBytes(msg.getContent());
}
}
- 客户端 --- NettyClient.java：
public class NettyClient {
public static void main(String[] args) throws Exception {
// 1. 创建事件循环组
EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
try {
// 2. 创建启动对象
Bootstrap bootstrap = new Bootstrap();
// 3. 设置相关参数
bootstrap.group(eventLoopGroup) // 设置线程组
.channel(NioSocketChannel.class) // 设置通道
.handler(new NettyClientInitializer());
// 4. 连接服务端
ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6666).sync();
// 5. 监听通道关闭
channelFuture.channel().closeFuture().sync();
} finally {
eventLoopGroup.shutdownGracefully();
}
}
}
- 客户端 --- NettyClientInitializer.java：
public class NettyClientInitializer extends ChannelInitializer<SocketChannel>{
@Override
protected void initChannel(SocketChannel sc) throws Exception {
ChannelPipeline pipeline = sc.pipeline();
pipeline.addLast(new MessageEncoder());
pipeline.addLast(new MessageDecoder());
pipeline.addLast(new NettyClientHandler());
}
}
- 客户端 --- NettyClientHandler.java：
public class NettyClientHandler extends SimpleChannelInboundHandler<MessageProtocol>{
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
// 发送10条数据
for (int i=0; i<5; i++) {
String msg = "hello " + i;
byte[] bys = msg.getBytes("utf-8");
int len = msg.getBytes("utf-8").length;
// 创建协议包
MessageProtocol message = new MessageProtocol();
message.setLen(len);
message.setContent(bys);
// 发送
ctx.writeAndFlush(message);
}
}
@Override
protected void channelRead0(ChannelHandlerContext ch, MessageProtocol msg) throws Exception {
int len = msg.getLen();
byte[] bys = msg.getContent();
System.out.println("客户端收到消息：长度 = " + len + ", 内容 = " + new String(bys, Charset.forName("utf-8")));
}
}
public class NettyServer {
public static void main(String[] args) throws Exception {
// 1. 创建boss group (boss group和work group含有的子线程数默认是cpu数 * 2)
EventLoopGroup bossGroup = new NioEventLoopGroup();
// 2. 创建work group
EventLoopGroup workGroup = new NioEventLoopGroup();
try {
// 3. 创建服务端启动对象
ServerBootstrap bootstrap = new ServerBootstrap();
// 4. 配置启动参数
bootstrap.group(bossGroup, workGroup) // 设置两个线程组
.channel(NioServerSocketChannel.class) // 使用NioSocketChannel 作为服务器的通道
.childHandler(new NettyServerInitializer());
// 5. 启动服务器并绑定端口
ChannelFuture cf = bootstrap.bind(6666).sync();
// 6. 对关闭通道进行监听
cf.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
}
- 服务端 NettyServerInitializer.java：
public class NettyServerInitializer extends ChannelInitializer<SocketChannel>{
@Override
protected void initChannel(SocketChannel sc) throws Exception {
sc.pipeline().addLast(new MessageDecoder());
sc.pipeline().addLast(new MessageEncoder());
sc.pipeline().addLast(new NettyServerHandler());
}
}
- 服务端 NettyServerHandler.java：
public class NettyServerHandler extends SimpleChannelInboundHandler<MessageProtocol> {
private int count;
@Override
protected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {
// 接收数据并处理
int len = msg.getLen();
byte[] bys = msg.getContent();
System.out.println("服务端第" + (++count) + "次收到消息：长度 = " + len + ", 内容 = " + new String(bys, Charset.forName("utf-8")));
// 给客户端回复消息
String responseContent = UUID.randomUUID().toString();
byte[] rbys = responseContent.getBytes("utf-8");
int rlen = responseContent.getBytes("utf-8").length;
MessageProtocol rmsg = new MessageProtocol();
rmsg.setContent(rbys);
rmsg.setLen(rlen);
ctx.writeAndFlush(rmsg);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println(cause.getMessage());
ctx.close();
}
}
怎么样，你学废了吗？
