概述
网络中数据是以二进制字节流进行传输，编码器的作用是将数据编码为二进制字节流，而解码器的作用是将二进制字节流解码为程序能处理的数据格式，本章节将对Netty编解码器实现原理进行分析。
编码器
Netty编码器有两个抽象基类：MessageToByteEncoder与MessageToMessageEncoder，MessageToByteEncoder会将消息编码为字节，MessageToMessageEncode会将消息编码为另一种消息。通过继承这两个基类可以实现自定义编码器，如Netty内置的编码器：StringEncoder、ObjectEncoder、ProtobufEncoder等。
从类图可以看出MessageToByteEncoder与MessageToMessageEncoder均实现了ChannelOutboundHandler接口，也就是说编码器是一个出站事件处理器，其会处理并传播出站事件。以MessageToByteEncoder为例，分析下其实现原理：
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
ByteBuf buf = null;
try {
// 判断编码器是否能处理该消息
if (acceptOutboundMessage(msg)) {
@SuppressWarnings("unchecked")
// 将Object消息类型转换为实际类型
I cast = (I) msg;
// 分配内存
buf = allocateBuffer(ctx, cast, preferDirect);
try {
// 调用子类方法进行编码
encode(ctx, cast, buf);
} finally {
// 如果消息是ByteBuf类型则释放内存
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方法首先会调用acceptOutboundMessage方法判断是否能处理传入的消息，如果不能处理则将该事件向下传播。acceptOutboundMessage通过TypeParameterMatcher#match方法来进行判断。其中TypeParameterMatcher是在实例化编码器时初始化的，具体判断方法在其内部类ReflectiveMatcher中，其中type为编码器中的泛型：
private static final class ReflectiveMatcher extends TypeParameterMatcher {
private final Class<?> type;
ReflectiveMatcher(Class<?> type) {
this.type = type;
}
@Override
public boolean match(Object msg) {
return type.isInstance(msg);
}
}
编码器的核心方法为encode，该方法会将消息编码为ByteBuf。基类MessageToMessageEncode中的encode方法为抽象方法，这里使用了模版方法设计模式，需要由子类实现具体的编码逻辑：
protected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception;
以ObjectEncoder为例，看下ObjectEncoder这个编码器是怎么编码的：
protected void encode(ChannelHandlerContext ctx, Serializable msg, ByteBuf out) throws Exception {
// 获取ByteBuf对象写指针索引位置
int startIdx = out.writerIndex();
// 创建输出流
ByteBufOutputStream bout = new ByteBufOutputStream(out);
// 写入4个字节进行占位
bout.write(LENGTH_PLACEHOLDER);
ObjectOutputStream oout = new CompactObjectOutputStream(bout);
// 将消息写入到输出流
oout.writeObject(msg);
oout.flush();
oout.close();
int endIdx = out.writerIndex();
// 在输入流开始位置写入int类型的消息长度，即前四个字节
out.setInt(startIdx, endIdx - startIdx - 4);
}
解码器
Netty解码器同样有两个抽象基类：MessageToByteDecoder与MessageToMessageDecoder，MessageToByteDecoder会将字节解码为消息，MessageToMessageDecoder会将消息解码为另一种消息。以MessageToByteDecoder为例，看下具体实现：
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if (msg instanceof ByteBuf) {
// 获取CodecOutputList，用于存放解码后的消息
CodecOutputList out = CodecOutputList.newInstance();
try {
ByteBuf data = (ByteBuf) msg;
first = cumulation == null;
if (first) {
// 第一次解码将数据赋值给cumulation
cumulation = data;
} else {
// 不是第一次解码则将数据进行累加到cumulation
cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
}
// 解码
callDecode(ctx, cumulation, out);
} catch (DecoderException e) {
throw e;
} catch (Throwable t) {
throw new DecoderException(t);
} finally {
// 如果cumulation不可读则，则释放cumulation
if (cumulation != null && !cumulation.isReadable()) {
numReads = 0;
cumulation.release();
cumulation = null;
} else if (++ numReads >= discardAfterReads) {
// 如果读次数超过16，则尝试丢弃一些字节
// We did enough reads already try to discard some bytes so we not risk to see a OOME.
// See https://github.com/netty/netty/issues/4275
numReads = 0;
discardSomeReadBytes();
}
int size = out.size();
decodeWasNull = !out.insertSinceRecycled();
// 传播事件
fireChannelRead(ctx, out, size);
out.recycle();
}
} else {
ctx.fireChannelRead(msg);
}
}
CodecOutputList是一个可复用的List，内部维护了一个长度为16的对象数组，用来缓存解码后的消息。CodecOutputList#newInstance方法首先会从对象池中获取，如果获取不到会进行创建。
读取到消息后，首先会将其放入到cumulation这个ByteBuf中，然后尝试解码，每次解码成功就会传播channelRead事件，将消息向下传播。如果消息不能解码，比如出现半包，或者出现粘包（一些字节不足以解码为一条消息），则会被累加到cumulation中等待下次接收到消息后处理。
解码的核心逻辑在callDecode中，该方法会将cumulation中累加的消息进行解码：
protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
try {
while (in.isReadable()) {
int outSize = out.size();
if (outSize > 0) {
fireChannelRead(ctx, out, outSize);
out.clear();
// Check if this handler was removed before continuing with decoding.
// If it was removed, it is not safe to continue to operate on the buffer.
//
// See:
// - https://github.com/netty/netty/issues/4635
if (ctx.isRemoved()) {
break;
}
outSize = 0;
}
int oldInputLength = in.readableBytes();
decode(ctx, in, out);
// Check if this handler was removed before continuing the loop.
// If it was removed, it is not safe to continue to operate on the buffer.
//
// See https://github.com/netty/netty/issues/1664
if (ctx.isRemoved()) {
break;
}
if (outSize == out.size()) {
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 (Throwable cause) {
throw new DecoderException(cause);
}
}
其中decode是一个模版方法，需要由子类解码器实现具体的解码逻辑，解码后的消息会放入到CodecOutputList中。
解码后会调用fireChannelRead方法传播事件，该方法内部会遍历CodecOutputList集合中解码后的消息，将消息传播给下一个处理器：
static void fireChannelRead(ChannelHandlerContext ctx, CodecOutputList msgs, int numElements) {
for (int i = 0; i < numElements; i ++) {
ctx.fireChannelRead(msgs.getUnsafe(i));
}
}
