如何安全替换Netty中的ByteToMessageDecoder避免消息丢失

解码器替换阶段的消息流转规则 #

• 不存在消息被直接丢弃的情况，但操作不当必然出现字节丢失、流错乱问题：
  • 已经被旧解码器读取到内部半包累积缓冲区、还没完成解码的字节，若替换时未手动转存，会随旧解码器实例被回收，直接导致Content字段截断
  • 还停留在Socket接收缓冲区、尚未进入Netty Pipeline处理流程的字节，会在解码器替换完成后正常交由新解码器处理，不会丢失
  • 若在Channel对应的EventLoop（IO线程）之外执行解码器替换操作，会触发Pipeline并发读写异常，导致整条连接的字节流解析错乱

原有方案失效的核心原因 #

你之前参考的动态替换方案存在两个硬伤：

• 没有处理旧解码器内部残留的累积字节：LengthFieldBasedFrameDecoder自带字节累积区，当你解析完Header的瞬间，大概率已经有部分Content分片字节被读入这个累积区，直接移除旧解码器就会丢失这部分数据
• 协议匹配度差：你的协议中Content段没有长度前缀，是靠末尾###end###分隔符标识结束，本身就不是纯长度帧结构，硬拆成两个独立解码器拆分解析，很容易出现边界判定错误

可落地的实现方案 #

方案1：单解码器状态机解析（优先推荐，无替换风险） #

不需要做Pipeline动态替换，直接自定义继承ByteToMessageDecoder的解码器，内部维护解析状态机分阶段处理字节流，是Netty自定义私有协议的标准实现方式，完全规避动态替换Handler的所有风险：

• 第一阶段：读取4字节的Header长度字段，累积字节直到满足Header长度要求，解析出完整Header存入上下文
• 第二阶段：切换到Content解析状态，持续累积输入字节，每次写入后扫描缓冲区是否存在###end###分隔符
• 第三阶段：定位到分隔符后，提取分隔符前的所有字节作为完整Content，和Header组装成业务消息对象向后传递，重置解码器状态等待下一条消息
  核心实现参考：

public class CustomProtocolDecoder extends ByteToMessageDecoder {
    private enum DecodeState {READ_HEADER_LEN, READ_HEADER, READ_CONTENT}
    private DecodeState currentState = DecodeState.READ_HEADER_LEN;
    private int headerLength;
    private String header;
    private final ByteBuf contentBuffer = Unpooled.buffer();
    private static final byte[] END_DELIMITER = "###end###".getBytes(StandardCharsets.UTF_8);

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // 读取头长度
        if (currentState == DecodeState.READ_HEADER_LEN) {
            if (in.readableBytes() < Integer.BYTES) return;
            headerLength = in.readInt();
            currentState = DecodeState.READ_HEADER;
        }
        // 读取完整头
        if (currentState == DecodeState.READ_HEADER) {
            if (in.readableBytes() < headerLength) return;
            byte[] headerBytes = new byte[headerLength];
            in.readBytes(headerBytes);
            header = new String(headerBytes, StandardCharsets.UTF_8);
            currentState = DecodeState.READ_CONTENT;
        }
        // 读取内容直到遇到结束符
        if (currentState == DecodeState.READ_CONTENT) {
            while (in.isReadable()) {
                contentBuffer.writeByte(in.readByte());
                int delimiterPos = ByteBufUtil.indexOf(Unpooled.wrappedBuffer(END_DELIMITER), contentBuffer);
                if (delimiterPos != -1) {
                    // 提取内容，剔除末尾分隔符
                    byte[] contentBytes = new byte[delimiterPos];
                    contentBuffer.getBytes(0, contentBytes);
                    String content = new String(contentBytes, StandardCharsets.UTF_8);
                    // 组装消息向后传递
                    out.add(new CustomProtocolMessage(header, content));
                    // 重置状态处理下一条消息
                    resetState();
                    return;
                }
            }
        }
    }

    private void resetState() {
        currentState = DecodeState.READ_HEADER_LEN;
        headerLength = 0;
        header = null;
        contentBuffer.clear();
    }
}

方案2：必须动态替换解码器的正确操作规范 #

如果业务场景强制要求拆分两个解码器做动态替换，必须严格遵守两个规则：

1. 所有Pipeline的Handler增删操作，必须提交到对应Channel的EventLoop线程执行，禁止跨线程操作
2. 移除旧解码器前，必须把旧解码器内部累积区中未处理的残留字节全部取出，手动注入到新解码器的累积区后，再完成替换
   替换逻辑参考：

// 旧解码器解析完Header后，提交替换任务到IO线程
ctx.channel().eventLoop().execute(() -> {
    // 取出旧解码器内部残留的未处理字节（需要自行在子类中暴露获取内部累积区的方法）
    ByteBuf remainingBytes = ((HeaderDecoder) ctx.pipeline().get("headerDecoder")).getRemainingBytes();
    // 移除旧解码器
    ctx.pipeline().remove("headerDecoder");
    // 初始化新的内容解码器
    ContentDecoder contentDecoder = new ContentDecoder();
    // 把新解码器加入Pipeline
    ctx.pipeline().addBefore("businessHandler", "contentDecoder", contentDecoder);
    // 先把残留字节喂给新解码器，再处理后续新到的字节
    contentDecoder.appendCumulation(ctx.alloc().buffer(), remainingBytes);
});

注意：LengthFieldBasedFrameDecoder没有对外暴露内部累积缓冲区的获取方法，需要自行继承扩展，否则依然会丢失残留字节。

内容的提问来源于stack exchange，提问作者william