Netty Review - ByteBuf内存池源码解析

@Override
public final void read() {
    final ChannelConfig config = config();  // 获取通道配置信息
    if (shouldBreakReadReady(config)) {  // 判断是否应该中断读就绪操作
        clearReadPending();  // 清除读等待标志
        return;
    }
    final ChannelPipeline pipeline = pipeline();  // 获取通道的管道
    final ByteBufAllocator allocator = config.getAllocator();  // 获取分配器
    final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();  // 获取接收字节缓冲区分配句柄
    allocHandle.reset(config);  // 重置分配句柄状态

    ByteBuf byteBuf = null;  // 字节缓冲区
    boolean close = false;  // 是否关闭标志
    try {
        do {
            byteBuf = allocHandle.allocate(allocator);  // 分配字节缓冲区
            allocHandle.lastBytesRead(doReadBytes(byteBuf));  // 读取数据到缓冲区
            if (allocHandle.lastBytesRead() <= 0) {
                // 如果没有读取到数据
                // 释放缓冲区
                byteBuf.release();
                byteBuf = null;
                close = allocHandle.lastBytesRead() < 0;  // 是否关闭标志
                if (close) {
                    // 如果收到 EOF，表示没有数据可读了
                    readPending = false;  // 清除读等待标志
                }
                break;
            }

            allocHandle.incMessagesRead(1);  // 增加读取消息数
            readPending = false;  // 清除读等待标志
            pipeline.fireChannelRead(byteBuf);  // 触发通道读事件到管道
            byteBuf = null;
        } while (allocHandle.continueReading());  // 继续读取数据，直到不再需要读取为止

        allocHandle.readComplete();  // 读操作完成
        pipeline.fireChannelReadComplete();  // 触发通道读完成事件到管道

        if (close) {
            closeOnRead(pipeline);  // 如果需要关闭通道，执行关闭操作
        }
    } catch (Throwable t) {
        handleReadException(pipeline, byteBuf, t, close, allocHandle);  // 处理读取异常
    } finally {
        // 检查是否有未处理的读等待操作
        // 这可能有两个原因：
        // 1. 用户在 channelRead(...) 方法中调用了 Channel.read() 或 ChannelHandlerContext.read()
        // 2. 用户在 channelReadComplete(...) 方法中调用了 Channel.read() 或 ChannelHandlerContext.read()
        // 详见 https://github.com/netty/netty/issues/2254
        if (!readPending && !config.isAutoRead()) {
            removeReadOp();  // 移除读操作
        }
    }
}

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

@Override
public ByteBuf ioBuffer(int initialCapacity) {
    if (PlatformDependent.hasUnsafe()) { // 检查当前平台是否支持直接内存
        return directBuffer(initialCapacity); // 如果支持直接内存，则调用 directBuffer() 方法创建直接内存的 ByteBuf 实例
    }
    return heapBuffer(initialCapacity); // 如果不支持直接内存，则调用 heapBuffer() 方法创建堆内存的 ByteBuf 实例
}

   @Override
public ByteBuf directBuffer(int initialCapacity) {
    return directBuffer(initialCapacity, DEFAULT_MAX_CAPACITY); // 调用重载方法 directBuffer(int initialCapacity, int maxCapacity)，传入默认的最大容量值 DEFAULT_MAX_CAPACITY
}

@Override
public ByteBuf directBuffer(int initialCapacity, int maxCapacity) {
    if (initialCapacity == 0 && maxCapacity == 0) { // 如果初始容量和最大容量都为0
        return emptyBuf; // 返回一个空的 ByteBuf 实例
    }
    validate(initialCapacity, maxCapacity); // 验证初始容量和最大容量的合法性
    return newDirectBuffer(initialCapacity, maxCapacity); // 创建一个新的直接内存的 ByteBuf 实例
}

@Override
protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
    // 获取当前线程的线程缓存
    PoolThreadCache cache = threadCache.get();
    // 获取直接内存池
    PoolArena<ByteBuffer> directArena = cache.directArena;

    final ByteBuf buf;
    if (directArena != null) { // 如果直接内存池可用
        // 从直接内存池中分配内存
        buf = directArena.allocate(cache, initialCapacity, maxCapacity);
    } else { // 如果直接内存池不可用
        // 使用平台相关的方式创建直接内存的 ByteBuf 实例
        buf = PlatformDependent.hasUnsafe() ?
            UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity) :
            new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
    }

    // 返回一个包装了泄漏感知器的 ByteBuf 实例
    return toLeakAwareBuffer(buf);
}

PooledByteBuf<T> allocate(PoolThreadCache cache, int reqCapacity, int maxCapacity) {
    // 创建一个新的 PooledByteBuf 实例，其中 maxCapacity 为指定的最大容量
    PooledByteBuf<T> buf = newByteBuf(maxCapacity);
    // 使用指定的线程缓存和请求容量来分配内存给 ByteBuf
    allocate(cache, buf, reqCapacity);
    // 返回分配的 ByteBuf
    return buf;
}

@Override
protected PooledByteBuf<ByteBuffer> newByteBuf(int maxCapacity) {
    // 如果支持 Unsafe
    if (HAS_UNSAFE) {
        // 创建 PooledUnsafeDirectByteBuf 的实例
        return PooledUnsafeDirectByteBuf.newInstance(maxCapacity);
    } else {
        // 创建 PooledDirectByteBuf 的实例
        return PooledDirectByteBuf.newInstance(maxCapacity);
    }
}

static PooledUnsafeDirectByteBuf newInstance(int maxCapacity) {
    // 从对象池中获取 PooledUnsafeDirectByteBuf 实例
    PooledUnsafeDirectByteBuf buf = RECYCLER.get();
    // 重用实例并设置最大容量
    buf.reuse(maxCapacity);
    return buf;
}