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。

相关文章

jupyter代码块没有运行图标的解决方案

jupyter代码块没有运行图标的解决方案

《jupyter代码块没有运行图标的解决方案》：本文主要介绍jupyter代码块没有运行图标的解决方案,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录jupyter代码块没有运行图标的解决1.找到Jupyter notebook的系统配置文件2.这时候一般会搜索到
阅读更多...
Nginx中配置HTTP/2协议的详细指南

Nginx中配置HTTP/2协议的详细指南

《Nginx中配置HTTP/2协议的详细指南》HTTP/2是HTTP协议的下一代版本,旨在提高性能、减少延迟并优化现代网络环境中的通信效率,本文将为大家介绍Nginx配置HTTP/2协议想详细步骤,需... 目录一、HTTP/2 协议概述1.HTTP/22. HTTP/2 的核心特性3. HTTP/2 的优
阅读更多...
关于WebSocket协议状态码解析

关于WebSocket协议状态码解析

《关于WebSocket协议状态码解析》：本文主要介绍关于WebSocket协议状态码的使用方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录WebSocket协议状态码解析1. 引言2. WebSocket协议状态码概述3. WebSocket协议状态码详解3
阅读更多...
Linux samba共享慢的原因及解决方案

Linux samba共享慢的原因及解决方案

《Linuxsamba共享慢的原因及解决方案》：本文主要介绍Linuxsamba共享慢的原因及解决方案,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录linux samba共享慢原因及解决问题表现原因解决办法总结Linandroidux samba共享慢原因及解决
阅读更多...
Java的IO模型、Netty原理解析

Java的IO模型、Netty原理解析

《Java的IO模型、Netty原理解析》Java的I/O是以流的方式进行数据输入输出的,Java的类库涉及很多领域的IO内容：标准的输入输出,文件的操作、网络上的数据传输流、字符串流、对象流等,这篇... 目录1.什么是IO2.同步与异步、阻塞与非阻塞3.三种IO模型BIO（blocking I/O）NI
阅读更多...
Python Dash框架在数据可视化仪表板中的应用与实践记录

Python Dash框架在数据可视化仪表板中的应用与实践记录

《PythonDash框架在数据可视化仪表板中的应用与实践记录》Python的PlotlyDash库提供了一种简便且强大的方式来构建和展示互动式数据仪表板,本篇文章将深入探讨如何使用Dash设计一... 目录python Dash框架在数据可视化仪表板中的应用与实践1. 什么是Plotly Dash？1.1
阅读更多...
找不到Anaconda prompt终端的原因分析及解决方案

找不到Anaconda prompt终端的原因分析及解决方案

《找不到Anacondaprompt终端的原因分析及解决方案》因为anaconda还没有初始化,在安装anaconda的过程中,有一行是否要添加anaconda到菜单目录中,由于没有勾选,导致没有菜... 目录问题原因问http://www.chinasem.cn题解决安装了 Anaconda 却找不到 An
阅读更多...
Spring定时任务只执行一次的原因分析与解决方案

Spring定时任务只执行一次的原因分析与解决方案

《Spring定时任务只执行一次的原因分析与解决方案》在使用Spring的@Scheduled定时任务时,你是否遇到过任务只执行一次,后续不再触发的情况？这种情况可能由多种原因导致,如未启用调度、线程... 目录1. 问题背景2. Spring定时任务的基本用法3. 为什么定时任务只执行一次？3.1 未启用
阅读更多...
基于Flask框架添加多个AI模型的API并进行交互

基于Flask框架添加多个AI模型的API并进行交互

《基于Flask框架添加多个AI模型的API并进行交互》：本文主要介绍如何基于Flask框架开发AI模型API管理系统,允许用户添加、删除不同AI模型的API密钥,感兴趣的可以了解下... 目录1. 概述2. 后端代码说明2.1 依赖库导入2.2 应用初始化2.3 API 存储字典2.4 路由函数2.5 应
阅读更多...
MySQL新增字段后Java实体未更新的潜在问题与解决方案

MySQL新增字段后Java实体未更新的潜在问题与解决方案

《MySQL新增字段后Java实体未更新的潜在问题与解决方案》在Java+MySQL的开发中,我们通常使用ORM框架来映射数据库表与Java对象,但有时候,数据库表结构变更（如新增字段）后,开发人员可... 目录引言1. 问题背景：数据库与 Java 实体不同步1.1 常见场景1.2 示例代码2. 不同操作
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2