深入了解Netty编码器 - MessageToByte在Netty中的核心作用
目录
- 简介
- MessageToByte框架简介
- MessageToByteEncoder
- ByteToMessageDecoder
- ByteToMessageCodec
- 总结
简介
之前的文章中,我们讲解了netty中从一个message转换成为另外一个message的框架叫做MessageToMessage编码器。但是message to message只考虑了channel中消息在处理过程中的转换,但是我们知道channel中最终传输的数据一定是ByteBuf,所以我们还需要一个message和ByteBuf相互转换的框架,这个框架就叫做MessageToByte。
注意,这里的byte指的是ByteBuf而不是byte这个字节类型。
MessageToByte框架简介
为了方便扩展和用户的自定义,netty封装了一套MessageToByte框架,这个框架中有三个核心的类,分别是MessageToByteEncoder,ByteToMessageDecoder和ByteToMessageCodec。
我们分别看一下这三个核心类的定义:
public abstract class MessageToByteEncoder<I> extends ChannelOutboundHandlerAdapter
public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter
public abstract class ByteToMessageCodec<I> extends ChannelDuplexHandler
这三个类分别继承自ChannelOutboundHandlerAdapter,ChannelInboundHandlerAdapter和ChannelDuplexHandler,分别表示的是向channel中写消息,从channel中读消息和一个向channel中读写消息的双向操作。
这三个类都是抽象类,接下来我们会详细分析这三个类的具体实现逻辑。
MessageToByteEncoder
先来看encoder,如果你对比MessageToByteEncoder和MessageToMessageEncoder的源码实现,可以发现他们有诸多相似之处。
首先在MessageToByteEncoder中定义了一个用作消息类型匹配的TypeParameterMatcher。
这个matcher用来匹配收到的消息类型,如果类型匹配则进行消息的转换操作,否则直接将消息写入channel中。
和MessageToMessageEncoder不同的是,MessageToByteEncoder多了一个preferDirect字段,这个字段表示消息转换成为ByteBuf的时候是使用diret Buf还是heap Buf。
这个字段的使用情况如下:
protected ByteBuf allocateBuffer(ChannelHandlerContext ctx, @SuppressWarnings("unused") I msg, boolean preferDirect) throws Exception { if (preferDirect) { return ctx.alloc().ioBuffer(); } else { return ctx.alloc().heapBuffer(); } }
最后来看一下它的核心方法write:
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { ByteBuf buf = null; try { if (acceptOutboundMessage(msg)) { @SuppressWarnings("unchecked") I cast = (I) msg; buf = allocateBuffer(ctx, cast, preferDirect); try { encode(ctx, cast, buf); } finally { ReferenceCountUtil.release(cast); } if (buf.isReadable()) { ctx.write(buf, promise); } else { buf.release(); ctx.write(Unpooled.EMPTY_BUFFER, promise); } buf = null; } else { ctx.write(msg, promise); } } catch (EncoderException e) { throw e; } catch (Throwable e) { throw new EncoderException(e); } finally { if (buf != null) { buf.release(); } } }
上面我们已经提到了,write方法首先通过matcher来判断是否是要接受的消息类型,如果是的话就调用encode方法,将消息对象转换成为ByteBuf,如果不是,则直接将消息写入channel中。
和MessageToMessageEncoder不同的是,encode方法需要传入一个ByteBuf对象,而不是CodecOutputList。
MessageToByteEncoder有一个需要实现的抽象方法encode如下,
protected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception;
ByteToMessageDecoder
ByteToMessageDecoder用来将channel中的ByteBuf消息转换成为特定的消息类型,其中Decoder中最重要的方法就是好channelRead方法,如下所示:
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { if (msg instanceof ByteBuf) { CodecOutputList out = CodecOutputList.newInstance(); try { first = cumulation == null; cumulation = cumulator.cumulate(ctx.alloc(), first ? Unpooled.EMPTY_BUFFER : cumulation, (ByteBuf) msg); callDecode(ctx, cumulation, out); } catch (DecoderException e) { throw e; } catch (Exception e) { throw new DecoderException(e); } finally { try { if (cumulation != null && !cumulation.isReadable()) { numReads = 0; cumulation.release(); cumulation = null; } else if (++numReads >= discardAfterReads) { numReads = 0; discardSomeReadBytes(); } int size = out.size(); firedChannelRead |= out.insertSinceRecycled(); fireChannelRead(ctx, out, size); } finally { out.recycle(); } } } else { ctx.fireChannelRead(msg); } }
channelRead接收要进行消息读取的Object对象,因为这里只接受ByteBuf消息,所以在方法内部调用了msg instanceof ByteBuf
来判断消息的类型,如果不是ByteBuf类型的消息则不进行消息的转换。
输出的对象是CodecOutputList,在将ByteBuf转换成为CodecOutputList之后,调用fireChannelRead方法将out对象传递下去。
这里的关键就是如何将接收到的ByteBuf转换成为CodecOutputList。
转换的方法叫做callDecode,它接收一个叫做cumulation的参数,在上面的方法中,我们还看到一个和cumulation非常类似的名称叫做cumulator。那么他们两个有什么区别呢?
在ByteToMessageDecoder中cumulation是一个ByteBuf对象,而Cumulator是一个接口,这个接口定义了一个cumulate方法:
public interface Cumulator { ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in); }
Cumulator用来将传入的ByteBuf合并成为一个新的ByteBuf。
ByteToMessageDecoder中定义了两种Cumulator,分别是MERGE_CUMULATOR和COMPOSITE_CUMULATOR。
MERGE_CUMULATOR是将传入的ByteBuf通过memory copy的方式拷贝到目标ByteBuf cumulation中。
而COMPOSITE_CUMULATOR则是将ByteBuf添加到一个 CompositeByteBuf 的结构中,并不做memory copy,因为目标的结构比较复杂,所以速度会比直接进行memory copy要慢。
用户要扩展的方法就是decode方法,用来将一个ByteBuf转换成为其他对象:
protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;
ByteToMessageCodec
最后要介绍的类是ByteToMessageCodec,ByteToMessageCodec表示的是message和ByteBuf之间的互相转换,它里面的encoder和decoder分别就是上面讲到的MessageToByteEncoder和ByteToMessageDecoder。
用户可以继承ByteToMessageCodec来同时实现encode和decode的功能,所以需要实现encode和decode这两个方法:
protected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception; protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;
ByteToMessageCodec的本质就是封装了MessageToByteEncoder和ByteToMessageDecoder,然后实现了编码和解码的功能。
总结
如果想实现ByteBuf和用户自定义消息的直接转换,那么选择netty提供的上面三个编码器是一个很好的选择。
推荐阅读
-
【Netty】「萌新入门」(七)ByteBuf 的性能优化-堆内存的分配和释放都是由 Java 虚拟机自动管理的,这意味着它们可以快速地被分配和释放,但是也会产生一些开销。 直接内存需要手动分配和释放,因为它由操作系统管理,这使得分配和释放的速度更快,但是也需要更多的系统资源。 另外,直接内存可以映射到本地文件中,这对于需要频繁读写文件的应用程序非常有用。 此外,直接内存还可以避免在使用 NIO 进行网络传输时发生数据拷贝的情况。在使用传统的 I/O 时,数据必须先从文件或网络中读取到堆内存中,然后再从堆内存中复制到直接缓冲区中,最后再通过 SocketChannel 发送到网络中。而使用直接缓冲区时,数据可以直接从文件或网络中读取到直接缓冲区中,并且可以直接从直接缓冲区中发送到网络中,避免了不必要的数据拷贝和内存分配。 通过 ByteBufAllocator.DEFAULT.directBuffer 方法来创建基于直接内存的 ByteBuf: ByteBuf directBuf = ByteBufAllocator.DEFAULT.directBuffer(16); 通过 ByteBufAllocator.DEFAULT.heapBuffer 方法来创建基于堆内存的 ByteBuf: ByteBuf heapBuf = ByteBufAllocator.DEFAULT.heapBuffer(16); 注意: 直接内存是一种特殊的内存分配方式,可以通过在堆外申请内存来避免 JVM 堆内存的限制,从而提高读写性能和降低 GC 压力。但是,直接内存的创建和销毁代价昂贵,因此需要慎重使用。 此外,由于直接内存不受 JVM 垃圾回收的管理,我们需要主动释放这部分内存,否则会造成内存泄漏。通常情况下,可以使用 ByteBuffer.clear 方法来释放直接内存中的数据,或者使用 ByteBuffer.cleaner 方法来手动释放直接内存空间。 测试代码: public static void testCreateByteBuf { ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(16); System.out.println(buf.getClass); ByteBuf heapBuf = ByteBufAllocator.DEFAULT.heapBuffer(16); System.out.println(heapBuf.getClass); ByteBuf directBuf = ByteBufAllocator.DEFAULT.directBuffer(16); System.out.println(directBuf.getClass); } 运行结果: class io.netty.buffer.PooledUnsafeDirectByteBuf class io.netty.buffer.PooledUnsafeHeapByteBuf class io.netty.buffer.PooledUnsafeDirectByteBuf 池化技术 在 Netty 中,池化技术指的是通过对象池来重用已经创建的对象,从而避免了频繁地创建和销毁对象,这种技术可以提高系统的性能和可伸缩性。 通过设置 VM options,来决定池化功能是否开启: -Dio.netty.allocator.type={unpooled|pooled} 在 Netty 4.1 版本以后,非 Android 平台默认启用池化实现,Android 平台启用非池化实现; 这里我们使用非池化功能进行测试,依旧使用的是上面的测试代码 testCreateByteBuf,运行结果如下所示: class io.netty.buffer.UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeDirectByteBuf class io.netty.buffer.UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeHeapByteBuf class io.netty.buffer.UnpooledByteBufAllocator$InstrumentedUnpooledUnsafeDirectByteBuf 可以看到,ByteBuf 类由 PooledUnsafeDirectByteBuf 变成了 UnpooledUnsafeDirectByteBuf; 在没有池化的情况下,每次使用都需要创建新的 ByteBuf 实例,这个操作会涉及到内存的分配和初始化,如果是直接内存则代价更为昂贵,而且频繁的内存分配也可能导致内存碎片问题,增加 GC 压力。 使用池化技术可以避免频繁内存分配带来的开销,并且重用池中的 ByteBuf 实例,减少了内存占用和内存碎片问题。另外,池化技术还可以采用类似 jemalloc 的内存分配算法,进一步提升分配效率。 在高并发环境下,池化技术的优点更加明显,因为内存的分配和释放都是比较耗时的操作,频繁的内存分配和释放会导致系统性能下降,甚至可能出现内存溢出的风险。使用池化技术可以将内存分配和释放的操作集中到预先分配的池中,从而有效地降低系统的内存开销和风险。 内存释放 当在 Netty 中使用 ByteBuf 来处理数据时,需要特别注意内存回收问题。 Netty 提供了不同类型的 ByteBuf 实现,包括堆内存(JVM 内存)实现 UnpooledHeapByteBuf 和堆外内存(直接内存)实现 UnpooledDirectByteBuf,以及池化技术实现的 PooledByteBuf 及其子类。 UnpooledHeapByteBuf:通过 Java 的垃圾回收机制来自动回收内存; UnpooledDirectByteBuf:由于 JVM 的垃圾回收机制无法管理这些内存,因此需要手动调用 release 方法来释放内存; PooledByteBuf:使用了池化机制,需要更复杂的规则来回收内存; 由于池化技术的特殊性质,释放 PooledByteBuf 对象所使用的内存并不是立即被回收的,而是被放入一个内存池中,待下次分配内存时再次使用。因此,释放 PooledByteBuf 对象的内存可能会延迟到后续的某个时间点。为了避免内存泄漏和占用过多内存,我们需要根据实际情况来设置池化技术的相关参数,以便及时回收内存; Netty 采用了引用计数法来控制 ByteBuf 对象的内存回收,在博文 「源码解析」ByteBuf 的引用计数机制 中将会通过解读源码的形式对 ByteBuf 的引用计数法进行深入理解; 每个 ByteBuf 对象被创建时,都会初始化为1,表示该对象的初始计数为1。 在使用 ByteBuf 对象过程中,如果当前 handler 已经使用完该对象,需要通过调用 release 方法将计数减1,当计数为0时,底层内存会被回收,该对象也就被销毁了。此时即使 ByteBuf 对象还在,其各个方法均无法正常使用。 但是,如果当前 handler 还需要继续使用该对象,可以通过调用 retain 方法将计数加1,这样即使其他 handler 已经调用了 release 方法,该对象的内存仍然不会被回收。这种机制可以有效地避免了内存泄漏和意外访问已经释放的内存的情况。 一般来说,应该尽可能地保证 retain 和 release 方法成对出现,以确保计数正确。
-
深入了解Netty编码器 - MessageToByte在Netty中的核心作用