Netty源码解析之编解码

编解码

为什么需要编解码？

• 数据在网络中以流的形式传递，在NIO中，读写的都是ByteBuffer
• socket数据传输的时候会发生粘包、半包等情况
• 通信双方的编解码格式预先定义，一定是一致的
• 业务代码与编解码逻辑需要解耦

编解码流程

• 首先，Client组装Request数据，发送到Server 端
• Client端首先对Request进行编码，比如使用JDK序列化成字节数组；【MessageToMessageEncoder】
• 字节数组不能直接被Channel读写，因此需要再次编码成ByteBuffer对象；这里用到【MessageToByteEncoder】
• 数据写入流，Server端需要通过Channel读取数据使用到的是ByteBuffer对象，但是socket会发生粘包、半包等
• 这个时候，Server端首先需要进行拆包，把一个未知的可能有多个包或者半包的ByteBuffer进行拆包，变成一个个独立的完整的数据包；【ByteToMessageDecoder】
• 拿到完整的数据包后，根据协议，解析为具体的Request对象；【MessageToMessageDecoder】

示例编码模式

• Object->MessageToMessageEncoder->(JSON String)->MessageToByteEncoder->(ByteBuf)->Socket->半包、粘包->ByteToMessageDecoder->(ByteBuf)->MessageToMessageDecoder->(JSON String)->Object

半包、粘包解决

• 发生半包、粘包是接收方需要处理的，因此需要用到的是【ByteToMessageDecoder】
• 一般的解决方案包括：消息定长、添加长度字段、分隔符等
• Netty提供了通用的解决方案：FixedLengthFrameDecoder、LengthFieldBasedFrameDecoder、DelimiterBasedFrameDecoder等

编码器

MessageToMessageEncoder

public abstract class MessageToMessageEncoder<I> extends ChannelOutboundHandlerAdapter{
}

• 消息到消息的编码器，比如java对象转换为JSON字符串
• 带泛型，只处理泛型类型的解码
• 自动释放资源

示例：StringEncoder

@Sharable
public class StringEncoder extends MessageToMessageEncoder<CharSequence> {

    private final Charset charset;

    public StringEncoder() {
        this(Charset.defaultCharset());
    }

    public StringEncoder(Charset charset) {
        if (charset == null) {
            throw new NullPointerException("charset");
        }
        this.charset = charset;
    }

    @Override
    protected void encode(ChannelHandlerContext ctx, CharSequence msg, List<Object> out) throws Exception {
        if (msg.length() == 0) {
            return;
        }

        out.add(ByteBufUtil.encodeString(ctx.alloc(), CharBuffer.wrap(msg), charset));
    }
}

• 可共享
• 处理CharSequence类型编码

MessageToByteEncoder

public abstract class MessageToByteEncoder<I> extends ChannelOutboundHandlerAdapter {
  private final TypeParameterMatcher matcher;
  private final boolean preferDirect;
}

• 其他消息类型到ByteBuf编码器
• 出站处理器，主要进行写数据
• matcher为类型参数匹配器，匹配泛型I和write方法的msg的类型，匹配的才处理。如子类指定泛型为String，那么这个类只会对msg类型为String的编码
• preferDirect指定写入消息的ByteBuf使用Direct还是Heap的ByteBuffer
• 自动释放资源

示例：ObjectEncoder

package io.netty.handler.codec.serialization;

@Sharable
public class ObjectEncoder extends MessageToByteEncoder<Serializable> {
    private static final byte[] LENGTH_PLACEHOLDER = new byte[4];

    @Override
    protected void encode(ChannelHandlerContext ctx, Serializable msg, ByteBuf out) throws Exception {
        int startIdx = out.writerIndex();

        ByteBufOutputStream bout = new ByteBufOutputStream(out);
        ObjectOutputStream oout = null;
        try {
            bout.write(LENGTH_PLACEHOLDER);
            oout = new CompactObjectOutputStream(bout);
            oout.writeObject(msg);
            oout.flush();
        } finally {
            if (oout != null) {
                oout.close();
            } else {
                bout.close();
            }
        }

        int endIdx = out.writerIndex();

        out.setInt(startIdx, endIdx - startIdx - 4);
    }
}

• 可共享的
• 使用write的对象是Serializable子类都可以被编码，通过JDK的序列化编码对象

编码总结

• 业务代码写的时候的数据通常是java对象
• 首先需要对java对象进行序列化，比如编程JDK序列化数组，JSON，XML等格式，使用MessageToMessageEncoder
• 序列化后的数据编程统一的格式，比如字节数组，字符串等，此时进一步编码为ByteBuf即可进行数据发送MessageToByteEncoder

解码器

ByteToMessageDecoder

public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter{
      protected ByteToMessageDecoder() {
        ensureNotSharable();
    }
}

• 非共享模式
• ByteBuf到其他类型消息解码器
• 入站处理器，主要进行数据读取
• 把流数据转换为其他消息类型
• Cumulator 累加器，把新的数据in与老的数据cumulation进行累计，处理半包读
  • MERGE_CUMULATOR 把in合并到cumulation
  • COMPOSITE_CUMULATOR cumulation应用>1还是合并，否则使用CompositeByteBuf接受多个ByteBuf

    public interface Cumulator {
        ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in);
    }

• CodecOutputList 当成List用就对了
• channelRead方法把新读取的数据和遗留数据通过Cumulator合并，调用decode方法进行解码，解码后可能包含的是多个完整消息，因此使用List封装，也就是CodecOutputList。然后循环当前list，依次传递每一个消息给下一个处理器，因此后续处理器只需要关注单个消息。

示例：FixedLengthFrameDecoder

package io.netty.handler.codec;

public class FixedLengthFrameDecoder extends ByteToMessageDecoder {

    private final int frameLength;

    public FixedLengthFrameDecoder(int frameLength) {
        checkPositive(frameLength, "frameLength");
        this.frameLength = frameLength;
    }

    @Override
    protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        Object decoded = decode(ctx, in);
        if (decoded != null) {
            out.add(decoded);
        }
    }

    protected Object decode(
            @SuppressWarnings("UnusedParameters") ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        if (in.readableBytes() < frameLength) {
            return null;
        } else {
            return in.readRetainedSlice(frameLength);
        }
    }
}

• 固定长度解码器
• 构造时指定消息长度
• 把指定长度的消息封装到一个ByteBuf，所以下一个处理器拿到的是一个个的ByteBuf

MessageToMessageDecoder

package io.netty.handler.codec;

public abstract class MessageToMessageDecoder<I> extends ChannelInboundHandlerAdapter {

    private final TypeParameterMatcher matcher;

}

• 消息到消息的解码器
• 带泛型，只处理泛型类型的解码
• 自动释放资源

示例：StringDecoder

package io.netty.handler.codec.string;

@Sharable
public class StringDecoder extends MessageToMessageDecoder<ByteBuf> {

    private final Charset charset;

    public StringDecoder() {
        this(Charset.defaultCharset());
    }

    public StringDecoder(Charset charset) {
        if (charset == null) {
            throw new NullPointerException("charset");
        }
        this.charset = charset;
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
        out.add(msg.toString(charset));
    }
}

• 把一个ByteBuf的全部内容解析为一个字符串

解码总结

• Channel收到的数据以流的形式存在，会存在半包，粘包
• 先通过ByteToMessageDecoder完成拆包，得到一个个完整的ByteBuf
• 然后使用MessageToMessageDecoder解析完整的ByteBuf得到解码后的类型
• 后续处理器的channelRead方法中的msg就是解码后的类型