当前位置： 首页 > news >正文

[netty5: ByteToMessageCodec MessageToByteEncoder ByteToMessageDecoder]-源码分析

news 2025/7/5 9:44:50

ByteToMessageCodec

ByteToMessageCodec 是一个结合了 ByteToMessageDecoder 和 MessageToByteEncoder 的编解码器，可以实时地将字节流编码或解码为消息，反之亦然。

public abstract class ByteToMessageCodec<I> extends ChannelHandlerAdapter {private final TypeParameterMatcher outboundMsgMatcher;private final MessageToByteEncoder<I> encoder;private final ByteToMessageDecoder decoder = new ByteToMessageDecoder() {@Overridepublic void decode(ChannelHandlerContext ctx, Buffer in) throws Exception {ByteToMessageCodec.this.decode(ctx, in);}@Overrideprotected void decodeLast(ChannelHandlerContext ctx, Buffer in) throws Exception {ByteToMessageCodec.this.decodeLast(ctx, in);}};protected ByteToMessageCodec() {this((BufferAllocator) null);}protected ByteToMessageCodec(Class<? extends I> outboundMessageType) {this(outboundMessageType, null);}protected ByteToMessageCodec(BufferAllocator allocator) {outboundMsgMatcher = TypeParameterMatcher.find(this, ByteToMessageCodec.class, "I");encoder = new Encoder(allocator);}protected ByteToMessageCodec(Class<? extends I> outboundMessageType, BufferAllocator allocator) {outboundMsgMatcher = TypeParameterMatcher.get(outboundMessageType);encoder = new Encoder(allocator);}@Overridepublic final boolean isSharable() {return false;}public boolean acceptOutboundMessage(Object msg) throws Exception {return outboundMsgMatcher.match(msg);}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {decoder.channelRead(ctx, msg);}@Overridepublic Future<Void> write(ChannelHandlerContext ctx, Object msg) {return encoder.write(ctx, msg);}@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {decoder.channelReadComplete(ctx);}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {decoder.channelInactive(ctx);}@Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {try {decoder.handlerAdded(ctx);} finally {encoder.handlerAdded(ctx);}}@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) throws Exception {try {decoder.handlerRemoved(ctx);} finally {encoder.handlerRemoved(ctx);}}protected abstract void encode(ChannelHandlerContext ctx, I msg, Buffer out) throws Exception;protected abstract void decode(ChannelHandlerContext ctx, Buffer in) throws Exception;protected void decodeLast(ChannelHandlerContext ctx, Buffer in) throws Exception {if (in.readableBytes() > 0) {decode(ctx, in);}}private final class Encoder extends MessageToByteEncoder<I> {private final BufferAllocator allocator;Encoder(BufferAllocator allocator) {this.allocator = allocator;}@Overridepublic boolean acceptOutboundMessage(Object msg) throws Exception {return ByteToMessageCodec.this.acceptOutboundMessage(msg);}@Overrideprotected Buffer allocateBuffer(ChannelHandlerContext ctx, I msg) throws Exception {BufferAllocator alloc = allocator != null? allocator : ctx.bufferAllocator();return alloc.allocate(256);}@Overrideprotected void encode(ChannelHandlerContext ctx, I msg, Buffer out) throws Exception {ByteToMessageCodec.this.encode(ctx, msg, out);}}
}

MessageToByteEncoder

MessageToByteEncoder 是一个用于将消息编码为字节流的抽象类。它继承自 ChannelHandlerAdapter，并通过两个抽象方法来实现编码功能：allocateBuffer 用于分配一个 Buffer，encode 用于将消息编码到 Buffer 中。它通过 acceptOutboundMessage 方法决定是否处理给定的消息类型，并在 write 方法中执行编码操作。

public abstract class MessageToByteEncoder<I> extends ChannelHandlerAdapter {private final TypeParameterMatcher matcher;protected MessageToByteEncoder() {matcher = TypeParameterMatcher.find(this, MessageToByteEncoder.class, "I");}protected MessageToByteEncoder(Class<? extends I> outboundMessageType) {matcher = TypeParameterMatcher.get(requireNonNull(outboundMessageType, "outboundMessageType"));}public boolean acceptOutboundMessage(Object msg) throws Exception {return matcher.match(msg);}@Overridepublic Future<Void> write(ChannelHandlerContext ctx, Object msg) {Buffer buf = null;try {if (acceptOutboundMessage(msg)) {@SuppressWarnings("unchecked")I cast = (I) msg;buf = allocateBuffer(ctx, cast);try (AutoCloseable ignore = autoClosing(cast)) {encode(ctx, cast, buf);}if (buf.readableBytes() > 0) {Future<Void> f = ctx.write(buf);buf = null;return f;}return ctx.write(ctx.bufferAllocator().allocate(0));}return ctx.write(msg);} catch (EncoderException e) {return ctx.newFailedFuture(e);} catch (Throwable e) {return ctx.newFailedFuture(new EncoderException(e));} finally {if (buf != null) {buf.close();}}}protected abstract Buffer allocateBuffer(ChannelHandlerContext ctx, I msg) throws Exception;protected abstract void encode(ChannelHandlerContext ctx, I msg, Buffer out) throws Exception;
}

ByteToMessageDecoder

ByteToMessageDecoder 是 Netty 5 中用于处理 TCP 粘包拆包的解码器基础类。它通过维护一个累积缓冲区 cumulation，将接收到的 Buffer 连续拼接、缓存，并在适当时机调用 decode(…) 方法将字节流转为高层消息对象

public abstract class ByteToMessageDecoder extends ChannelHandlerAdapter {public static final Cumulator MERGE_CUMULATOR = new MergeCumulator();public static final Cumulator COMPOSITE_CUMULATOR = new CompositeBufferCumulator();private final int discardAfterReads = 16;private final Cumulator cumulator;// 累积的字节缓冲区，保存之前未解码完的字节private Buffer cumulation;// 是否每次只解码一条消息，默认关闭以提升性能，开启后适合协议升级场景private boolean singleDecode;// 标记是否是第一次解码调用private boolean first;// 标记本次 decode() 是否产出并传递了消息，决定是否需要继续读。private boolean firedChannelRead;// 标记当前的 channelRead 是 decoder 主动触发的，防止重复触发 read()。private boolean selfFiredChannelRead;// 统计读操作次数private int numReads;// 包装的上下文对象private ByteToMessageDecoderContext context;protected ByteToMessageDecoder() {this(MERGE_CUMULATOR);}protected ByteToMessageDecoder(Cumulator cumulator) {this.cumulator = requireNonNull(cumulator, "cumulator");}@Overridepublic final boolean isSharable() {// Can't be sharable as we keep state.return false;}public void setSingleDecode(boolean singleDecode) {this.singleDecode = singleDecode;}public boolean isSingleDecode() {return singleDecode;}protected int actualReadableBytes() {return internalBuffer().readableBytes();}protected Buffer internalBuffer() {return cumulation;}@Overridepublic final void handlerAdded(ChannelHandlerContext ctx) throws Exception {context = new ByteToMessageDecoderContext(ctx);handlerAdded0(context);}protected void handlerAdded0(ChannelHandlerContext ctx) throws Exception {}@Overridepublic final void handlerRemoved(ChannelHandlerContext ctx) throws Exception {Buffer buf = cumulation;if (buf != null) {cumulation = null;numReads = 0;int readable = buf.readableBytes();if (readable > 0) {ctx.fireChannelRead(buf);ctx.fireChannelReadComplete();} else {buf.close();}}handlerRemoved0(context);}protected void handlerRemoved0(ChannelHandlerContext ctx) throws Exception { }// 1. TCP 是流式协议，发送方即便发送了一整条消息，接收方可能会分多次读取数据// 2. 每次读取都触发channelRead，读取到的msg是Buffer类型，内容可能不完整，需要累积合并到 cumulation// 3. 每次读取都会触发callDecode，去尝试根据现有的数据，进行解码，如果成功则向调用链传递结果，否则啥也不干，等待下次 channelRead@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {if (msg instanceof Buffer) {selfFiredChannelRead = true;try {Buffer data = (Buffer) msg;first = cumulation == null;if (first) {cumulation = data;} else {cumulation = cumulator.cumulate(ctx.bufferAllocator(), cumulation, data);}assert context.delegatingCtx() == ctx || ctx == context;// 尝试解码callDecode(context, cumulation);} catch (DecoderException e) {throw e;} catch (Exception e) {throw new DecoderException(e);} finally {// 如果数据已读尽且无剩余，释放 bufferif (cumulation != null && cumulation.readableBytes() == 0) {numReads = 0;if (cumulation.isAccessible()) {cumulation.close();}cumulation = null;} else if (++numReads >= discardAfterReads) {// 如果累计读取次数到达阈值，主动丢弃已消费字节，防止 cumulation 越来越大numReads = 0;discardSomeReadBytes();}// 跟踪本次读数据是否至少成功解码过一次, 向下游传播了消息firedChannelRead |= context.fireChannelReadCallCount() > 0;// 状态清理context.reset();}} else {ctx.fireChannelRead(msg);}}// 在 channelRead 之后，执行 channelReadComplete@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {// 重置读取次数，清理 cumulation 的无效字节。numReads = 0;discardSomeReadBytes();// 解码还未成功过，自动读取关闭，在 channelRead完成后，执行本方法时，手动触发下一次读取if (selfFiredChannelRead && !firedChannelRead && !ctx.channel().getOption(ChannelOption.AUTO_READ)) {ctx.read(); }// 重置状态标志位并通知下游 handler。firedChannelRead = false;selfFiredChannelRead = false;ctx.fireChannelReadComplete();}protected final void discardSomeReadBytes() {// 丢弃 cumulation 中已消费的字节。只在存在历史累积（!first）时执行if (cumulation != null && !first) {cumulator.discardSomeReadBytes(cumulation);}}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {assert context.delegatingCtx() == ctx || ctx == context;channelInputClosed(context, true);}@Overridepublic void channelShutdown(ChannelHandlerContext ctx, ChannelShutdownDirection direction) throws Exception {ctx.fireChannelShutdown(direction);if (direction == ChannelShutdownDirection.Inbound) {assert context.delegatingCtx() == ctx || ctx == context;channelInputClosed(context, false);}}// Channel 被关闭时，尝试从 cumulation 中解码剩余数据，释放资源，并向下传递 fire* 事件private void channelInputClosed(ByteToMessageDecoderContext ctx, boolean callChannelInactive) {try {channelInputClosed(ctx);} catch (DecoderException e) {throw e;} catch (Exception e) {throw new DecoderException(e);} finally {if (cumulation != null) {cumulation.close();cumulation = null;}if (ctx.fireChannelReadCallCount() > 0) {ctx.reset();ctx.fireChannelReadComplete();}if (callChannelInactive) {ctx.fireChannelInactive();}}}void channelInputClosed(ByteToMessageDecoderContext ctx) throws Exception {if (cumulation != null) {callDecode(ctx, cumulation);if (!ctx.isRemoved()) {if (cumulation == null) {try (Buffer buffer = ctx.bufferAllocator().allocate(0)) {decodeLast(ctx, buffer);}} else {decodeLast(ctx, cumulation);}}} else {try (Buffer buffer = ctx.bufferAllocator().allocate(0)) {decodeLast(ctx, buffer);}}}void callDecode(ByteToMessageDecoderContext ctx, Buffer in) {try {// 当仍有未解码的数据 (in.readableBytes() > 0) 且当前 Handler 还在 pipeline 中（未被移除），就继续调用 decode() 试图解码更多消息。while (in.readableBytes() > 0 && !ctx.isRemoved()) {int oldInputLength = in.readableBytes();int numReadCalled = ctx.fireChannelReadCallCount();decodeRemovalReentryProtection(ctx, in);if (ctx.isRemoved()) {break;}if (numReadCalled == ctx.fireChannelReadCallCount()) {if (oldInputLength == in.readableBytes()) {break;} else {continue;}}if (oldInputLength == in.readableBytes()) {throw new DecoderException(StringUtil.simpleClassName(getClass()) +".decode() did not read anything but decoded a message.");}if (isSingleDecode()) {break;}}} catch (DecoderException e) {throw e;} catch (Exception cause) {throw new DecoderException(cause);}}protected void decodeLast(ChannelHandlerContext ctx, Buffer in) throws Exception {if (in.readableBytes() > 0) {decodeRemovalReentryProtection(ctx, in);}}final void decodeRemovalReentryProtection(ChannelHandlerContext ctx, Buffer in) throws Exception {decode(ctx, in);}// 建议结合 FixedLengthFrameDecoder阅读，理解decode方法做了什么protected abstract void decode(ChannelHandlerContext ctx, Buffer in) throws Exception;// ...
}

ByteToMessageDecoderContext

ByteToMessageDecoderContext 是 Netty 为解码器设计的包装上下文，用于统计 fireChannelRead 次数，以精确控制解码行为和数据流转。

static final class ByteToMessageDecoderContext extends DelegatingChannelHandlerContext {private int fireChannelReadCalled;private ByteToMessageDecoderContext(ChannelHandlerContext ctx) {super(ctx);}void reset() {fireChannelReadCalled = 0;}int fireChannelReadCallCount() {return fireChannelReadCalled;}@Overridepublic ChannelHandlerContext fireChannelRead(Object msg) {fireChannelReadCalled ++;super.fireChannelRead(msg);return this;}
}

Cumulator

Cumulator 接口用于处理和累积多个 Buffer 数据。它有两个主要方法，分别负责将多个 Buffer 合并成一个更大的 Buffer，并管理已经处理过的数据。

public interface Cumulator {// BufferAllocator alloc：缓冲区分配器，用于分配新的 Buffer// Buffer cumulation：当前的累积数据缓冲区// Buffer in：新的输入数据缓冲区// 将多个 Buffer 合并成一个新的 Buffer，即将当前 Buffer 中的可读数据与新的数据合并Buffer cumulate(BufferAllocator alloc, Buffer cumulation, Buffer in);// 丢弃缓冲区中已经读取的数据，返回一个新的缓冲区，去除了之前已处理过的部分Buffer discardSomeReadBytes(Buffer cumulation);
}
CompositeBufferCumulator

CompositeBufferCumulator 适用于大量 Buffer 分段输入时，通过复合缓冲区高效拼接，避免数据拷贝和缓冲区重分配。

private static final class CompositeBufferCumulator implements Cumulator {@Overridepublic Buffer cumulate(BufferAllocator alloc, Buffer cumulation, Buffer in) {// 如果累加缓冲区 cumulation 没有可读字节了，直接释放并返回输入缓冲区 inif (cumulation.readableBytes() == 0) {cumulation.close();return in;}try (in) {// 输入缓冲区 in 没有可读字节，保留原来的 cumulation。if (in.readableBytes() == 0) {return cumulation;}// 如果累加缓冲区是只读的，复制出一个可写副本if (cumulation.readOnly()) {Buffer tmp = cumulation.copy();cumulation.close();cumulation = tmp;}// 如果当前累加区已经是 CompositeBuffer，直接扩展进去if (CompositeBuffer.isComposite(cumulation)) {CompositeBuffer composite = (CompositeBuffer) cumulation;composite.extendWith(prepareInForCompose(in));return composite;}return alloc.compose(Arrays.asList(cumulation.send(), prepareInForCompose(in)));}}// 确保 in 是只读安全的后，调用 send() 移交所有权用于合并。private static Send<Buffer> prepareInForCompose(Buffer in) {return in.readOnly() ? in.copy().send() : in.send();}// 通过 readSplit(0) 剪掉已经读过的部分，释放空间。相比 compact() 更适合复合缓冲区（CompositeBuffer）@Overridepublic Buffer discardSomeReadBytes(Buffer cumulation) {cumulation.readSplit(0).close();return cumulation;}@Overridepublic String toString() {return "CompositeBufferCumulator";}
}
MergeCumulator

MergeCumulator 更适合频繁接收小块数据的场景，追求高访问性能和代码简单；但当数据量变大时，可能因为频繁扩容带来复制开销。

    private static final class MergeCumulator implements Cumulator {@Overridepublic Buffer cumulate(BufferAllocator alloc, Buffer cumulation, Buffer in) {// 如果当前累积的 Buffer 为空if (cumulation.readableBytes() == 0) {cumulation.close();return in;}try (in) {final int required = in.readableBytes();// 如果当前的累积 Buffer 没有足够的空间，或者它是只读，就扩展大小。if (required > cumulation.writableBytes() || cumulation.readOnly()) {return expandCumulationAndWrite(alloc, cumulation, in);}cumulation.writeBytes(in);return cumulation;}}// 如果 cumulation 中的已读字节数超过可写字节数（readerOffset() > writableBytes()），则调用 compact() 方法来压缩缓冲区，移除已读的数据@Overridepublic Buffer discardSomeReadBytes(Buffer cumulation) {if (cumulation.readerOffset() > cumulation.writableBytes()) {cumulation.compact();}return cumulation;}private static Buffer expandCumulationAndWrite(BufferAllocator alloc, Buffer oldCumulation, Buffer in) {// 1. 计算新的 Buffer 大小：计算新的 Buffer 大小，确保它能够容纳当前的 cumulation 和输入的 Buffer。新的大小是当前已读字节和输入字节总和的下一个最接近的 2 的幂次方。final int newSize = safeFindNextPositivePowerOfTwo(oldCumulation.readableBytes() + in.readableBytes());// 创建新的 Buffer：根据是否是只读的 Buffer 来决定如何创建新的 Buffer。如果是只读的，会重新分配内存；否则，直接扩展当前的 Buffer。Buffer newCumulation = oldCumulation.readOnly() ? alloc.allocate(newSize) :oldCumulation.ensureWritable(newSize);// 3. 将旧数据和新数据写入新 Buffer：如果创建了新的 Buffer，将旧的 cumulation 和输入的 Buffer 都写入新的 Buffer。try {if (newCumulation != oldCumulation) {newCumulation.writeBytes(oldCumulation);}newCumulation.writeBytes(in);return newCumulation;} finally {// 4. 关闭旧的 Buffer：如果创建了新的 Buffer，则关闭原先的 cumulation，释放内存。if (newCumulation != oldCumulation) {oldCumulation.close();}}}@Overridepublic String toString() {return "MergeCumulator";}}
}

FixedLengthFrameDecoder

在此引入这个类，只是因为这个类最简单，方便大家理解 解码器的工作流程。

  1. FixedLengthFrameDecoder：按照固定的字节长度切割每一帧，适用于每条消息长度一致的协议（如定长二进制协议）。
  2. LengthFieldBasedFrameDecoder：根据消息中指定位置的“长度字段”值来动态切割完整帧，适用于二进制协议。
  3. LineBasedFrameDecoder：以 \n 或 \r\n 为分隔符，将每行作为一帧，适用于基于文本的行协议。
  4. DelimiterBasedFrameDecoder：使用自定义的分隔符（如 $_、#END#）拆分帧，适用于定界符结束的文本协议。
public class FixedLengthFrameDecoder extends ByteToMessageDecoder {private final int frameLength;public FixedLengthFrameDecoder(int frameLength) {checkPositive(frameLength, "frameLength");this.frameLength = frameLength;}public FixedLengthFrameDecoder(int frameLength, Cumulator cumulator) {super(cumulator);checkPositive(frameLength, "frameLength");this.frameLength = frameLength;}// ByteToMessageDecoder(channelRead[首次读] -> callDecode[循环读取数据]// ->// decodeRemovalReentryProtection) // -> // FixedLengthFrameDecoder.decode@Overrideprotected final void decode(ChannelHandlerContext ctx, Buffer in) throws Exception {Object decoded = decode0(ctx, in);if (decoded != null) {// 只要解码成功，就向下传递结果// ByteToMessageDecoderContext.fireChannelReadctx.fireChannelRead(decoded);}}protected Object decode0(@SuppressWarnings("UnusedParameters") ChannelHandlerContext ctx, Buffer in) throws Exception {if (in.readableBytes() < frameLength) {return null;} else {return in.readSplit(frameLength);}}
}
http://www.lryc.cn/news/579975.html

相关文章：

  • CCViM Block（上下文聚类视觉曼巴模块），通过多方向扫描（水平 / 垂直 / 翻转）提取目标延展特征，结合聚类层对边界点的动态聚合，提升目标的定位能力
  • Python爬虫 模拟登录状态 requests版
  • Vue2中的keep-alive：组件状态缓存与性能优化实战指南
  • Linux 如何上传本地文件以及下载文件到本地命令总结
  • Linux探秘坊-------13.进程间通信
  • 五、Flutter动画
  • 【AI总结】Git vs GitHub vs GitLab：深度解析三者联系与核心区别
  • 【Git】git命令合集
  • 网安系列【4】之OWASP与OWASP Top 10：Web安全入门指南
  • Rust 闭包
  • 暴雨服务器成功中标华中科技大学集成电路学院服务器采购项目
  • 封装一个png的编码解码操作
  • 数据库位函数：原理、应用与性能优化
  • 企业该怎么做竞争分析？一文了解
  • Linux-进程概念（3）
  • 【WEB】Polar靶场 6-10题 详细笔记
  • 类图+案例+代码详解：软件设计模式----原型模式
  • vue3 el-table 行筛选 设置为单选
  • 电商分拣的“效率密码”：艾立泰轻量化托盘引领自动化流水线革新
  • vue3 获取选中的el-table行数据
  • 【WRFDA第三期】OBSPROC namelist 变量总结
  • Ubuntu 22.04 + MySQL 8 无密码登录问题与 root 密码重置指南
  • OpenCV中DPM（Deformable Part Model）目标检测类cv::dpm::DPMDetector
  • 前端基础知识Webpack系列 - 03（webpack中常见的Loader？解决了什么问题？）
  • STM32CubeMX教程1 实现点灯点灯
  • 量化开发（系列第3篇）： C++在高性能量化交易中的核心应用与技术栈深度解析
  • 三态逻辑详解：单片机GPIO、计算机总线系统举例
  • 【python实用小脚本-128】基于 Python 的 Hacker News 爬虫工具：自动化抓取新闻数据
  • RK-Android11-性能优化-限制App内存上限默认512m
  • 基于Hadoop的公共自行车数据分布式存储和计算平台的设计与实现