1. 概述
所谓解码就是将一串二进制数据流解析成一个个自定义协议的数据包，也就是ByteBuf。后续业务就可以直接基于ByteBuf进行处理。
两个问题
- 解码器抽象的解码过程
- netty里面有哪些拆箱即用的解码器
主要内容
- 解码器基类
- netty中常见的解码器
2. 抽象解码器ByteToMessageDecoder
ByteToMessageDecoder解码步骤
- 累加字节流
- 调用子类的decode方法进行解析
- 将解析到的ByteBuf向下传播
具体分析
- 累加字节流
ByteToMessageDecoder是基于ByteBuf进行解码的，ByteToMessageDecoder的channelRead方法：
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if (msg instanceof ByteBuf) { // 先判断是不是ByteBuf，如果是ByteBuf进行解码器的处理
CodecOutputList out = CodecOutputList.newInstance(); // 存放解析完之后的对象
try {
ByteBuf data = (ByteBuf) msg;
first = cumulation == null; // 如果cumulation == null 说明累加器中没数据
if (first) {
cumulation = data; // 累加器为null， 将ByteBuf对象赋值给累加器
} else { // 将累加器中的数据和读进来的数据累加，然后赋值给累加器
cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
}
callDecode(ctx, cumulation, out); // 调用子类decode方法进行解析
} catch (DecoderException e) {
throw e;
} catch (Exception e) {
throw new DecoderException(e);
} finally {
if (cumulation != null && !cumulation.isReadable()) {
numReads = 0;
cumulation.release();
cumulation = null;
} else if (++ numReads >= discardAfterReads) {
// 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 {// 如果不是ByteBuf类型，直接将其向下传播
ctx.fireChannelRead(msg);
}
}
我们来看cumulator是什么：
public static final Cumulator MERGE_CUMULATOR = new Cumulator() {
@Override
public ByteBuf cumulate(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf in) {
try {
final ByteBuf buffer;
if (cumulation.writerIndex() > cumulation.maxCapacity() - in.readableBytes()
|| cumulation.refCnt() > 1 || cumulation.isReadOnly()) {
// Expand cumulation (by replace it) when either there is not more room in the buffer
// or if the refCnt is greater then 1 which may happen when the user use slice().retain() or
// duplicate().retain() or if its read-only.
//
// See:
// - https://github.com/netty/netty/issues/2327
// - https://github.com/netty/netty/issues/1764
buffer = expandCumulation(alloc, cumulation, in.readableBytes());
} else {
buffer = cumulation;
}
buffer.writeBytes(in); // 把当前数据写入累加器
return buffer;
} finally {
// We must release in in all cases as otherwise it may produce a leak if writeBytes(...) throw
// for whatever release (for example because of OutOfMemoryError)
in.release(); // 读进来的数据进行释放
}
}
};
cumulation.writerIndex() > cumulation.maxCapacity() - in.readableBytes() 判断当前累加器是否有足够的空间，如果空间不够就进行扩容。
- 调用子类decode
接下来看allDecode(ctx, cumulation, out);方法:
protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
try {
while (in.isReadable()) { // 判断累加器中是否有可读字节
int outSize = out.size();
if (outSize > 0) { // 看outList中是否有对象，只要有对象就使用事件传播机制向下传播
fireChannelRead(ctx, out, outSize);
out.clear(); // 然后当前的list进行清空
// 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之前，将当前的可读字节记录下来
decodeRemovalReentryProtection(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
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);
}
}
- 将子类解析到的ByteBuf向下传播
fireChannelRead(ctx, out, size);
/**
* Get {@code numElements} out of the {@link CodecOutputList} and forward these through the pipeline.
*/
static void fireChannelRead(ChannelHandlerContext ctx, CodecOutputList msgs, int numElements) {
for (int i = 0; i < numElements; i ++) { // 将解析到的每个对象向下传播
ctx.fireChannelRead(msgs.getUnsafe(i)); // 这就最终传播到业务解码器中，
}
}
3. 基于固定长度解码器 FixedLengthFrameDecoder 的分析
/**
* A decoder that splits the received {@link ByteBuf}s by the fixed number
* of bytes. For example, if you received the following four fragmented packets:
* <pre>
* +---+----+------+----+
* | A | BC | DEFG | HI |
* +---+----+------+----+
* </pre>
* A {@link FixedLengthFrameDecoder}{@code (3)} will decode them into the
* following three packets with the fixed length:
* <pre>
* +-----+-----+-----+
* | ABC | DEF | GHI |
* +-----+-----+-----+
* </pre>
*/
public class FixedLengthFrameDecoder extends ByteToMessageDecoder {
private final int frameLength; // 固定长度
/**
* Creates a new instance.
*
* @param frameLength the length of the frame
*/
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);
}
}
/**
* Create a frame out of the {@link ByteBuf} and return it.
*
* @param ctx the {@link ChannelHandlerContext} which this {@link ByteToMessageDecoder} belongs to
* @param in the {@link ByteBuf} from which to read data
* @return frame the {@link ByteBuf} which represent the frame or {@code null} if no frame could
* be created.
*/
protected Object decode(
@SuppressWarnings("UnusedParameters") ChannelHandlerContext ctx, ByteBuf in) throws Exception {
if (in.readableBytes() < frameLength) { // 小于固定长度，不解析
return null;
} else {
return in.readRetainedSlice(frameLength); // 从当前累加器中截取frmeLength的字节数据
}
}
}
4. 基于行解码器 LineBasedFrameDecoder 的分析
行解码器，如果字节流是以\r\n结尾或者是以\n结尾的字节流，行解码器就是以换行符为分隔，将字节流解析成一个个完整的数据包。
public class LineBasedFrameDecoder extends ByteToMessageDecoder {
/** Maximum length of a frame we're willing to decode. */
private final int maxLength; // 行解码器解析数据包的最大长度，超过最大长度，可能处理丢弃模式
/** Whether or not to throw an exception as soon as we exceed maxLength. */
private final boolean failFast; // 超过最大长度是否立即抛出异常，true表示立即抛出
private final boolean stripDelimiter; // 最终解析出的数据包带不带换行符，如果为true表示不带
/** True if we're discarding input because we're already over maxLength. */
private boolean discarding; // 数据流过长就会开启丢弃模式
private int discardedBytes; // 解码到现在当前已经丢弃的字节
/** Last scan position. */
private int offset;
/**
* Creates a new decoder.
* @param maxLength the maximum length of the decoded frame.
* A {@link TooLongFrameException} is thrown if
* the length of the frame exceeds this value.
*/
public LineBasedFrameDecoder(final int maxLength) {
this(maxLength, true, false);
}
/**
* Creates a new decoder.
* @param maxLength the maximum length of the decoded frame.
* A {@link TooLongFrameException} is thrown if
* the length of the frame exceeds this value.
* @param stripDelimiter whether the decoded frame should strip out the
* delimiter or not
* @param failFast If <tt>true</tt>, a {@link TooLongFrameException} is
* thrown as soon as the decoder notices the length of the
* frame will exceed <tt>maxFrameLength</tt> regardless of
* whether the entire frame has been read.
* If <tt>false</tt>, a {@link TooLongFrameException} is
* thrown after the entire frame that exceeds
* <tt>maxFrameLength</tt> has been read.
*/
public LineBasedFrameDecoder(final int maxLength, final boolean stripDelimiter, final boolean failFast) {
this.maxLength = maxLength;
this.failFast = failFast;
this.stripDelimiter = stripDelimiter;
}
@Override
protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
Object decoded = decode(ctx, in); // 调用重载的方法
if (decoded != null) { // 如果解析到对象，放入outList
out.add(decoded);
}
}
/**
* Create a frame out of the {@link ByteBuf} and return it.
*
* @param ctx the {@link ChannelHandlerContext} which this {@link ByteToMessageDecoder} belongs to
* @param buffer the {@link ByteBuf} from which to read data
* @return frame the {@link ByteBuf} which represent the frame or {@code null} if no frame could
* be created.
*/
protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
final int eol = findEndOfLine(buffer); // 找行的结尾
if (!discarding) {
if (eol >= 0) {
final ByteBuf frame;
final int length = eol - buffer.readerIndex();
final int delimLength = buffer.getByte(eol) == '\r'? 2 : 1;
if (length > maxLength) { // 数据流长度> 最大解析长度
buffer.readerIndex(eol + delimL ength); // 如果超过，把readerIndex直接指向换行符之后的字节，相当于把超长的消息给丢弃了
fail(ctx, length); // 传播异常
return null;
}
if (stripDelimiter) { // 不带分隔符
frame = buffer.readRetainedSlice(length); // 从累加器读取指定长度
buffer.skipBytes(delimLength);
} else { // 解析出来的消息带分隔符
frame = buffer.readRetainedSlice(length + delimLength);
}
return frame;
} else { // 非丢弃模式，找不到换行符的时候
final int length = buffer.readableBytes();
if (length > maxLength) { // 大于最大长度了
discardedBytes = length; // 丢弃
buffer.readerIndex(buffer.writerIndex());
discarding = true; // 当前进入丢弃模式
offset = 0;
if (failFast) { // 传播异常
fail(ctx, "over " + discardedBytes);
}
}
return null;
}
} else { // 丢弃模式
if (eol >= 0) {
final int length = discardedBytes + eol - buffer.readerIndex(); // 计算丢弃模式下后半段消息的长度（前半段因为超长已经丢弃了，这里处理的是后半段垃圾信息）
final int delimLength = buffer.getByte(eol) == '\r'? 2 : 1;
buffer.readerIndex(eol + delimLength); // 读指针移到新行开头，把length那段整体丢弃
discardedBytes = 0;
discarding = false; // 进入正常模式--非丢弃模式
if (!failFast ) {
fail(ctx, length);
}
} else {// 丢弃模式下未找到换行符，继续丢弃，然后计算长度
discardedBytes += buffer.readableBytes();
buffer.readerIndex(buffer.writerIndex());
// We skip everything in the buffer, we need to set the offset to 0 again.
offset = 0;
}
return null;
}
}
private void fail(final ChannelHandlerContext ctx, int length) {
fail(ctx, String.valueOf(length));
}
private void fail(final ChannelHandlerContext ctx, String length) {
ctx.fireExceptionCaught(
new TooLongFrameException(
"frame length (" + length + ") exceeds the allowed maximum (" + maxLength + ')'));
}
/**
* Returns the index in the buffer of the end of line found.
* Returns -1 if no end of line was found in the buffer.
*/
private int findEndOfLine(final ByteBuf buffer) {
int totalLength = buffer.readableBytes();
int i = buffer.forEachByte(buffer.readerIndex() + offset, totalLength - offset, ByteProcessor.FIND_LF);
if (i >= 0) {
offset = 0;
if (i > 0 && buffer.getByte(i - 1) == '\r') {
i--;
}
} else {
offset = totalLength;
}
return i;
}
}
5. 基于分隔符解码器 DelimiterBasedFrameDecoder 分析
它最少要有两个参数，一个是解析数据流最大长度，一个是分隔符，它可以传多个分隔符进去。
基于分隔符解码器分析解码步骤
- 行处理器
- 解码
public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder {
private final ByteBuf[] delimiters;
private final int maxFrameLength;
private final boolean stripDelimiter;
private final boolean failFast;
private boolean discardingTooLongFrame;
private int tooLongFrameLength;
/** Set only when decoding with "\n" and "\r\n" as the delimiter. */
private final LineBasedFrameDecoder lineBasedDecoder;
/**
* Creates a new instance.
*
* @param maxFrameLength the maximum length of the decoded frame.
* A {@link TooLongFrameException} is thrown if
* the length of the frame exceeds this value.
* @param delimiter the delimiter
*/
public DelimiterBasedFrameDecoder(int maxFrameLength, ByteBuf delimiter) {
this(maxFrameLength, true, delimiter);
}
/**
* Creates a new instance.
*
* @param maxFrameLength the maximum length of the decoded frame.
* A {@link TooLongFrameException} is thrown if
* the length of the frame exceeds this value.
* @param stripDelimiter whether the decoded frame should strip out the
* delimiter or not
* @param delimiter the delimiter
*/
public DelimiterBasedFrameDecoder(
int maxFrameLength, boolean stripDelimiter, ByteBuf delimiter) {
this(maxFrameLength, stripDelimiter, true, delimiter);
}
/**
* Creates a new instance.
*
* @param maxFrameLength the maximum length of the decoded frame.
* A {@link TooLongFrameException} is thrown if
* the length of the frame exceeds this value.
* @param stripDelimiter whether the decoded frame should strip out the
* delimiter or not
* @param failFast If <tt>true</tt>, a {@link TooLongFrameException} is
* thrown as soon as the decoder notices the length of the
* frame will exceed <tt>maxFrameLength</tt> regardless of
* whether the entire frame has been read.
* If <tt>false</tt>, a {@link TooLongFrameException} is
* thrown after the entire frame that exceeds
* <tt>maxFrameLength</tt> has been read.
* @param delimiter the delimiter
*/
public DelimiterBasedFrameDecoder(
int maxFrameLength, boolean stripDelimiter, boolean failFast,
ByteBuf delimiter) {
this(maxFrameLength, stripDelimiter, failFast, new ByteBuf[] {
delimiter.slice(delimiter.readerIndex(), delimiter.readableBytes())});
}
/**
* Creates a new instance.
*
* @param maxFrameLength the maximum length of the decoded frame.
* A {@link TooLongFrameException} is thrown if
* the length of the frame exceeds this value.
* @param delimiters the delimiters
*/
public DelimiterBasedFrameDecoder(int maxFrameLength, ByteBuf... delimiters) {
this(maxFrameLength, true, delimiters);
}
/**
* Creates a new instance.
*
* @param maxFrameLength the maximum length of the decoded frame.
* A {@link TooLongFrameException} is thrown if
* the length of the frame exceeds this value.
* @param stripDelimiter whether the decoded frame should strip out the
* delimiter or not
* @param delimiters the delimiters
*/
public DelimiterBasedFrameDecoder(
int maxFrameLength, boolean stripDelimiter, ByteBuf... delimiters) {
this(maxFrameLength, stripDelimiter, true, delimiters);
}
/**
* Creates a new instance.
*
* @param maxFrameLength the maximum length of the decoded frame.
* A {@link TooLongFrameException} is thrown if
* the length of the frame exceeds this value.
* @param stripDelimiter whether the decoded frame should strip out the
* delimiter or not
* @param failFast If <tt>true</tt>, a {@link TooLongFrameException} is
* thrown as soon as the decoder notices the length of the
* frame will exceed <tt>maxFrameLength</tt> regardless of
* whether the entire frame has been read.
* If <tt>false</tt>, a {@link TooLongFrameException} is
* thrown after the entire frame that exceeds
* <tt>maxFrameLength</tt> has been read.
* @param delimiters the delimiters
*/
public DelimiterBasedFrameDecoder(
int maxFrameLength, boolean stripDelimiter, boolean failFast, ByteBuf... delimiters) {
validateMaxFrameLength(maxFrameLength);
if (delimiters == null) {
throw new NullPointerException("delimiters");
}
if (delimiters.length == 0) {
throw new IllegalArgumentException("empty delimiters");
}
if (isLineBased(delimiters) && !isSubclass()) {
lineBasedDecoder = new LineBasedFrameDecoder(maxFrameLength, stripDelimiter, failFast);
this.delimiters = null;
} else {
this.delimiters = new ByteBuf[delimiters.length];
for (int i = 0; i < delimiters.length; i ++) {
ByteBuf d = delimiters[i];
validateDelimiter(d);
this.delimiters[i] = d.slice(d.readerIndex(), d.readableBytes());
}
lineBasedDecoder = null;
}
this.maxFrameLength = maxFrameLength;
this.stripDelimiter = stripDelimiter;
this.failFast = failFast;
}
/** Returns true if the delimiters are "\n" and "\r\n". */
private static boolean isLineBased(final ByteBuf[] delimiters) {
if (delimiters.length != 2) {
return false;
}
ByteBuf a = delimiters[0];
ByteBuf b = delimiters[1];
if (a.capacity() < b.capacity()) {
a = delimiters[1];
b = delimiters[0];
}
return a.capacity() == 2 && b.capacity() == 1
&& a.getByte(0) == '\r' && a.getByte(1) == '\n'
&& b.getByte(0) == '\n';
}
/**
* Return {@code true} if the current instance is a subclass of DelimiterBasedFrameDecoder
*/
private boolean isSubclass() {
return getClass() != DelimiterBasedFrameDecoder.class;
}
@Override
protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
Object decoded = decode(ctx, in); // 重载一个decode方法
if (decoded != null) {
out.add(decoded); // 解析出的对象加入outList
}
}
/**
* Create a frame out of the {@link ByteBuf} and return it.
*
* @param ctx the {@link ChannelHandlerContext} which this {@link ByteToMessageDecoder} belongs to
* @param buffer the {@link ByteBuf} from which to read data
* @return frame the {@link ByteBuf} which represent the frame or {@code null} if no frame could
* be created.
*/
protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
if (lineBasedDecoder != null) { // 如果行处理器不为空，调用行处理器
return lineBasedDecoder.decode(ctx, buffer);
}
// Try all delimiters and choose the delimiter which yields the shortest frame.
int minFrameLength = Integer.MAX_VALUE; // 找到最小分隔符
ByteBuf minDelim = null;
for (ByteBuf delim: delimiters) {
int frameLength = indexOf(buffer, delim);
if (frameLength >= 0 && frameLength < minFrameLength) { // 以某个分隔符分割的最小数据包的长度
minFrameLength = frameLength;
minDelim = delim;
}
}
if (minDelim != null) { // 开始解码，找到分隔符
int minDelimLength = minDelim.capacity();
ByteBuf frame;
if (discardingTooLongFrame) { // 丢弃模式
// We've just finished discarding a very large frame.
// Go back to the initial state.
discardingTooLongFrame = false; // 设置正常模式，非丢弃模式
buffer.skipBytes(minFrameLength + minDelimLength);
int tooLongFrameLength = this.tooLongFrameLength;
this.tooLongFrameLength = 0;
if (!failFast) {
fail(tooLongFrameLength);
}
return null;
}
if (minFrameLength > maxFrameLength) {
// Discard read frame. 丢弃
buffer.skipBytes(minFrameLength + minDelimLength);
fail(minFrameLength); // 传播异常
return null;
}
if (stripDelimiter) { // 是否包含分隔符
frame = buffer.readRetainedSlice(minFrameLength);
buffer.skipBytes(minDelimLength);
} else {
frame = buffer.readRetainedSlice(minFrameLength + minDelimLength);
}
return frame;
} else { // 没有找到分隔符
if (!discardingTooLongFrame) { // 非丢弃模式
if (buffer.readableBytes() > maxFrameLength) { // 数据包过长，丢弃
// Discard the content of the buffer until a delimiter is found.
tooLongFrameLength = buffer.readableBytes(); // 记录丢弃的字节数
buffer.skipBytes(buffer.readableBytes());
discardingTooLongFrame = true; // 标记为丢弃状态
if (failFast) {
fail(tooLongFrameLength);
}
}
} else { // 当前处于丢弃模式
// Still discarding the buffer since a delimiter is not found.
tooLongFrameLength += buffer.readableBytes();
buffer.skipBytes(buffer.readableBytes());
}
return null;
}
}
private void fail(long frameLength) {
if (frameLength > 0) {
throw new TooLongFrameException(
"frame length exceeds " + maxFrameLength +
": " + frameLength + " - discarded");
} else {
throw new TooLongFrameException(
"frame length exceeds " + maxFrameLength +
" - discarding");
}
}
/**
* Returns the number of bytes between the readerIndex of the haystack and
* the first needle found in the haystack. -1 is returned if no needle is
* found in the haystack.
*/
private static int indexOf(ByteBuf haystack, ByteBuf needle) {
for (int i = haystack.readerIndex(); i < haystack.writerIndex(); i ++) {
int haystackIndex = i;
int needleIndex;
for (needleIndex = 0; needleIndex < needle.capacity(); needleIndex ++) {
if (haystack.getByte(haystackIndex) != needle.getByte(needleIndex)) {
break;
} else {
haystackIndex ++;
if (haystackIndex == haystack.writerIndex() &&
needleIndex != needle.capacity() - 1) {
return -1;
}
}
}
if (needleIndex == needle.capacity()) {
// Found the needle from the haystack!
return i - haystack.readerIndex();
}
}
return -1;
}
private static void validateDelimiter(ByteBuf delimiter) {
if (delimiter == null) {
throw new NullPointerException("delimiter");
}
if (!delimiter.isReadable()) {
throw new IllegalArgumentException("empty delimiter");
}
}
private static void validateMaxFrameLength(int maxFrameLength) {
checkPositive(maxFrameLength, "maxFrameLength");
}
}
6. 基于长度域的解码器 LengthFieldBasedFrameDecoder 分析
重要参数
lengthFieldOffset: 长度域在二进制数据流中偏移量。
lengthFieldLength：长度域的长度，有几个取值1，2，3，4，8。
lengthAdjustment：lengthFieldLength不代表一个消息的完整长度，lengthFieldLength指定长度+lengthAdjustment 才是。用于lengthFieldLength指定的长度包含长度域的情况，这时候lengthAdjustment设置为负值方便解码。
initialBytesToStrip：我在解码的时候，需要跳过的字节数，常用于跳过长度域。
解码过程
LengthFieldBasedFrameDecoder 解码过程如下：
- 计算需要抽取的数据包长度
- 跳过字节逻辑处理
- 丢弃模式下的处理
我们来看它的核心解码代码：
protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
if (discardingTooLongFrame) {
discardingTooLongFrame(in);
}
if (in.readableBytes() < lengthFieldEndOffset) { // 当前字节数还不够 解析出lengthFieldLength
return null;
}
int actualLengthFieldOffset = in.readerIndex() + lengthFieldOffset; // 因为是流，要加上读当前指针的位置，计算实际偏移量
long frameLength = getUnadjustedFrameLength(in, actualLengthFieldOffset, lengthFieldLength, byteOrder); // lengthFieldLength中指定的数据包长度
if (frameLength < 0) {
failOnNegativeLengthField(in, frameLength, lengthFieldEndOffset);
}
frameLength += lengthAdjustment + lengthFieldEndOffset; // 计算出需要抽取的数据包长度
if (frameLength < lengthFieldEndOffset) {
failOnFrameLengthLessThanLengthFieldEndOffset(in, frameLength, lengthFieldEndOffset);
}
if (frameLength > maxFrameLength) { // 这里面会设置进入丢弃模式
exceededFrameLength(in, frameLength);
return null;
}
// never overflows because it's less than maxFrameLength
int frameLengthInt = (int) frameLength;
if (in.readableBytes() < frameLengthInt) {
return null;
}
if (initialBytesToStrip > frameLengthInt) {
failOnFrameLengthLessThanInitialBytesToStrip(in, frameLength, initialBytesToStrip);
}
// 到这里已经有一个完整的数据包了
in.skipBytes(initialBytesToStrip); // 跳过略过的长度
// extract frame
int readerIndex = in.readerIndex();
int actualFrameLength = frameLengthInt - initialBytesToStrip; // 真正需要拿的字节
ByteBuf frame = extractFrame(ctx, in, readerIndex, actualFrameLength);
in.readerIndex(readerIndex + actualFrameLength);
return frame;
}
frameLength > maxFrameLength就会设置丢弃模式
丢弃模式处理
private void exceededFrameLength(ByteBuf in, long frameLength) {
long discard = frameLength - in.readableBytes(); // 计算下次还要丢弃的字节数
tooLongFrameLength = frameLength; // 记录
if (discard < 0) { // 说明一次丢弃就能解决问题
// buffer contains more bytes then the frameLength so we can discard all now
in.skipBytes((int) frameLength);
} else { // 说明还需要一次丢弃才能解决问题
// Enter the discard mode and discard everything received so far.
discardingTooLongFrame = true; // 设置为丢弃模式
bytesToDiscard = discard;
in.skipBytes(in.readableBytes()); // 跳过，即丢弃
}
failIfNecessary(true);
}
我们来看 failNecessary方法：
private void failIfNecessary(boolean firstDetectionOfTooLongFrame) {
if (bytesToDiscard == 0) { // 说明不需要丢弃
// Reset to the initial state and tell the handlers that
// the frame was too large.
long tooLongFrameLength = this.tooLongFrameLength;
this.tooLongFrameLength = 0;
discardingTooLongFrame = false; // 设置为正常模式
if (!failFast || firstDetectionOfTooLongFrame) { // 是否需要快速失败
fail(tooLongFrameLength);
}
} else {
// Keep discarding and notify handlers if necessary.
if (failFast && firstDetectionOfTooLongFrame) { // 如果是快速失败，且是第一次
fail(tooLongFrameLength); // 发出失败通知
}
}
}
7. 解码器总结
1. 解码器抽象的解码过程是怎样的？
答：抽象的解码过程是通过ByteToMessageDecoder实现的，它的解码过程分为以下三步骤：第一步累加字节流，它会把当前读取的字节流累加到累加器cumulator中；第二步调用子类的decode方法进行解析，decode方法实际上是一个抽象方法，不同的子类解码器的decode有不同的实现。调用decode时会传入两个比较重要的参数，一个是当前累加的字节流，一个是outList，子类在解码时从累加器中读取一段数据，如果成功解析出一个数据包，就把这个数据包添加到outList中；第三步，如果outList中有解析出的数据包，就通过pipeline的事件传播机制往下传播。
2. Netty中有哪些拆箱即用的解码器？
netty提供了许许多多开箱即用的解码器，99%的业务场景下使用netty提供的解码器就能完成需求，使用netty提供的解码器代码的健壮性能够得到保证。比如固定长度解码器FixedLengthFrameDecoder，行解码器（LineBasedFrameDecoder）；固定分隔符解码器（DelimiterBasedFrameDecoder），可以传一些分隔符进去，如果是行分隔符netty会做特殊处理；基于长度域的解码器（LengthFieldBasedFrameDecoder），比较通用的解码器，解码的过程比较复杂，主要有三个步骤，第一步计算需要抽取的数据包的长度，第二步对拿到的数据包进行处理比如跳过指定字节数，处理完之后一个完整的数据包就可以放入outList中，接下来就是丢弃模式的处理。
