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。

相关文章

idea粘贴空格时显示NBSP的问题及解决方案

idea粘贴空格时显示NBSP的问题及解决方案

《idea粘贴空格时显示NBSP的问题及解决方案》在IDEA中粘贴代码时出现大量空格占位符NBSP,可以通过取消勾选AdvancedSettings中的相应选项来解决... 目录1、背景介绍2、解决办法3、处理完成总结1、背景介绍python在idehttp://www.chinasem.cna粘贴代码，出
阅读更多...
SpringBoot项目整合Netty启动失败的常见错误总结

SpringBoot项目整合Netty启动失败的常见错误总结

《SpringBoot项目整合Netty启动失败的常见错误总结》本文总结了SpringBoot集成Netty时常见的8类问题及解决方案,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参... 目录一、端口冲突问题1. Tomcat与Netty端口冲突二、主线程被阻塞问题1. Netty启动阻
阅读更多...
Java序列化之serialVersionUID的用法解读

Java序列化之serialVersionUID的用法解读

《Java序列化之serialVersionUID的用法解读》Java序列化之serialVersionUID：本文介绍了Java对象的序列化和反序列化过程,强调了serialVersionUID的作... 目录JavChina编程a序列化之serialVersionUID什么是序列化为什么要序列化serialV
阅读更多...
Apache服务器IP自动跳转域名的问题及解决方案

Apache服务器IP自动跳转域名的问题及解决方案

《Apache服务器IP自动跳转域名的问题及解决方案》本教程将详细介绍如何通过Apache虚拟主机配置实现这一功能,并解决常见问题,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,... 目录​​问题背景​​解决方案​​方法 1：修改 httpd-vhosts.conf（推荐）​​步骤
阅读更多...
idea-java序列化serialversionUID自动生成方式

idea-java序列化serialversionUID自动生成方式

《idea-java序列化serialversionUID自动生成方式》Java的Serializable接口用于实现对象的序列化和反序列化,通过将对象转换为字节流来存储或传输,实现Serializa... 目录简介实现序列化serialVersionUID配置使用总结简介Java.io.Seripyth
阅读更多...
java反序列化serialVersionUID不一致问题及解决

java反序列化serialVersionUID不一致问题及解决

《java反序列化serialVersionUID不一致问题及解决》文章主要讨论了在Java中序列化和反序列化过程中遇到的问题,特别是当实体类的`serialVersionUID`发生变化或未设置时,... 目录前言一、序列化、反序列化二、解决方法总结前言serialVersionUID变化后，反序列化失
阅读更多...
Linux(centos7)虚拟机没有IP问题及解决方案

Linux(centos7)虚拟机没有IP问题及解决方案

《Linux(centos7)虚拟机没有IP问题及解决方案》文章介绍了在CentOS7中配置虚拟机网络并使用Xshell连接虚拟机的步骤,首先,检查并配置网卡ens33的ONBOOT属性为yes,然后... 目录输入查看ZFhrxIP命令：ip addr查看，没有虚拟机IP修改ens33配置文件重启网络Xh
阅读更多...
Java编译错误java.lang.NoSuchFieldError的解决方案详析

Java编译错误java.lang.NoSuchFieldError的解决方案详析

《Java编译错误java.lang.NoSuchFieldError的解决方案详析》java.lang.NoSuchFieldError是Java中的一种运行时错误,：本文主要介绍Java编译错... 目录前言解决方案1. 统一JDK版本环境2. 优化maven-compiler-plugin配置3. 清
阅读更多...
Navicat连接Mysql8.0.11出现1251错误的解决方案

Navicat连接Mysql8.0.11出现1251错误的解决方案

《Navicat连接Mysql8.0.11出现1251错误的解决方案》在重装电脑并安装最新版MySQL后,Navicat和Sqlyog连接MySQL时遇到的1251和2058错误,通过将MySQL用户... 目录Navicat连接mysql8.0.11出现1251错误原因分析解决问题方法有两种总结Navic
阅读更多...
tomcat日志中文乱码问题及解决方案

tomcat日志中文乱码问题及解决方案

《tomcat日志中文乱码问题及解决方案》文章主要介绍了在使用Tomcat时遇到的乱码问题及其解决方法,页面输出乱码可能由于server.xml配置、HTML标签、编程输出编码不一致引起,解决方法包括... 目录一、页面输出乱码1.server.XML配置未注明编码格式2.catalina.bawww.cp
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2