wordpress分享视频网站,网站建设海淀,网站设计与建设ppt,网站开发语言用到解析编解码支持的原理
以编码为例#xff0c;要将对象序列化成字节流#xff0c;你可以使用MessageToByteEncoder或MessageToMessageEncoder类。 这两个类都继承自ChannelOutboundHandlerAdapter适配器类#xff0c;用于进行数据的转换。
其中#xff0c;对于MessageToMe…解析编解码支持的原理
以编码为例要将对象序列化成字节流你可以使用MessageToByteEncoder或MessageToMessageEncoder类。 这两个类都继承自ChannelOutboundHandlerAdapter适配器类用于进行数据的转换。
其中对于MessageToMessageEncoder来说如果把口标设置为ByteBuf,那么效果等同于使用MessageToByteEncodero这就是它们都可以进行数据编码的原因。
//MessageToMessageEncoderOverridepublic 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, ListObject out) throws Exception;最终的目标是把对象转换为ByteBuf,具体的转换代码则委托子类继承的encode方法来实现。
Netty提供了很多子类来支持前面提及的各种数据编码方式。
解析典型Netty数据编解码的实现
HttpObjectEncoder编码器
//HttpObjectEncoder编码器OverrideSuppressWarnings(ConditionCoveredByFurtherCondition)protected void encode(ChannelHandlerContext ctx, Object msg, ListObject 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, ListObject 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 hrefhttps://tools.ietf.org/html/rfc7230#section-3.3.3RFC 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 MessageToMessageEncoderResponseMessage {/*** 编码器类用于将ResponseMessage对象编码为ByteBuf对象并添加到输出列表中*/Overrideprotected void encode(ChannelHandlerContext ctx, ResponseMessage responseMessage, ListObject out) throws Exception {/*** 获取一个ByteBuf对象用于存储编码后的数据*/ByteBuf buffer ctx.alloc().buffer();/*** 对ResponseMessage对象进行编码并将编码后的数据写入ByteBuf对象中*/responseMessage.encode(buffer);/*** 将编码后的ByteBuf对象添加到输出列表中*/out.add(buffer);}
}
接下来再实现对应的Netty解码处理程序。
/*** 订单协议解码器*/
public class OrderProtocolDecoder extends MessageToMessageDecoderByteBuf {Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, ListObject out) throws Exception {// 创建一个请求消息对象RequestMessage requestMessage new RequestMessage();// 对字节缓冲区进行解码将解码后的消息填充到请求消息对象中requestMessage.decode(byteBuf);// 将请求消息对象添加到输出列表中out.add(requestMessage);}
}
最后将这对编解码处理程序添加到处理程序流水线(pipeline)中就可以完成集成工作了。
这是我们第一次提及处理程序流水线这个概念。在这里只需要将它理解成一串”有序的处理程序集合并有一个初步印象即可后续会详细介绍相关内容。
为了完成处理程序流水线的设置还要构建ServerBootstrap这个“启动”对象。 ServerBootstrap serverBootstrap new ServerBootstrap(); // 创建一个ServerBootstrap对象serverBootstrap.childHandler(new ChannelInitializerNioSocketChannel() { // 为子通道设置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));
}处理程序对于读取操作和写出操作的执行顺序刚好是相反的。
