Netty 源码解析 ——— AdaptiveRecvByteBufAllocator

本文是Netty文集中“Netty 源码解析”系列的文章。主要对Netty的重要流程以及类进行源码解析，以使得我们更好的去使用Netty。Netty是一个非常优秀的网络框架，对其源码解读的过程也是不断学习的过程。

AdaptiveRecvByteBufAllocator主要用于构建一个最优大小的缓冲区来接收数据。比如，在读事件中就会通过该类来获取一个最优大小的的缓冲区来接收对端发送过来的可读取的数据。

关于AdaptiveRecvByteBufAllocator的分析，会通过一层层的Java doc来展开。如果一开始看这些方法说明有些个不大明白，没关系，在文章的最后会结合Netty中真实的使用场景来对AdaptiveRecvByteBufAllocator的使用说明分析，这样就能更好的理解AdaptiveRecvByteBufAllocator的用途。

本文主要针对NIO网络传输模式对AdaptiveRecvByteBufAllocator类展开的分析。

RecvByteBufAllocator

分配一个新的接收缓存，该缓存的容量会尽可能的足够大以读入所有的入站数据并且该缓存的容量也尽可能的小以不会浪费它的空间。

Handle newHandle() 创建一个新的处理器，该处理器提供一个真实的操作并持有内部信息，该信息是用于预测一个最优缓冲区大小的必要信息。

内部接口

interface Handle

• ByteBuf allocate(ByteBufAllocator alloc); 创建一个新的接收缓冲区，该缓冲区的大小可能足够大去读取所有的数据并且足够小以至于不会浪费它的空间。
• int guess() 类似于一个「allocate(ByteBufAllocator)」操作，除了它不会去分配任何东西只是告诉你分配容量的大小。
• void reset(ChannelConfig config) 重置所有的计数器，该计数器已经累计并会推荐下次读循环应该读取的消息次数以及读取字节的数量。 它可能被「continueReading()」方式使用去决定是否读操作应该完成。 这仅仅只是一个暗示并且可能被忽略在实现的时候。 参数 config：config是Channel的配置，它可能会影响对象的行为。
• void incMessagesRead(int numMessages) 增加当前读循环中已经读取消息的次数(注意，这里不是读取数据的字节数，读取消息指的是一个读操作)。
• void lastBytesRead(int bytes) 设置最后一次读操作已经读取到的字节数。 这可能被用于增加已经读取的字节数。 参数 bytes：由读操作提供的字节数。这可能是一个负数如果一个读错误发生。如果看到负值，那么它会预计在下一次调用「lastBytesRead()」时返回。对于该类来说一个负值将标识一个外部强制执行终止的情况，并且它不要求在「continueReading()」时执行。
• int lastBytesRead() 获取最近一次读操作的字节数。
• void attemptedBytesRead(int bytes) 设置有多少字节读操作将尝试读取，或已经读取。
• int attemptedBytesRead() 获取有多少字节读操作将尝试读取，或已经读取。
• boolean continueReading() 检测是否当前的读循环可以继续。 返回true：如果读循环应该继续读取数据；false：如果读循环已经完成。
• void readComplete() 读操作已经完成。

interface ExtendedHandle extends Handle

• boolean continueReading(UncheckedBooleanSupplier maybeMoreDataSupplier) 和「Handle#continueReading()」方法一样，除了通过supplier参数让更多的数据来决定结果。

内部类

class DelegatingHandle implements Handle 该类只有一个真实Handler的引用，所有方法的请求最终都将由这个真实Handler来完成。 该类实现了代理模式中的静态代理。

MaxMessagesRecvByteBufAllocator

限制事件循环尝试读操作(比如，READ事件触发)时尝试读操作的次数。比如，处理READ事件时，限制读循环操作可执行的读操作的次数。

• int maxMessagesPerRead(); 返回每个读循环读取消息的最大次数，即，在NIO传输模式下一个读循环最大能执行几次循环操作。 如果该值大于1，那么一次事件循环可能尝试去读多次以获取多个消息。
• MaxMessagesRecvByteBufAllocator maxMessagesPerRead(int maxMessagesPerRead) 设置一个读循环可读取的最大消息个数。 如果该值大于1，那么一次事件循环可能尝试去读多次以获取多个消息。

DefaultMaxMessagesRecvByteBufAllocator

MaxMessagesRecvByteBufAllocator的默认实现，它遵守「ChannelConfig#isAutoRead()」并防止溢出。

    private volatile int maxMessagesPerRead;

    public DefaultMaxMessagesRecvByteBufAllocator() {
        this(1);
    }

    public DefaultMaxMessagesRecvByteBufAllocator(int maxMessagesPerRead) {
        maxMessagesPerRead(maxMessagesPerRead);
    }

    @Override
    public int maxMessagesPerRead() {
        return maxMessagesPerRead;
    }

    @Override
    public MaxMessagesRecvByteBufAllocator maxMessagesPerRead(int maxMessagesPerRead) {
        if (maxMessagesPerRead <= 0) {
            throw new IllegalArgumentException("maxMessagesPerRead: " + maxMessagesPerRead + " (expected: > 0)");
        }
        this.maxMessagesPerRead = maxMessagesPerRead;
        return this;
    }

默认构造方法提供的每个读取循环最大可读取消息个数为1。

内部类

public abstract class MaxMessageHandle implements ExtendedHandle 专注于施行每次读操作最多可读取消息个数的条件，用于「continueReading()」中。

// Channel的配置对象
private ChannelConfig config;

// 每个读循环可循环读取消息的最大次数。
private int maxMessagePerRead;

// 目前读循环已经读取的消息个数。即，在NIO传输模式下也就是读循环已经执行的循环次数
private int totalMessages;

// 目前已经读取到的消息字节总数
private int totalBytesRead;

// 本次将要进行的读操作，期望读取的字节数。也就是有这么多个字节等待被读取。
private int attemptedBytesRead;

// 最后一次读操作读取到的字节数。
private int lastBytesRead;

        private final UncheckedBooleanSupplier defaultMaybeMoreSupplier = new UncheckedBooleanSupplier() {
            @Override
            public boolean get() {
                return attemptedBytesRead == lastBytesRead;
            }
        };

默认判断是否可读取更多消息的提供器，会在continueReading操作中使用。 表示，如果‘最近一次读操作所期望读取的字节数’与‘最近一次读操作真实读取的字节数’一样，则表示当前的缓冲区容量已经被写满了，可能还有数据等待着被读取。

• reset(ChannelConfig config)

        public void reset(ChannelConfig config) {
            this.config = config;
            maxMessagePerRead = maxMessagesPerRead();
            totalMessages = totalBytesRead = 0;
        }

重新设置config成员变量，并将totalMessages、totalBytesRead重置为0。重置maxMessagePerRead。

• ByteBuf allocate(ByteBufAllocator alloc)

        public ByteBuf allocate(ByteBufAllocator alloc) {
            return alloc.ioBuffer(guess());
        }

根据给定的缓冲区分配器，以及guess()所返回的预测的缓存区容量大小，构建一个新的缓冲区。

• void lastBytesRead(int bytes)

        public final void lastBytesRead(int bytes) {
            lastBytesRead = bytes;
            if (bytes > 0) {
                totalBytesRead += bytes;
            }
        }

设置最近一次读操作的读取字节数。这里只有当bytes>0时才会进行totalBytesRead的累加。因为当bytes<0时，不是真实的读取字节的数量了，而标识一个外部强制执行终止的情况。

• continueReading

        public boolean continueReading() {
            return continueReading(defaultMaybeMoreSupplier);
        }

        @Override
        public boolean continueReading(UncheckedBooleanSupplier maybeMoreDataSupplier) {
            return config.isAutoRead() &&
                   maybeMoreDataSupplier.get() &&
                   totalMessages < maxMessagePerRead &&
                   totalBytesRead > 0;
        }

continueReading返回true需要同时满足如下条件： ① ChannelConfig的设置为可自动读取。即，autoRead属性为1。 ② maybeMoreDataSupplier.get()返回为true，这个我们在上面已经讨论过了。也就是当‘最近一次读操作所期望读取的字节数’与‘最近一次读操作真实读取的字节数’一样，则表示当前可能还有数据等待被读取。则就会返回true。 ③ totalMessages < maxMessagePerRead ： 已经读取的消息次数 < 一个读循环最大能读取消息的次数 ④ totalBytesRead > 0 ：因为totalBytesRead是int类型，所以totalBytesRead的最大值是’Integer.MAX_VALUE’(即，2147483647)。所以，也限制了一个读循环最大能读取的字节数为2147483647。

• int totalBytesRead()

        protected final int totalBytesRead() {
            return totalBytesRead < 0 ? Integer.MAX_VALUE : totalBytesRead;
        }

返回已经读取的字节个数，若‘totalBytesRead < 0’则说明已经读取的字节数已经操作了’Integer.MAX_VALUE’，则返回Integer.MAX_VALUE；否则返回真实的已经读取的字节数。

同时，我们可以留意MaxMessageHandle抽象类，将‘incMessagesRead(int amt)’、‘lastBytesRead(int bytes)’、‘int lastBytesRead()’、‘int totalBytesRead()’这几个方法都定义为了final修饰符，这使得子类不能够对这几个方法进行重写。

AdaptiveRecvByteBufAllocator

RecvByteBufAllocator会根据反馈自动的增加和减少可预测的buffer的大小。 它会逐渐地增加期望的可读到的字节数如果之前的读循环操作所读取到的字节数据已经完全填充满了分配好的buffer( 也就是，上一次的读循环操作中执行的所有读取操作所累加的读到的字节数，已经大于等于预测分配的buffer的容量大小，那么它就会很优雅的自动的去增加可读的字节数量，也就是自动的增加缓冲区的大小 )。它也会逐渐的减少期望的可读的字节数如果’连续’两次读循环操作后都没有填充满分配的buffer。否则，它会保持相同的预测。

// 在调整缓冲区大小时，若是增加缓冲区容量，那么增加的索引值。
// 比如，当前缓冲区的大小为SIZE_TABLE[20],若预测下次需要创建的缓冲区需要增加容量大小，
// 则新缓冲区的大小为SIZE_TABLE[20 + INDEX_INCREMENT]，即SIZE_TABLE[24]
private static final int INDEX_INCREMENT = 4;    

// 在调整缓冲区大小时，若是减少缓冲区容量，那么减少的索引值。
// 比如，当前缓冲区的大小为SIZE_TABLE[20],若预测下次需要创建的缓冲区需要减小容量大小，
// 则新缓冲区的大小为SIZE_TABLE[20 - INDEX_DECREMENT]，即SIZE_TABLE[19]
private static final int INDEX_DECREMENT = 1;    

// 用于存储缓冲区容量大小的数组
private static final int[] SIZE_TABLE;    

    static {
        List<Integer> sizeTable = new ArrayList<Integer>();
        for (int i = 16; i < 512; i += 16) {
            sizeTable.add(i);
        }

        for (int i = 512; i > 0; i <<= 1) {
            sizeTable.add(i);
        }

        SIZE_TABLE = new int[sizeTable.size()];
        for (int i = 0; i < SIZE_TABLE.length; i ++) {
            SIZE_TABLE[i] = sizeTable.get(i);
        }
    }

① 依次往sizeTable添加元素：[16 , (512-16)]之间16的倍数。即，16、32、48...496 ② 然后再往sizeTable中添加元素：[512 , 512 * (2^N))，N > 1; 直到数值超过Integer的限制(2^31 - 1)； ③ 根据sizeTable长度构建一个静态成员常量数组SIZE_TABLE，并将sizeTable中的元素赋值给SIZE_TABLE数组。注意List是有序的，所以是根据插入元素的顺序依次的赋值给SIZE_TABLE，SIZE_TABLE从下标0开始。

SIZE_TABLE为预定义好的以从小到大的顺序设定的可分配缓冲区的大小值的数组。因为AdaptiveRecvByteBufAllocator作用是可自动适配每次读事件使用的buffer的大小。这样当需要对buffer大小做调整时，只要根据一定逻辑从SIZE_TABLE中取出值，然后根据该值创建新buffer即可。

• 构造方法

static final int DEFAULT_MINIMUM = 64;    // 默认缓冲区的最小容量大小为64
static final int DEFAULT_INITIAL = 1024;    // 默认缓冲区的容量大小为1024
static final int DEFAULT_MAXIMUM = 65536;    // 默认缓冲区的最大容量大小为65536

// 使用默认参数创建一个新的AdaptiveRecvByteBufAllocator。
// 默认参数，预计缓冲区大小从1024开始，最小不会小于64，最大不会大于65536。
public AdaptiveRecvByteBufAllocator() {
    this(DEFAULT_MINIMUM, DEFAULT_INITIAL, DEFAULT_MAXIMUM);
}

    private final int minIndex;    // 缓冲区最小容量对应于SIZE_TABLE中的下标位置
    private final int maxIndex;    // 缓冲区最大容量对应于SIZE_TABLE中的下标位置
    private final int initial;     // 缓冲区默认容量大小 

    // 使用指定的参数创建AdaptiveRecvByteBufAllocator对象。
    // 其中minimum、initial、maximum是正整数。然后通过getSizeTableIndex()方法获取相应容量在SIZE_TABLE中的索引位置。
    // 并将计算出来的索引赋值给相应的成员变量minIndex、maxIndex。同时保证「SIZE_TABLE[minIndex] >= minimum」以及「SIZE_TABLE[maxIndex] <= maximum」.
    public AdaptiveRecvByteBufAllocator(int minimum, int initial, int maximum) {
        if (minimum <= 0) {
            throw new IllegalArgumentException("minimum: " + minimum);
        }
        if (initial < minimum) {
            throw new IllegalArgumentException("initial: " + initial);
        }
        if (maximum < initial) {
            throw new IllegalArgumentException("maximum: " + maximum);
        }

        int minIndex = getSizeTableIndex(minimum);
        if (SIZE_TABLE[minIndex] < minimum) {
            this.minIndex = minIndex + 1;
        } else {
            this.minIndex = minIndex;
        }

        int maxIndex = getSizeTableIndex(maximum);
        if (SIZE_TABLE[maxIndex] > maximum) {
            this.maxIndex = maxIndex - 1;
        } else {
            this.maxIndex = maxIndex;
        }

        this.initial = initial;
    }

• int getSizeTableIndex(final int size)

    private static int getSizeTableIndex(final int size) {
        for (int low = 0, high = SIZE_TABLE.length - 1;;) {
            if (high < low) {
                return low;
            }
            if (high == low) {
                return high;
            }

            int mid = low + high >>> 1;
            int a = SIZE_TABLE[mid];
            int b = SIZE_TABLE[mid + 1];
            if (size > b) {
                low = mid + 1;
            } else if (size < a) {
                high = mid - 1;
            } else if (size == a) {
                return mid;
            } else {
                return mid + 1;
            }
        }
    }

因为SIZE_TABLE数组是一个有序数组，因此此处用二分查找法，查找size在SIZE_TABLE中的位置，如果size存在于SIZE_TABLE中，则返回对应的索引值；否则返回接近于size大小的SIZE_TABLE数组元素的索引值。

• Handle newHandle()

    public Handle newHandle() {
        return new HandleImpl(minIndex, maxIndex, initial);
    }

创建一个HandleImpl对象，参数为minIndex，maxIndex以及initail为AdaptiveRecvByteBufAllocator对象的成员变量值。

内部类

private final class HandleImpl extends MaxMessageHandle HandleImpl是AdaptiveRecvByteBufAllocator一个内部类，该处理器类用于提供真实的操作并保留预测最佳缓冲区容量所需的内部信息。

// 缓冲区最小容量对应于SIZE_TABLE中的下标位置，同外部类AdaptiveRecvByteBufAllocator是一个值
private final int minIndex;    

// 缓冲区最大容量对应于SIZE_TABLE中的下标位置，同外部类AdaptiveRecvByteBufAllocator是一个值
private final int maxIndex;    

// 缓冲区默认容量对应于SIZE_TABLE中的下标位置，外部类AdaptiveRecvByteBufAllocator记录的是容量大小值，而HandleImpl中记录是其值对应于SIZE_TABLE中的下标位置
private int index;            

// 下一次创建缓冲区时的其容量的大小。
private int nextReceiveBufferSize;    

// 在record()方法中使用，用于标识是否需要减少下一次创建的缓冲区的大小。
private boolean decreaseNow;        

• 构造方法

    public HandleImpl(int minIndex, int maxIndex, int initial) {
        this.minIndex = minIndex;
        this.maxIndex = maxIndex;

        index = getSizeTableIndex(initial);
        nextReceiveBufferSize = SIZE_TABLE[index];
    }

给成员变量minIndex、maxIndex赋值。同时通过getSizeTableIndex(initial)计算出初始容量在SIZE_TABLE的索引值，将其赋值为成员变量index。并将初始容量大小(即，SIZE_TABLE[index])赋值给成员变量nextReceiveBufferSize。

• int guess()

public int guess() {
    return nextReceiveBufferSize;
}

返回推测的缓冲区容量大小，即，返回成员变量nextReceiveBufferSize的值。

• record(int actualReadBytes)

        private void record(int actualReadBytes) {
            if (actualReadBytes <= SIZE_TABLE[Math.max(0, index - INDEX_DECREMENT - 1)]) {
                if (decreaseNow) {
                    index = Math.max(index - INDEX_DECREMENT, minIndex);
                    nextReceiveBufferSize = SIZE_TABLE[index];
                    decreaseNow = false;
                } else {
                    decreaseNow = true;
                }
            } else if (actualReadBytes >= nextReceiveBufferSize) {
                index = Math.min(index + INDEX_INCREMENT, maxIndex);
                nextReceiveBufferSize = SIZE_TABLE[index];
                decreaseNow = false;
            }
        }

record方法很重要，它就是完成预测下一个缓冲区容量大小的操作。逻辑如下： 若发现两次，本次读循环真实读取的字节总数 <= ‘SIZE_TABLE[Math.max(0, index - INDEX_DECREMENT - 1)]’ 则，减少预测的缓冲区容量大小。重新给成员变量index赋值为为‘index - 1’，若‘index - 1’ < minIndex，则index新值为minIndex。根据算出来的新的index索引，给成员变量nextReceiveBufferSize重新赋值'SIZE_TABLE[index]’。最后将decreaseNow置位false，该字段用于表示是否有’连续’的两次真实读取的数据满足可减少容量大小的情况。注意，这里说的‘连续’并不是真的连续发送，而是指满足条件(即，‘actualReadBytes <= SIZE_TABLE[Math.max(0, index - INDEX_DECREMENT - 1)]’)两次的期间没有发生‘actualReadBytes >= nextReceiveBufferSize’的情况。 若，本次读循环真实读取的字节总数 >= 预测的缓冲区大小，则进行增加预测的缓冲区容量大小。新的index为‘index + 4’，若‘index + 4’ > maxIndex，则index新值为maxIndex。根据算出来的新的index索引，给成员变量nextReceiveBufferSize重新赋值'SIZE_TABLE[index]’。最后将decreaseNow置位false。

• readComplete()

        public void readComplete() {
            record(totalBytesRead());
        }

每次读循环完成后，会调用该方法。根据本次读循环读取的字节数来调整预测的缓冲区容量大小。

结合Netty源码中的实际使用场景对AdaptiveRecvByteBufAllocator做进一步的分析

下面结合NioServerSocketChannel的ACCEPT事件以及NioSocketChannel的READ事件处理中对AdaptiveRecvByteBufAllocator使用，来进一步的了解AdaptiveRecvByteBufAllocator的真正用途。本文仅对AdaptiveRecvByteBufAllocator的使用进行解释，具体的事件的处理流程并不是本文的重点。

ACCEPT事件中对AdaptiveRecvByteBufAllocator的使用

当服务端收到客户端的一个连接请求时，‘SelectionKey.OP_ACCEPT’将会触发。在NioEventLoop的事件循环中会对该事件进行处理：

if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
    unsafe.read();
}

我们来看看unsafe.read()的实现，在NioServerSocketChannel中unsafe是一个NioMessageUnsafe实例：

① 获取一个AdaptiveRecvByteBufAllocator.HandleImpl实例

如果recvHandler不存在则创建一个AdaptiveRecvByteBufAllocator.HandleImpl实例，然后返回；否则直接返回。

② 对config成员变量赋值，即，Handle中持有NioServerSocketChannelConfig对象的引用；将maxMessagePerRead重置(这里为16，是在初始化NioServerSocketChannel的时候进行的相关设置)，totalMessages、totalBytesRead重置为0；

在启动流程中初始化NioServerSocketChannel的时候，会同时构建一个Channel的配置对象DefaultChannelConfig，在初始化该对象的过程中就会完成对AdaptiveRecvByteBufAllocator的创建，并修改其maxMessagesPerRead属性：

而channel.metadata()返回的是NioServerSocketChannel中的静态常量「ChannelMetadata METADATA」，它表示一个hasDisconnect为false，defaultMaxMessagesPerRead为16的Channel属性实现。 因此，NioServerSocketChannel关联的AdaptiveRecvByteBufAllocator的maxMessagePerRead属性值为16。 ③ 这步主要是通过serverSocket.accpet()来接受生成和客户端通信的socket，并将其放入到readBuf集合中。如果该流程操作正确，则返回’1’，表示已经读取一条消息(即，在处理ACCEPT事件中消息的读取指的就是接收一个客户端的请求「serverSocket.accpet()」操作)；否则返回0，若返回0，则会退出while读循环。

④ 在执行完消息的读取后(即，在处理ACCEPT事件中消息的读取指的就是接收一个客户端的请求「serverSocket.accpet()」操作)，将执行allocHandle.incMessagesRead(localRead)来增加已经读取消息的个数。底层就是根据localRead的值对totalMessages属性进行累加。

⑤ 在进行下一次循环进行消息的读取前，会先执行该判断，判断是否可以继续的去读取消息。

a) config.isAutoRead()：

NioServerSocketChannelConfig的autoRead默认为1，因此该判断为true b) maybeMoreDataSupplier.get() ：

因为accept中并没有真实读取字节的操作，因此此时，attemptedBytesRead、lastBytesRead都为0。该判断为true； c) totalMessages < maxMessagePerRead：根据上面的流程我们可以知道，maxMessagePerRead为16，totalMessages也为1。因为此判断为true。 d) totalBytesRead > 0：因为accept操作只是接收一个新的连接，并没有进行真实的数据读取操作，因此totalBytesRead为0。因此此判断为false。 也就是allocHandle.continueReading()将返回false。因此退出循环，会继续读取消息。

⑥ 根据本次读取的字节数，对下次创建的缓冲区大小做调整。其实本方法，在ACCEPT事件中并没用处。因为，ServerSocektChannel是没有READ事件的，它只会处理ACCEPT事件，所以它不会读取任何的字节数据。再者在前面处理ACCEPT事件的流程中，我们也可以看到，我们并没有使用allocHandle.allocate(allocator);来真实的创建一个缓冲区。

总结，对于ACCEPT事件，每次读循环执行一次读操作(但并没有读取任何字节数据，totalBytesRead > 0 为false)这也是符合NIO规范的，因为每次ACCEPT事件被触发时，都表示有一个客户端向服务器端发起了连接请求。

READ事件中对AdaptiveRecvByteBufAllocator的使用

当有可读数据准备被读取时，‘SelectionKey.OP_READ’将会触发。在NioEventLoop的事件循环中会对该事件进行处理：

if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
    unsafe.read();
}

我们来看看unsafe.read()的实现，在NioSocketChannel中unsafe是一个NioByteUnsafe实例：

① 同ACCEPT事件流程类似，会获取到一个获取一个AdaptiveRecvByteBufAllocator.HandleImpl实例

② 同ACCEPT事件流程类似，对config成员变量赋值，即，Handle中持有NioSocketChannelConfig对象的引用；将maxMessagePerRead重置(这里为16，是在初始化NioSocketChannel的时候进行的相关设置)，totalMessages、totalBytesRead重置为0。只是这里是在实例化NioSocketChannel实例时完成的。

③ 根据提供的缓冲区分配器，创建一个由AdaptiveRecvByteBufAllocator.HandleImpl推测的容量大小的ByteBuf，并使用该ByteBuf来接收接下来要读取的数据。第一次读循环，默认的大小为1024。

④ allocHandle.lastBytesRead(doReadBytes(byteBuf)) ：

首先，会先根据byteBuf可写的字节数来设置AdaptiveRecvByteBufAllocator.HandleImpl的attemptedBytesRead属性。正如前面对AdaptiveRecvByteBufAllocator类的介绍所说的：“它会分配一个新的接收缓存，该缓存的容量会尽可能的足够大以读入入站数据并且该缓存的容量也尽可能的小以不会浪费它的空间。”因此，它认为它分配的ByteBuf中可写的字节数，就应该是本次尝试读取到的字节数。然后。使用上面构造的ByteBuf来接收SocketChannel的数据，并且限制了最大读取的字节数为attemptedBytesRead，然后将真是读取的字节数设置到AdaptiveRecvByteBufAllocator.HandleImpl中的lastBytesRead属性上。

⑤ 在执行完消息的读取后，将执行allocHandle.incMessagesRead(1)来增加已经读取消息的次数。底层就是将totalMessages值+1。

⑥ 在进行下一次循环进行消息的读取前，会先执行该判断，判断是否可以继续的去读取消息。

a) config.isAutoRead()：

NioSocketChannelConfig的autoRead默认为1，因此该判断为true b) maybeMoreDataSupplier.get() ：

当本次读操作读取到的字节数与AdaptiveRecvByteBufAllocator推测出的ByteBuf容量大小不一样时，就会返回false；否则返回true。当然，如上面所说，如果本次读操作可读取的字节大于了attemptedBytesRead的话，一次读操作也只会先读取attemptedBytesRead的字节数。在满足allocHandle.continueReading()的条件下，可以在读循环中进行下一次的数据读取。每次读循环都会构建一个新的ByteBuf。但是，请注意，一个读循环(可以包含多次读操作)中每次的读操作构建的ByteBuf大小都是一样的。 c) totalMessages < maxMessagePerRead：根据上面的流程我们可以知道，maxMessagePerRead为16，totalMessages为读循环已经执行的读操作次数(即，循环次数)。 d) totalBytesRead > 0：当本次读操作有读取到字节数时，或者以读取到的字节数小于Integer.MAX_VALUE，那么该判断都会大于0，即，为true；否则为false。

⑦ 最后，通过allocHandle.readComplete()来标识本次读循环结束，并根据本次读循环的数据信息来预测下一次读事件触发时，应该分配多大的ByteBuf容量更加合理些。具体的调整逻辑，在上面的HandleImpl.record(int actualReadBytes)已经进行了详细的说明。

总结，对于READ事件，一个读循环可能会执行多次读操作(即，循环的次数)，至于进行几次读操作，这将根据「allocHandle.continueReading()」以及当前这次读取的字节数来决定。若‘allocHandle.continueReading()’为false，或者本次读取到的字节数<=0(当没有数据可读取时为0，当远端已经关闭时为-1)，都不会继续进行读循环操作。再者，一个读循环中的每次读操作分配的ByteBuf的容量大小都是一样的。我们是在一个读循环操作完成后，才会根据本次的读循环的情况对下一次读操作的ByteBuf容量大小做预测。

后记

若文章有任何错误，望大家不吝指教:)

参考

圣思园《精通并发与Netty》