netty框架tcp协议粘包半包序列化解决方案

2023-11-21 07:20
文章标签 解决方案 协议 框架 tcp 序列化 netty 粘包 半包

本文主要是介绍netty框架tcp协议粘包半包序列化解决方案，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

主要依赖

<dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.39.Final</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.16.18</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.76</version></dependency>

粘包半包解决

粘包半包问题，这里使用LengthFieldBasedFrameDecoder来作为解决方案
在发送消息前，先约定用定长字节表示接下来数据的长度
LengthFieldBasedFrameDecoder()构造方法参数：

  1. maxFrameLength:数据帧最大长度
  2. lengthFieldoffset:长度字段偏移量
  3. lengthFieldLength:长度字段长度(字节)
  4. lengthAdjustment:长度字段后第几个字节是正文内容
  5. initialBytesToStrip：从帧首去除字节数

序列化(二进制序列化)

java序列化

注意事项：
序列化对象需要实现Serializable接口(java.io.Serializable)
代码：

public class MessageCodecJava extends MessageToMessageCodec<ByteBuf, Message> {@Overrideprotected void encode(ChannelHandlerContext ctx, Message msg, List<Object> outList) throws Exception {ByteBuf out = ctx.alloc().buffer();//获取内容ByteArrayOutputStream bos = new ByteArrayOutputStream();ObjectOutputStream oos = new ObjectOutputStream(bos);oos.writeObject(msg);byte[] bytes = bos.toByteArray();//长度,我们设定为4个字节out.writeInt(bytes.length);// 写入内容out.writeBytes(bytes);outList.add(out);}@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {int length = in.readInt();byte[] bytes = new byte[length];in.readBytes(bytes, 0, length);ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));Message message = (Message) ois.readObject();System.out.println(message);out.add(message);}
}

fastjson序列化

注意事项： 不需要实现Serializable接口(java.io.Serializable)
代码：

public class MessageCodecJson extends MessageToMessageCodec<ByteBuf, Message> {@Overrideprotected void encode(ChannelHandlerContext ctx, Message msg, List<Object> list) throws Exception {ByteBuf out = ctx.alloc().buffer();//获取内容字节byte[] jsonBytes = JSON.toJSONString(msg, SerializerFeature.WriteClassName).getBytes(StandardCharsets.UTF_8);// 获取内容长度out.writeInt(jsonBytes.length);//  写入内容out.writeBytes(jsonBytes);list.add(out);}@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> list) throws Exception {//获取内容长度int length = in.readInt();//将二进制反序列化为对象byte[] bytes = new byte[length];in.readBytes(bytes, 0, length);Message msg = JSON.parseObject(new String(bytes, StandardCharsets.UTF_8),Message.class);System.out.println(msg);list.add(msg);}
}

粘包半包序列化解决举例：

客户端代码：

package com.haust.blog;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;public class MyClient {public static void main(String[] args) {NioEventLoopGroup group = new NioEventLoopGroup();try {Bootstrap bootstrap = new Bootstrap();bootstrap.group(group).channel(NioSocketChannel.class).handler(new MyClientInitializer());ChannelFuture channelFuture = bootstrap.connect("localhost", 7000).sync();channelFuture.channel().closeFuture().sync();}catch (Exception e){e.printStackTrace();}finally {group.shutdownGracefully();}}
}

服务器端代码：

package com.haust.blog;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;public class MyServer {public static void main(String[] args) {NioEventLoopGroup boosGroup = new NioEventLoopGroup();NioEventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(boosGroup,workerGroup).option(ChannelOption.SO_BACKLOG,128).childOption(ChannelOption.SO_KEEPALIVE,true).channel(NioServerSocketChannel.class).childHandler(new MyServerInitializer());ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();channelFuture.channel().closeFuture().sync();}catch (Exception e){e.printStackTrace();}finally {boosGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}

POJO类
为提高编解码的扩展性，我们单独定义一个父类，然后让其他POJO类集成此父类

package com.haust.blog.pojo;import java.io.Serializable;public class Message implements Serializable {
}
package com.haust.blog.pojo;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class JsonBeanMessage extends Message {private int id;private String name;
}
package com.haust.blog.pojo;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class BinaryBeanMessage extends Message{private int id;private String name;private String sex;
}

客户端初始化类

package com.haust.blog;import com.haust.blog.handler.MyClientHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;public class MyClientInitializer extends ChannelInitializer<SocketChannel> {protected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline pipeline = socketChannel.pipeline();pipeline.addLast(new LengthFieldBasedFrameDecoder(1024,0,4,0,0));pipeline.addLast(new MessageCodecJson());pipeline.addLast(new MyClientHandler());}
}

服务器端初始化类

package com.haust.blog;import com.haust.blog.handler.MyServerHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;public class MyServerInitializer extends ChannelInitializer<SocketChannel> {protected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline pipeline = socketChannel.pipeline();pipeline.addLast(new LengthFieldBasedFrameDecoder(1024,0,4,0,0));
//        pipeline.addLast(new MessageCodecJava());pipeline.addLast(new MessageCodecJson());pipeline.addLast(new MyServerHandler());}
}

客户端处理类

package com.haust.blog.handler;import com.haust.blog.pojo.BinaryBeanMessage;
import com.haust.blog.pojo.JsonBeanMessage;
import com.haust.blog.pojo.Message;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;public class MyClientHandler extends SimpleChannelInboundHandler<Message> {@Overrideprotected void channelRead0(ChannelHandlerContext channelHandlerContext, Message message) throws Exception {System.out.println("客户端收到了服务器发来的数据:"+message);}@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {
//        BinaryBeanMessage message = new BinaryBeanMessage(1, "张三", "男");JsonBeanMessage message = new JsonBeanMessage(1, "李四");ctx.writeAndFlush(message);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.out.println(cause.getMessage());ctx.close();}
}

服务器端处理类

package com.haust.blog.handler;import com.haust.blog.pojo.BinaryBeanMessage;
import com.haust.blog.pojo.Message;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;public class MyServerHandler extends SimpleChannelInboundHandler<Message> {@Overrideprotected void channelRead0(ChannelHandlerContext channelHandlerContext, Message message) throws Exception {System.out.println("服务器收到了客户端发来的数据:"+message);}@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {BinaryBeanMessage message = new BinaryBeanMessage(1, "张三", "男");ctx.writeAndFlush(message);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.out.println(cause.getMessage());ctx.close();}
}

java序列化类

package com.haust.blog;import com.haust.blog.pojo.Message;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageCodec;
import lombok.extern.slf4j.Slf4j;import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.List;@Slf4j
@ChannelHandler.Sharable
/*** 必须和 LengthFieldBasedFrameDecoder 一起使用，确保接到的 ByteBuf 消息是完整的*/
public class MessageCodecJava extends MessageToMessageCodec<ByteBuf, Message> {@Overrideprotected void encode(ChannelHandlerContext ctx, Message msg, List<Object> outList) throws Exception {ByteBuf out = ctx.alloc().buffer();//获取内容的字节数组ByteArrayOutputStream bos = new ByteArrayOutputStream();ObjectOutputStream oos = new ObjectOutputStream(bos);oos.writeObject(msg);byte[] bytes = bos.toByteArray();//长度out.writeInt(bytes.length);//写入内容out.writeBytes(bytes);outList.add(out);}@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {int length = in.readInt();byte[] bytes = new byte[length];in.readBytes(bytes, 0, length);ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));Message message = (Message) ois.readObject();System.out.println(message);out.add(message);}
}

fastjson序列化类

package com.haust.blog;import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.haust.blog.pojo.Message;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageCodec;
import java.nio.charset.StandardCharsets;
import java.util.List;@ChannelHandler.Sharable
public class MessageCodecJson extends MessageToMessageCodec<ByteBuf, Message> {@Overrideprotected void encode(ChannelHandlerContext ctx, Message msg, List<Object> list) throws Exception {ByteBuf out = ctx.alloc().buffer();//获取内容字节byte[] jsonBytes = JSON.toJSONString(msg, SerializerFeature.WriteClassName).getBytes(StandardCharsets.UTF_8);// 获取内容长度out.writeInt(jsonBytes.length);//  写入内容out.writeBytes(jsonBytes);list.add(out);}@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> list) throws Exception {//获取内容长度int length = in.readInt();//将二进制反序列化为对象byte[] bytes = new byte[length];in.readBytes(bytes, 0, length);Message msg = JSON.parseObject(new String(bytes, StandardCharsets.UTF_8),Message.class);System.out.println(msg);list.add(msg);}
}

包结构：
在这里插入图片描述
结果：
在这里插入图片描述
在这里插入图片描述

总结：

  1. 如何解决粘包半包？思路：自定义协议，在对内容进行编码时，将内容的长度也作为信息发送过去，使解码时可以确定信息边界正确解码，从而避免粘包半包。
  2. 在已经确定好编解码规则(自定义协议)的情况下，为何还会需要LengthFieldBasedFrameDecoder类解决粘包半包问题？当发送数据过大时，导致一组消息多次发送(一个完整的消息，经过发送多次才发送结束)，若直接按照解码规则进行解码就会导致半包问题出现，若我们使用LengthFieldBasedFrameDecoder在解码前进行预处理，LengthFieldBasedFrameDecoder会等到接收到完整的消息之后才会传给自定义decoder进行解码，从而避免了半包问题等。

代码举例：

package com.haust.blog;import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.haust.blog.pojo.JsonBeanMessage;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.logging.LoggingHandler;import java.nio.charset.StandardCharsets;public class test {public static void main(String[] args) throws Exception {EmbeddedChannel channel = new EmbeddedChannel(new LoggingHandler(),new LengthFieldBasedFrameDecoder(1024, 0, 4, 0, 0),new MessageCodecJson());//encodeJsonBeanMessage json = new JsonBeanMessage(1, "张三");channel.writeOutbound(json);//decodeByteBuf buf = ByteBufAllocator.DEFAULT.buffer();byte[] jsonBytes = JSON.toJSONString(json, SerializerFeature.WriteClassName).getBytes(StandardCharsets.UTF_8);// 获取内容长度buf.writeInt(jsonBytes.length);buf.writeBytes(jsonBytes);ByteBuf s1 = buf.slice(0, 50);ByteBuf s2 = buf.slice(50, buf.readableBytes() - 50);s1.retain(); // 引用计数 2channel.writeInbound(s1);channel.writeInbound(s2);}
}

结果：传过来74B信息，第一次只传过来50B信息，此时LengthFieldBasedFrameDecoder进行拦截（若此时直接将50B信息传给解码器就会出现半包），直到另外24B信息到来(此时信息已经完整)，LengthFieldBasedFrameDecoder才将完整消息传给decoder(自定义)进行解码操作

17:34:28 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] REGISTERED
17:34:28 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] ACTIVE
17:34:28 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] WRITE: 74B+-------------------------------------------------+|  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 46 7b 22 40 74 79 70 65 22 3a 22 63 6f |...F{"@type":"co|
|00000010| 6d 2e 68 61 75 73 74 2e 62 6c 6f 67 2e 70 6f 6a |m.haust.blog.poj|
|00000020| 6f 2e 4a 73 6f 6e 42 65 61 6e 4d 65 73 73 61 67 |o.JsonBeanMessag|
|00000030| 65 22 2c 22 69 64 22 3a 31 2c 22 6e 61 6d 65 22 |e","id":1,"name"|
|00000040| 3a 22 e5 bc a0 e4 b8 89 22 7d                   |:"......"}      |
+--------+-------------------------------------------------+----------------+
17:34:28 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] FLUSH
17:34:28 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] READ: 50B+-------------------------------------------------+|  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 00 00 46 7b 22 40 74 79 70 65 22 3a 22 63 6f |...F{"@type":"co|
|00000010| 6d 2e 68 61 75 73 74 2e 62 6c 6f 67 2e 70 6f 6a |m.haust.blog.poj|
|00000020| 6f 2e 4a 73 6f 6e 42 65 61 6e 4d 65 73 73 61 67 |o.JsonBeanMessag|
|00000030| 65 22                                           |e"              |
+--------+-------------------------------------------------+----------------+
17:34:28 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] READ COMPLETE
17:34:28 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] READ: 24B+-------------------------------------------------+|  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 2c 22 69 64 22 3a 31 2c 22 6e 61 6d 65 22 3a 22 |,"id":1,"name":"|
|00000010| e5 bc a0 e4 b8 89 22 7d                         |......"}        |
+--------+-------------------------------------------------+----------------+
JsonBeanMessage(id=1, name=张三)
17:34:28 [DEBUG] [main] i.n.h.l.LoggingHandler - [id: 0xembedded, L:embedded - R:embedded] READ COMPLETEProcess finished with exit code 0

  1. 自定义编码器加@ChannelHandler.Sharable注解时在这里插入图片描述
    需要extends MessageToMessageCodec<ByteBuf, Message>
    Message是当接收到Message及其子类时进入此handler进行处理

  2. 为什么需要定义父类Message，然后其他POJO类继承？提高了可扩展性，如果父类message实现了序列化接口，子类就无需再显示的实现序列化接口。其次在编解码(handler)处理时指定了泛型，那么泛型类型及其子类均可以进入handler进行处理。

这篇关于netty框架tcp协议粘包半包序列化解决方案的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/400758。

相关文章

java.sql.SQLTransientConnectionException连接超时异常原因及解决方案

java.sql.SQLTransientConnectionException连接超时异常原因及解决方案

《java.sql.SQLTransientConnectionException连接超时异常原因及解决方案》：本文主要介绍java.sql.SQLTransientConnectionExcep... 目录一、引言二、异常信息分析三、可能的原因3.1 连接池配置不合理3.2 数据库负载过高3.3 连接泄漏
阅读更多...
HTTP 与 SpringBoot 参数提交与接收协议方式

HTTP 与 SpringBoot 参数提交与接收协议方式

《HTTP与SpringBoot参数提交与接收协议方式》HTTP参数提交方式包括URL查询、表单、JSON/XML、路径变量、头部、Cookie、GraphQL、WebSocket和SSE,依据... 目录HTTP 协议支持多种参数提交方式，主要取决于请求方法(Method)和内容类型(Content-Ty
阅读更多...
GSON框架下将百度天气JSON数据转JavaBean

GSON框架下将百度天气JSON数据转JavaBean

《GSON框架下将百度天气JSON数据转JavaBean》这篇文章主要为大家详细介绍了如何在GSON框架下实现将百度天气JSON数据转JavaBean,文中的示例代码讲解详细,感兴趣的小伙伴可以了解下... 目录前言一、百度天气jsON1、请求参数2、返回参数3、属性映射二、GSON属性映射实战1、类对象映
阅读更多...
C#文件复制异常："未能找到文件"的解决方案与预防措施

C#文件复制异常："未能找到文件"的解决方案与预防措施

《C#文件复制异常：未能找到文件的解决方案与预防措施》在C#开发中,文件操作是基础中的基础,但有时最基础的File.Copy()方法也会抛出令人困惑的异常,当targetFilePath设置为D:2... 目录一个看似简单的文件操作问题问题重现与错误分析错误代码示例错误信息根本原因分析全面解决方案1. 确保
阅读更多...
C# LiteDB处理时间序列数据的高性能解决方案

C# LiteDB处理时间序列数据的高性能解决方案

《C#LiteDB处理时间序列数据的高性能解决方案》LiteDB作为.NET生态下的轻量级嵌入式NoSQL数据库,一直是时间序列处理的优选方案,本文将为大家大家简单介绍一下LiteDB处理时间序列数... 目录为什么选择LiteDB处理时间序列数据第一章：LiteDB时间序列数据模型设计1.1 核心设计原则
阅读更多...
SpringBoot3匹配Mybatis3的错误与解决方案

SpringBoot3匹配Mybatis3的错误与解决方案

《SpringBoot3匹配Mybatis3的错误与解决方案》文章指出SpringBoot3与MyBatis3兼容性问题,因未更新MyBatis-Plus依赖至SpringBoot3专用坐标,导致类冲... 目录SpringBoot3匹配MyBATis3的错误与解决mybatis在SpringBoot3如果
阅读更多...
Java整合Protocol Buffers实现高效数据序列化实践

Java整合Protocol Buffers实现高效数据序列化实践

《Java整合ProtocolBuffers实现高效数据序列化实践》ProtocolBuffers是Google开发的一种语言中立、平台中立、可扩展的结构化数据序列化机制,类似于XML但更小、更快... 目录一、Protocol Buffers简介1.1 什么是Protocol Buffers1.2 Pro
阅读更多...
C++ vector越界问题的完整解决方案

C++ vector越界问题的完整解决方案

《C++vector越界问题的完整解决方案》在C++开发中,std::vector作为最常用的动态数组容器,其便捷性与性能优势使其成为处理可变长度数据的首选,然而,数组越界访问始终是威胁程序稳定性的... 目录引言一、vector越界的底层原理与危害1.1 越界访问的本质原因1.2 越界访问的实际危害二、基
阅读更多...
Python 字符串裁切与提取全面且实用的解决方案

Python 字符串裁切与提取全面且实用的解决方案

《Python字符串裁切与提取全面且实用的解决方案》本文梳理了Python字符串处理方法,涵盖基础切片、split/partition分割、正则匹配及结构化数据解析（如BeautifulSoup、j... 目录python 字符串裁切与提取的完整指南 基础切片方法1. 使用切片操作符[start:end]2
阅读更多...
Linux部署中的文件大小写问题的解决方案

Linux部署中的文件大小写问题的解决方案

《Linux部署中的文件大小写问题的解决方案》在本地开发环境（Windows/macOS）一切正常,但部署到Linux服务器后出现模块加载错误,核心原因是Linux文件系统严格区分大小写,所以本文给大... 目录问题背景解决方案配置要求问题背景在本地开发环境（Windows/MACOS）一切正常，但部署到
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2