netty自适应缓冲区实现

netty的自适应缓冲区用于接收从channel读取的数据，且可以动态调整缓冲区大小，减少内存的浪费。当连续两次读取的字节数小于当前缓冲区大小时，就会缩小，当读取的字节数刚好等于缓冲区大小时，就会扩大。

因为是缓冲区是存放从channel读取的数据，所以缓冲区是在生成channel对象时而生成的，以NioServerSocketChannel为例子，看自适应缓冲区的实例化过程

public NioServerSocketChannel(ServerSocketChannel channel) {
    super(null, channel, SelectionKey.OP_ACCEPT);
    config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}

netty中，对channel的相关配置参数包括缓冲区都是放在ChannelConfig对象中，因此再跟进channelConfig的构造方法时，会看到如下

public DefaultChannelConfig(Channel channel) {
        this(channel, new AdaptiveRecvByteBufAllocator());
    }

    protected DefaultChannelConfig(Channel channel, RecvByteBufAllocator allocator) {
        setRecvByteBufAllocator(allocator, channel.metadata());
        this.channel = channel;
    }

可以看出先生成一个自适应缓冲区分配器对象，然后设置到channelConfig中。通过分配器来分配缓冲区ByteBuf对象，所以，接下来我们先看下分配器的内部重要成员变量，最后再看下方法的处理逻辑

AdaptiveRecvByteBufAllocator

SIZE_TABLE
整型数组，数组第一个元素的值为16，接下来的每个元素值依次再前一个值基础上+16，到了512之后的每个元素值是前一个元素值的2倍。缓冲区大小在调节过程中的值都是从这个数组取的。
getSizeTableIndex
这个方法用于传入一个值，可以获取到该值对应在SIZE_TABLE的数组下标，用的是二分查找法
initial、minIndex、maxIndex
initial表示缓冲区初始化的默认大小，minIndex表示当缓冲区变小时，最小值对应的数组下标，同理可得，naxIndex表示当缓冲区变大时，最大值对应的数组下标
INDEX_INCREMENT、INDEX_DECREMENT
INDEX_INCREMENT默认值为4，表示缓冲区每次增大，值为SIZE_TABLE[index+4]，index为当前值对应的下标。INDEX_DECREMENT默认为1，表示缓冲区每次减小，值为SIZE_TABLE[index-1]

通过RecvByteBufAllocator接口定义可知，allocator内部逻辑的处理又是委托给内部类Handle处理的，因此我们看下AdaptiveRecvByteBufAllocator内部HandleImpl的几个重要属性

//缓冲区最小值对应的数组下标
private final int minIndex;
//缓冲区最大值对应的数组下标
private final int maxIndex;
//缓冲区当前值对应数组的下标
private int index;
//下一次缓冲区的大小
private int nextReceiveBufferSize;
//当前是否需要减小缓冲区的大小
private boolean decreaseNow;

HandleImpl的几个重要属性其实也是基于外部类AdaptiveRecvByteBufAllocator的。
AdaptiveRecvByteBufAllocator只是和缓冲区大小的设置相关，是否需要扩大或者缩小缓冲区，要看下其继承的类DefaultMaxMessagesRecvByteBufAllocator

DefaultMaxMessagesRecvByteBufAllocator

maxMessagesPerRead
之前在nio的编程中，从channel中获取数据可以放在一个循环里面，直到channel没数据可读取为止。也可以每次只读一次，如果有数据，下次select循环中，又会触发read事件。在netty中，就是用这个值来控制每次从channel读取数据的循环次数。
respectMaybeMoreData
表示是否有更多数据可以读取。
相应的DefaultMaxMessagesRecvByteBufAllocator也有一个内部类MaxMessageHandle，实际处理逻辑的。

//每次读取数据的循环次数
private int maxMessagePerRead;
//总读取次数
private int totalMessages;
//总读取字节数
private int totalBytesRead;
//尝试读取的字节数
private int attemptedBytesRead;
//上次读取的字节数
private int lastBytesRead;
private final boolean respectMaybeMoreData = DefaultMaxMessagesRecvByteBufAllocator.this.respectMaybeMoreData;
private final UncheckedBooleanSupplier defaultMaybeMoreSupplier = new UncheckedBooleanSupplier() {
    @Override
    public boolean get() {
        return attemptedBytesRead == lastBytesRead;
    }
};

在了解了AdaptiveRecvByteBufAllocator和DefaultMaxMessagesRecvByteBufAllocator基本属性及内部委托处理的Handle（HandleImpl和MaxMessageHandle）之后，结合实际的read操作，来看下netty是如何实现自适应调整缓冲区大小的。

netty中读取socket的数据时在NioEventLoop的循环中处理的，这个可以先下我之前写的另一篇文章NioEventLoop事件循环处理，直接跟踪到这行代码

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

这里就以NioSocketChannel的unsafe为例子来看下读取逻辑。

public final void read() {
    final ChannelConfig config = config();
    if (shouldBreakReadReady(config)) {
        clearReadPending();
        return;
    }
    final ChannelPipeline pipeline = pipeline();
    final ByteBufAllocator allocator = config.getAllocator();
//通过之前的分析，这里拿到的是AdaptiveRecvByteBufAllocator的HandleImpl实例对象
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
//这里是重置HandleImpl的一些参数为默认初始值，比如maxMessagePerRead=16，totalMessages = totalBytesRead = 0;
    allocHandle.reset(config);
    ByteBuf byteBuf = null;
    boolean close = false;
    try {
        do {
//分配一个初始的ByteBuf,大小为1024，在创建自适应缓冲区时的默认参数，大小是通过HandleImpl的guess方法拿到 
 //的，guess方法返回的是nextReceiveBufferSize值
            byteBuf = allocHandle.allocate(allocator);
//这里的两个步骤，doReadBytes会设置Handle的attemptedBytesRead值，也就是byteBuf的可写字节数，然后将 
 //从channel读取attemptedBytesRead的字节数到bytebuf中，当然了，实际读取的字节数不一定那么多
//doReadBytes方法返回实际读取到的字节数，并传给handle的lastBytesRead方法。这个方法的逻辑在下面单独分析
            allocHandle.lastBytesRead(doReadBytes(byteBuf));
//这次读取的字节数如果小于等于0，有两种情况，=0的话是此次没数据读取了，释放缓冲区即可，若小于0，表示连接断开了，还需要设置readPending 标志位，在finally中处理
            if (allocHandle.lastBytesRead() <= 0) {
                // nothing was read. release the buffer.
                byteBuf.release();
                byteBuf = null;
                close = allocHandle.lastBytesRead() < 0;
                if (close) {
                    // There is nothing left to read as we received an EOF.
                    readPending = false;
                }
                break;
            }
                    //读取完这次数据后，对Handle的totalMessages加1
            allocHandle.incMessagesRead(1);
            readPending = false;
//触发pipeline的的channelRead方法
            pipeline.fireChannelRead(byteBuf);
            byteBuf = null;
//这边的这个判断很重要，决定是否继续循环从channel读取数据，看下面专门的分析
        } while (allocHandle.continueReading());

        allocHandle.readComplete();
        pipeline.fireChannelReadComplete();

//如果是连接关闭了，那么会触发一个用户事件，传递给pipeline，应用程序可以针对这个事件进行捕获处理
        if (close) {
            closeOnRead(pipeline);
        }
    } catch (Throwable t) {
        handleReadException(pipeline, byteBuf, t, close, allocHandle);
    } finally {
              //如果是在read多次循环中间突然连接断开，那么readPending 会等于false，且channelConfig配的不是自动读取，那么需要将这个channel的read事件从selector上移除
        if (!readPending && !config.isAutoRead()) {
                    removeReadOp();
                }
    }
}

HandleImpl的lastBytesRead方法，在上面已经分析了，真实从channel读取的字节数会传给这个方法

public void lastBytesRead(int bytes) {
    //假如真实读取的字节数等于是ByteBuf可写的字节数，也就是ByteBuf在这次读取中，数据完全被填充满了，那么，要进入一个调整缓冲区大小的方法record
    if (bytes == attemptedBytesRead()) {
        record(bytes);
    }
//记录上次读取的真实字节数
    super.lastBytesRead(bytes);
}

/**
这个方法有两个判断逻辑
1. 获取当前缓冲区缩小后的值，假如此次读取的字节数小于等于这个值，说明缓冲区可能需要被缩小，  
但是netty会给两次机会，如果是第一次这样的话，记录一下decreaseNow 标志位为true，如果是第二次的话，才会将 nextReceiveBufferSize值变小，
这样在下次循环分配器在分配缓冲区时，缓冲区大小就变小了。
2. 假如此次读取的字节数大于等于nextReceiveBufferSize了，那么就需要调整nextReceiveBufferSize的值了，下次分配缓冲区时，缓冲区大小就变大了。
缓冲区变大或者变小的参数在一开始介绍AdaptiveRecvByteBufAllocator有说过了，可以翻上去看下
*/
private void record(int actualReadBytes) {
    if (actualReadBytes <= SIZE_TABLE[max(0, index - INDEX_DECREMENT)]) {
        if (decreaseNow) {
            index = max(index - INDEX_DECREMENT, minIndex);
            nextReceiveBufferSize = SIZE_TABLE[index];
            decreaseNow = false;
        } else {
            decreaseNow = true;
        }
    } else if (actualReadBytes >= nextReceiveBufferSize) {
        index = min(index + INDEX_INCREMENT, maxIndex);
        nextReceiveBufferSize = SIZE_TABLE[index];
        decreaseNow = false;
    }
}

allocHandle.continueReading()决定是否继续循环从channel读取数据

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

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

从这个方法可以看出，接着循环读取数据需要满足的条件有几个

channel配置了自动读取数据
当前这个循环读取的字节数刚好等于ByteBuf可写的字节数（也就是attemptedBytesRead ）
已经循环读取的次数小于maxMessagePerRead，默认是16
总读取的字节数大于0

netty自适应缓冲区实现