Netty-2-数据编解码

解析编解码支持的原理

以编码为例,要将对象序列化成字节流,你可以使用MessageToByteEncoder或MessageToMessageEncoder类。

在这里插入图片描述
这两个类都继承自ChannelOutboundHandlerAdapter适配器类,用于进行数据的转换。

其中,对于MessageToMessageEncoder来说,如果把口标设置为ByteBuf,那么效果等同于使用MessageToByteEncodero这就是它们都可以进行数据编码的原因。

//MessageToMessageEncoder@Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {// 创建一个CodecOutputList对象,并将其初始化为nullCodecOutputList out = null;try {// 检查消息是否满足输出条件if (acceptOutboundMessage(msg)) {// 创建一个CodecOutputList对象,并将其赋值给out变量out = CodecOutputList.newInstance();// 将msg强制转换为I类型,并赋值给cast变量@SuppressWarnings("unchecked")I cast = (I) msg;try {// 调用encode方法,将ctx、cast和out作为参数传入encode(ctx, cast, out);} catch (Throwable th) {// 释放cast的引用计数ReferenceCountUtil.safeRelease(cast);// 抛出异常PlatformDependent.throwException(th);}// 释放cast的引用计数ReferenceCountUtil.release(cast);// 检查out是否为空if (out.isEmpty()) {// 抛出编码异常throw new EncoderException(StringUtil.simpleClassName(this) + " must produce at least one message.");}} else {// 直接将msg写入通道ctx.write(msg, promise);}} catch (EncoderException e) {// 抛出编码异常throw e;} catch (Throwable t) {// 抛出编码异常throw new EncoderException(t);} finally {// 最终,释放out的引用计数if (out != null) {try {// 获取out的元素个数final int sizeMinusOne = out.size() - 1;if (sizeMinusOne == 0) {// 将out的第一个元素直接写入通道ctx.write(out.getUnsafe(0), promise);} else if (sizeMinusOne > 0) {// 检查promise是否为voidPromiseif (promise == ctx.voidPromise()) {// 使用voidPromise来减少GC压力writeVoidPromise(ctx, out);} else {// 使用writePromiseCombiner方法来减少GC压力writePromiseCombiner(ctx, out, promise);}}} finally {// 释放out的资源out.recycle();}}}}protected abstract void encode(ChannelHandlerContext ctx, I msg, List<Object> out) throws Exception;

最终的目标是把对象转换为ByteBuf,具体的转换代码则委托子类继承的encode方法来实现。

Netty提供了很多子类来支持前面提及的各种数据编码方式。
在这里插入图片描述

解析典型Netty数据编解码的实现

HttpObjectEncoder编码器

//HttpObjectEncoder编码器@Override@SuppressWarnings("ConditionCoveredByFurtherCondition")protected void encode(ChannelHandlerContext ctx, Object msg, List<Object> out) throws Exception {// 为了处理不需要类检查的常见模式的fast-pathif (msg == Unpooled.EMPTY_BUFFER) {out.add(Unpooled.EMPTY_BUFFER);return;}// 以这种顺序进行instanceof检查的原因是,不依赖于ReferenceCountUtil::release作为一种通用释放机制,// 参见https://bugs.openjdk.org/browse/JDK-8180450。// https://github.com/netty/netty/issues/12708包含有关先前版本的此代码如何与JIT instanceof优化交互的更多详细信息。if (msg instanceof FullHttpMessage) {encodeFullHttpMessage(ctx, msg, out);return;}// 判断msg是否为HttpMessage的实例if (msg instanceof HttpMessage) {final H m;try {// 将msg转换为H类型m = (H) msg;} catch (Exception rethrow) {// 出现异常时,释放msg的引用计数并抛出异常ReferenceCountUtil.release(msg);throw rethrow;}// 判断m是否为LastHttpContent的实例if (m instanceof LastHttpContent) {// 调用encodeHttpMessageLastContent方法对LastHttpContent进行编码encodeHttpMessageLastContent(ctx, m, out);} // 判断m是否为HttpContent的实例else if (m instanceof HttpContent) {// 调用encodeHttpMessageNotLastContent方法对HttpContent进行编码encodeHttpMessageNotLastContent(ctx, m, out);} // m既不是LastHttpContent也不是HttpContent的实例else {// 调用encodeJustHttpMessage方法对m进行编码encodeJustHttpMessage(ctx, m, out);}} // msg不是HttpMessage的实例else {// 调用encodeNotHttpMessageContentTypes方法对非HttpMessage的内容类型进行编码encodeNotHttpMessageContentTypes(ctx, msg, out);}}

HttpObjectDecoder解码器

//HttpObjectDecoder.java/*** 定义了一个私有枚举类型State,表示不同的状态*/private enum State {/*** 用于跳过控制字符*/SKIP_CONTROL_CHARS,/*** 读取初始内容*/READ_INITIAL,/*** 读取头部信息*/READ_HEADER,/*** 读取可变长度的内容*/READ_VARIABLE_LENGTH_CONTENT,/*** 读取固定长度的内容*/READ_FIXED_LENGTH_CONTENT,/*** 读取分块大小*/READ_CHUNK_SIZE,/*** 读取分块内容*/READ_CHUNKED_CONTENT,/*** 读取分块分隔符*/READ_CHUNK_DELIMITER,/*** 读取分块脚注*/READ_CHUNK_FOOTER,/*** 错误消息*/BAD_MESSAGE,/*** 升级协议*/UPGRADED}//解码器相应实现@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {// 如果 resetRequested 为真if (resetRequested) {// 调用 resetNow() 方法resetNow();}switch (currentState) {case SKIP_CONTROL_CHARS:// 跳过控制字符case READ_INITIAL: try {// 解析缓冲区中的数据AppendableCharSequence line = lineParser.parse(buffer);if (line == null) {return;}// 拆分初始行String[] initialLine = splitInitialLine(line);if (initialLine.length < 3) {// 初始行无效 - 忽略currentState = State.SKIP_CONTROL_CHARS;return;}// 创建消息对象message = createMessage(initialLine);currentState = State.READ_HEADER;// 继续读取头部} catch (Exception e) {// 处理异常情况out.add(invalidMessage(buffer, e));return;}case READ_HEADER: try {State nextState = readHeaders(buffer);if (nextState == null) {return;}currentState = nextState;switch (nextState) {case SKIP_CONTROL_CHARS:// 快速路径// 无需期望任何内容out.add(message);out.add(LastHttpContent.EMPTY_LAST_CONTENT);resetNow();return;case READ_CHUNK_SIZE:if (!chunkedSupported) {throw new IllegalArgumentException("不支持分块消息");}// 分块编码 - 首先生成HttpMessage。后续将跟随HttpChunks。out.add(message);return;default:/*** <a href="https://tools.ietf.org/html/rfc7230#section-3.3.3">RFC 7230, 3.3.3</a> 规定,如果请求没有传输编码头或内容长度头,则消息体长度为0。* 但是对于响应,body长度是在服务器关闭连接之前接收到的字节数目。因此我们将此情况视为可变长度的分块编码。*/long contentLength = contentLength();if (contentLength == 0 || contentLength == -1 && isDecodingRequest()) {out.add(message);out.add(LastHttpContent.EMPTY_LAST_CONTENT);resetNow();return;}assert nextState == State.READ_FIXED_LENGTH_CONTENT ||nextState == State.READ_VARIABLE_LENGTH_CONTENT;out.add(message);if (nextState == State.READ_FIXED_LENGTH_CONTENT) {// 随着READ_FIXED_LENGTH_CONTENT状态逐块读取数据,分块大小将减小。chunkSize = contentLength;}// 在这里返回,这将强制再次调用解码方法,在那里我们将解码内容return;}} catch (Exception e) {out.add(invalidMessage(buffer, e));return;}case READ_VARIABLE_LENGTH_CONTENT: {// 一直读取数据直到连接结束。int toRead = Math.min(buffer.readableBytes(), maxChunkSize);if (toRead > 0) {// 从缓冲区中读取指定长度的数据,并以保留引用的形式分割成多个片段ByteBuf content = buffer.readRetainedSlice(toRead);out.add(new DefaultHttpContent(content));}return;}case READ_FIXED_LENGTH_CONTENT: {int readLimit = buffer.readableBytes();// 首先检查缓冲区是否可读,因为我们使用可读字节计数来创建HttpChunk。需要这样做,以防止创建包含空缓冲区的HttpChunk,从而被当作最后一个HttpChunk进行处理。// 参见:https://github.com/netty/netty/issues/433if (readLimit == 0) {return;}int toRead = Math.min(readLimit, maxChunkSize);if (toRead > chunkSize) {toRead = (int) chunkSize;}ByteBuf content = buffer.readRetainedSlice(toRead);chunkSize -= toRead;if (chunkSize == 0) {// 读取所有内容。out.add(new DefaultLastHttpContent(content, validateHeaders));resetNow();} else {out.add(new DefaultHttpContent(content));}return;}/*** 从这里开始处理读取分块的内容。基本上,读取分块大小,读取分块,忽略CRLF,然后重复直到分块大小为0*/case READ_CHUNK_SIZE: try {AppendableCharSequence line = lineParser.parse(buffer);if (line == null) {return;}int chunkSize = getChunkSize(line.toString());this.chunkSize = chunkSize;if (chunkSize == 0) {currentState = State.READ_CHUNK_FOOTER;return;}currentState = State.READ_CHUNKED_CONTENT;// fall-through} catch (Exception e) {out.add(invalidChunk(buffer, e));return;}case READ_CHUNKED_CONTENT: {// 判断chunkSize是否小于等于Integer的最大值assert chunkSize <= Integer.MAX_VALUE;// 计算本次需要读取的字节数,取chunkSize和maxChunkSize中的较小值int toRead = Math.min((int) chunkSize, maxChunkSize);// 如果不允许部分chunk,且buffer中可读取的字节数小于toRead,则返回if (!allowPartialChunks && buffer.readableBytes() < toRead) {return;}// 如果buffer中可读取的字节数小于toRead,则将toRead更新为buffer中可读取的字节数toRead = Math.min(toRead, buffer.readableBytes());// 如果toRead为0,则返回if (toRead == 0) {return;}// 从buffer中获取长度为toRead的slice,并用其创建HttpContent对象HttpContent chunk = new DefaultHttpContent(buffer.readRetainedSlice(toRead));// 更新剩余的chunkSizechunkSize -= toRead;// 将chunk添加到out中// 如果chunkSize不为0,则返回if (chunkSize != 0) {return;}// 设置当前状态为READ_CHUNK_DELIMITERcurrentState = State.READ_CHUNK_DELIMITER;// 继续执行下一个case语句// fall-through}case READ_CHUNK_DELIMITER: {// 读取分隔符final int wIdx = buffer.writerIndex();int rIdx = buffer.readerIndex();while (wIdx > rIdx) {byte next = buffer.getByte(rIdx++);if (next == HttpConstants.LF) {currentState = State.READ_CHUNK_SIZE;break;}}buffer.readerIndex(rIdx);return;}case READ_CHUNK_FOOTER: {try {// 读取尾部的Http头部信息LastHttpContent trailer = readTrailingHeaders(buffer);if (trailer == null) {return;}out.add(trailer);resetNow();return;} catch (Exception e) {// 发生异常时,将异常信息和当前buffer一起添加到输出channelout.add(invalidChunk(buffer, e));return;}}case BAD_MESSAGE: {// 直到断开连接为止,丢弃消息buffer.skipBytes(buffer.readableBytes());break;}case UPGRADED: {int readableBytes = buffer.readableBytes();if (readableBytes > 0) {// 读取可读字节数,如果大于0,则执行以下操作// 由于否则可能会触发一个DecoderException异常,其他处理器会在某个时刻替换此codec为升级的协议codec来接管流量。// 参见 https://github.com/netty/netty/issues/2173out.add(buffer.readBytes(readableBytes));}break;}default:break;}}

自定义编解码

下面先实现一个Netty编码处理程序。

public class OrderProtocolEncoder extends MessageToMessageEncoder<ResponseMessage> {/*** 编码器类,用于将ResponseMessage对象编码为ByteBuf对象并添加到输出列表中*/@Overrideprotected void encode(ChannelHandlerContext ctx, ResponseMessage responseMessage, List<Object> out) throws Exception {/*** 获取一个ByteBuf对象用于存储编码后的数据*/ByteBuf buffer = ctx.alloc().buffer();/*** 对ResponseMessage对象进行编码,并将编码后的数据写入ByteBuf对象中*/responseMessage.encode(buffer);/*** 将编码后的ByteBuf对象添加到输出列表中*/out.add(buffer);}
}

接下来,再实现对应的Netty解码处理程序。

/*** 订单协议解码器*/
public class OrderProtocolDecoder extends MessageToMessageDecoder<ByteBuf> {@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, List<Object> out) throws Exception {// 创建一个请求消息对象RequestMessage requestMessage = new RequestMessage();// 对字节缓冲区进行解码,将解码后的消息填充到请求消息对象中requestMessage.decode(byteBuf);// 将请求消息对象添加到输出列表中out.add(requestMessage);}
}

最后,将这对编解码处理程序添加到处理程序流水线(pipeline)中就可以完成集成工作了。

这是我们第一次提及处理程序流水线这个概念。在这里,只需要将它理解成"一串”有序的处理程序集合并有一个初步印象即可,后续会详细介绍相关内容。

为了完成处理程序流水线的设置,还要构建ServerBootstrap这个“启动”对象。

        ServerBootstrap serverBootstrap = new ServerBootstrap();  // 创建一个ServerBootstrap对象serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {  // 为子通道设置ChannelInitializer处理器@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {  // 初始化连接通道ChannelPipeline pipeline = ch.pipeline();  // 获取通道的编排器// 省略其他非核心代码pipeline.addLast("protocolDecoder", new OrderProtocolDecoder());  // 添加一个解码器到通道的最后pipeline.addLast("protocolEncoder", new OrderProtocolEncoder());  // 添加一个编码器到通道的最后// 省略其他非核心代码}});

常见疑问解析

为什么Netty自带的编解码方案很少有人使用

其中个很重要的因素就是历史原因,但实际上,除历史原因之外,更重要的原因在于Netty自带的编解码方案大多是具有封帧和解帧功能的编解码器,并且融两层编码于一体,因此从结构上看并不清晰。

另外,Netty自带的编解码方案在使用方式上不够灵活。

在进行序列化和反序列时,字段的顺序弄反了

我们在序列化对象的字段时,使用的顺序是a b c;但是,等到我们解析时,顺序可能不小心写成了 c b a, 因此,我们一定要完全对照好顺序才行。

编解码的顺序问题

有时候,我们往往采用多层编解码。
例如,在得到可传输的字节流之后,我们可能想压缩一下以进一步减少所传输内容占用的空间。
此时,多级编解码就可以派上用场了:对于发送者, 先编码后压缩;而对于接收者,先解压后解码。

但是,代码的添加顺序和我们想要的顺序不一定完全匹配。如果顺序错了,那么代码可能无法工作。

if (compressor != null) {pipeline.addLast("frameDecompressorn", new Frame.Decompressor(compressor));pipeline.addLast("frameCompressor", new Frame.Compressor(compressor));pipeline.addLast("messageDecoder", messageDecoder);pipeline.addLast("messageEncoder", messageEncoderFor(protocolversion));
}

在这里插入图片描述
处理程序对于读取操作和写出操作的执行顺序刚好是相反的。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/241566.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

基于 Webpack 插件体系的 Mock 服务

背景 在软件研发流程中&#xff0c;对于前后端分离的架构体系而言&#xff0c;为了能够更快速、高效的实现功能的开发&#xff0c;研发团队通常来说会在产品原型阶段对前后端联调的数据接口进行结构设计及约定&#xff0c;进而可以分别同步进行对应功能的实现&#xff0c;提升研…

深度学习 | 基础卷积神经网络

卷积神经网络是人脸识别、自动驾驶汽车等大多数计算机视觉应用的支柱。可以认为是一种特殊的神经网络架构&#xff0c;其中基本的矩阵乘法运算被卷积运算取代&#xff0c;专门处理具有网格状拓扑结构的数据。 1、全连接层的问题 1.1、全连接层的问题 “全连接层”的特点是每个…

kubernetes集群 应用实践 kafka部署

kubernetes集群 应用实践 kafka部署 零.1、环境说明 零.2、kafka架构说明 zookeeper在kafka集群中的作用 一、Broker注册 二、Topic注册 三、Topic Partition选主 四、生产者负载均衡 五、消费者负载均衡 一、持久化存储资源准备 1.1 创建共享目录 [rootnfsserver ~]# mkdir -…

锯齿云服务器租赁使用教程

首先登陆锯齿云账号 网盘上传数据集与代码 随后我们需要做的是将所需要的数据集与代码上传到网盘&#xff08;也可以直接在租用服务器后将数据集与代码传到服务器的硬盘上&#xff0c;但这样做会消耗大量时间&#xff0c;造成资源浪费&#xff09; 点击工作空间&#xff1a;…

谷粒商城-商品服务-新增商品功能开发(商品图片无法展示问题没有解决)

在网关配置路由 - id: member_routeuri: lb://gulimemberpredicates:- Path/api/gulimember/**filters:- RewritePath/api/(?<segment>.*),/$\{segment}并将所有逆向生成的工程调式出来 获取分类关联的品牌 例如&#xff1a;手机&#xff08;分类&#xff09;-> 品…

Python算法例26 落单的数Ⅳ

1. 问题描述 给定数组&#xff0c;除了一个数出现一次外&#xff0c;所有数都出现两次&#xff0c;并且所有出现两次的数都挨着&#xff0c;找出出现一次的数。 2. 问题示例 给出nums[3&#xff0c;3&#xff0c;2&#xff0c;2&#xff0c;4&#xff0c;5&#xff0c;5]&am…

ZooKeeper 使用介绍和原理详解

目录 1. 介绍 重要性 应用场景 2. ZooKeeper 架构 服务角色 数据模型 工作原理 3. 安装和配置 下载 ZooKeeper 安装和配置 启动 ZooKeeper 验证和管理 停止和关闭 4. ZooKeeper 数据模型 数据结构和层次命名空间&#xff1a; 节点类型和 Watcher 机制&#xff…

基于python的excel检查和读写软件

软件版本&#xff1a;python3.6 窗口和界面gui代码&#xff1a; class mygui:def _init_(self):passdef run(self):root Tkinter.Tk()root.title(ExcelRun)max_w, max_h root.maxsize()root.geometry(f500x500{int((max_w - 500) / 2)}{int((max_h - 300) / 2)}) # 居中显示…

【MySQL】MySQL的数据类型

MySQL的数据类型 一、数据类型分类二、数值类型1、整数类型2、bit类型3、小数类型 三、字符串类型四、时间日期类型五、enum和set类型enum和set查找 数据类型的作用&#xff1a; 决定了存储数据时应该开辟的空间大小和数据的取值范围。决定了如何识别一个特定的二进制序列。 …

AI创作系统ChatGPT系统源码,支持Midjourney绘画,GPT语音对话+DALL-E3文生图

一、前言 SparkAi创作系统是基于ChatGPT进行开发的Ai智能问答系统和Midjourney绘画系统&#xff0c;支持OpenAI-GPT全模型国内AI全模型。本期针对源码系统整体测试下来非常完美&#xff0c;可以说SparkAi是目前国内一款的ChatGPT对接OpenAI软件系统。那么如何搭建部署AI创作Ch…

R语言基础 | 安徽某高校《统计建模与R软件》期末复习

第一节 数字、字符与向量 1.1 向量的赋值 c<-(1,2,3,4,5) 1.2 向量的运算 对于向量&#xff0c;我们可以直接对其作加&#xff08;&#xff09;&#xff0c;减&#xff08;-&#xff09;&#xff0c;乘&#xff08;*&#xff09;&#xff0c;除&#xff08;/&#xff09…

【shell脚本实战学习笔记】#1

shell脚本实战学习笔记#1 脚本编写场景需求&#xff1a; 编写一个比较数据大小的shell脚本&#xff0c;要求判断用户只能输入两位数字&#xff0c;不能是字符或其他特殊字符&#xff1b;并且在shell脚本中需要用到函数来控制执行顺序。 知识点&#xff1a;shell函数&#xff…

科研学习|论文解读——面向电商内容安全风险管控的协同过滤推荐算法研究

【论文完整内容详见知网链接】&#xff1a; 面向电商内容安全风险管控的协同过滤推荐算法研究 - 中国知网 (cnki.net) 面向电商内容安全风险管控的协同过滤推荐算法研究* 摘 要&#xff1a;[目的/意义]随着电商平台商家入驻要求降低以及商品上线审核流程简化&#xff0c;内容安…

Centos安装vsftpd:centos配置vsftpd,ftp报200和227错误

一、centos下载安装vsftpd&#xff08;root权限&#xff09; 1、下载安装 yum -y install vsftpd 2、vsftpd的配置文件 /etc/vsftpd.conf 3、备份原来的配置文件 sudo cp /etc/vsftpd.conf /etc/vsftpd.conf.backup 4、修改配置文件如下&#xff1a;vi /etc/vsftpd.conf …

体验一下 CodeGPT 插件

体验一下 CodeGPT 插件 0. 背景1. CodeGPT 插件安装2. CodeGPT 插件基本配置3. (可选)CodeGPT 插件预制提示词原始配置(英文)4. CodeGPT 插件预制提示词配置(中文)5. 简单验证一下 0. 背景 看到B站Up主 “wwwzhouhui” 一个关于 CodeGPT 的视频&#xff0c;感觉挺有意思&#…

SpringMVC:整合 SSM 中篇

文章目录 SpringMVC - 04整合 SSM 中篇一、优化二、总结三、说明注意&#xff1a; SpringMVC - 04 整合 SSM 中篇 一、优化 在 spring-dao.xml 中配置 dao 接口扫描&#xff0c;可以动态地实现 dao 接口注入到 Spring 容器中。 优化前&#xff1a;手动创建 SqlSessionTempl…

STM32实现三个小灯亮

led.c #include"led.h"void Led_Init(void) {GPIO_InitTypeDef GPIO_VALUE; //???RCC_APB2PeriphClockCmd(RCC_APB2Periph_GPIOC,ENABLE);//???GPIO_VALUE.GPIO_ModeGPIO_Mode_Out_PP;//???? ????GPIO_VALUE.GPIO_PinGPIO_Pin_1|GPIO_Pin_2|GPIO_P…

spring-validation实现分组校验

文章目录 前言实际开发可能会使用到分组校验maven添加依赖简单使用高级应用分组自定义分组组合分组 源码地址 前言 JSR 303中提出了Bean Validation&#xff0c;表示JavaBean的校验&#xff0c;Hibernate Validation是其具体实现&#xff0c;并对其进行了一些扩展&#xff0c;…

Arduino上U8g2库显示中文的经历

u8g2自带很多中文库&#xff1b;但是向u8g2_font_wqy12_t_chinese3 比较全的应该是u8g2_font_wqy12_t_gb2312 这个&#xff0c;只是我还没有调用成功 这个库&#xff0c;中文就显示不全&#xff1b;有些没有定义&#xff0c;如百家姓 #include <Arduino.h> #include <…

Java经典框架之Spring

Java经典框架之Spring Java 是第一大编程语言和开发平台。它有助于企业降低成本、缩短开发周期、推动创新以及改善应用服务。如今全球有数百万开发人员运行着超过 51 亿个 Java 虚拟机&#xff0c;Java 仍是企业和开发人员的首选开发平台。 课程内容的介绍 1. Spring简介 2.…