Netty Review - 底层零拷贝源码解析

@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;
}

private void allocate(PoolThreadCache cache, PooledByteBuf<T> buf, final int reqCapacity) {
    // 将请求容量规范化为标准容量
    final int normCapacity = normalizeCapacity(reqCapacity);
    // 如果容量小于页面大小，则分配小块或微型内存
    if (isTinyOrSmall(normCapacity)) {
        int tableIdx;
        PoolSubpage<T>[] table;
        boolean tiny = isTiny(normCapacity);
        if (tiny) { // < 512
            if (cache.allocateTiny(this, buf, reqCapacity, normCapacity)) {
                // 从缓存中成功分配，则直接返回
                return;
            }
            tableIdx = tinyIdx(normCapacity);
            table = tinySubpagePools;
        } else {
            if (cache.allocateSmall(this, buf, reqCapacity, normCapacity)) {
                // 从缓存中成功分配，则直接返回
                return;
            }
            tableIdx = smallIdx(normCapacity);
            table = smallSubpagePools;
        }

        final PoolSubpage<T> head = table[tableIdx];

        // 同步处理双向链表的头部
        synchronized (head) {
            final PoolSubpage<T> s = head.next;
            if (s != head) {
                // 分配内存
                assert s.doNotDestroy && s.elemSize == normCapacity;
                long handle = s.allocate();
                assert handle >= 0;
                s.chunk.initBufWithSubpage(buf, null, handle, reqCapacity);
                incTinySmallAllocation(tiny);
                return;
            }
        }
        // 没有可用的内存块，则进入分配普通内存块的逻辑
        synchronized (this) {
            allocateNormal(buf, reqCapacity, normCapacity);
        }

        incTinySmallAllocation(tiny);
        return;
    }
    // 大于页面大小的内存分配
    if (normCapacity <= chunkSize) {
        if (cache.allocateNormal(this, buf, reqCapacity, normCapacity)) {
            // 从缓存中成功分配，则直接返回
            return;
        }
        synchronized (this) {
            allocateNormal(buf, reqCapacity, normCapacity);
            ++allocationsNormal;
        }
    } else {
        // 大块内存分配
        allocateHuge(buf, reqCapacity);
    }
}