# ByteBuffer.java
// 创建堆外内存Buffer
public static ByteBuffer allocateDirect(int capacity) {
    return new DirectByteBuffer(capacity);
}
// 创建堆内内存Buffer
public static ByteBuffer allocate(int capacity) {
    if (capacity < 0)
        throw new IllegalArgumentException();
    return new HeapByteBuffer(capacity, capacity);
}

ByteBufAllocator alloc = PooledByteBufAllocator.DEFAULT;
ByteBuf buf = alloc.directBuffer(1024); // 分配内存
...
buf.release(); // 内存还给对象池

private void processSelectedKeys() {
    if (selectedKeys != null) {
        // flip 切换写入数组，并返回旧的数组
        processSelectedKeysOptimized(selectedKeys.flip());
    } else {
        processSelectedKeysPlain(selector.selectedKeys());
    }
}

/**
 * 
 * @param k selectedKey
 * @param ch key通过attachment保存的channel, 亦即触发此key的ch
 */
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
    final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
    if (!k.isValid()) {
        // (省略) key不可用, 异常处理...
    }

    try {
        int readyOps = k.readyOps();
        
        // (省略)... OP_ACCEPT事件的处理
        // (省略)... OP_WRITE事件的处理
        
        // 可读事件 or 网络连接 (断开连接也在这里进入)
        if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
            unsafe.read();
            if (!ch.isOpen()) {
                // Connection already closed - no need to handle write.
                return;
            }
        }
    } catch (CancelledKeyException ignored) {
        unsafe.close(unsafe.voidPromise());
    }
}

# AbstractNioByteChannel.java
public final void read() {
    final ChannelConfig config = config();
    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 {
            // 核心代码-1 创建ByteBuf缓存
            byteBuf = allocHandle.allocate(allocator);
            // 执行消费, 写入ByteBuf
            allocHandle.lastBytesRead(doReadBytes(byteBuf));
            // 无新消息, 释放buffer
            if (allocHandle.lastBytesRead() <= 0) {
                // nothing was read. release the buffer.
                byteBuf.release();
                byteBuf = null;
                // 遵循SocketChannel的定义，若为负，表明连接已关闭
                close = allocHandle.lastBytesRead() < 0;
                break;
            }

            // 增加读入消息数量
            allocHandle.incMessagesRead(1);
            readPending = false;
            // 触发pipeline 消费消息 事件
            pipeline.fireChannelRead(byteBuf);
            byteBuf = null;
        } while (allocHandle.continueReading()); // 判断是否要继续读取

        // 核心代码-2 读取结束后动作, 对于自适应的Allocator会根据此次读取的字节数调整
        allocHandle.readComplete();
        // 触发pipeline 读取消息完成 事件
        pipeline.fireChannelReadComplete();

        if (close) {
            // 关闭pipeline
            closeOnRead(pipeline);
        }
    } catch (Throwable t) {
        // 异常处理，注意在异常处理之前也会触发Channel的readComplete事件
        handleReadException(pipeline, byteBuf, t, close, allocHandle);
    } finally {
        // (省略) 非autoRead时复位interestOps
    }
}

# ByteBufUtil.java
static final ByteBufAllocator DEFAULT_ALLOCATOR;

static {
    /** 池化内存开关. 默认情况, 除了Android系统使用pooled */
    String allocType = SystemPropertyUtil.get(
            "io.netty.allocator.type", PlatformDependent.isAndroid() ? "unpooled" : "pooled");
    allocType = allocType.toLowerCase(Locale.US).trim();

    ByteBufAllocator alloc;
    if ("unpooled".equals(allocType)) {
        alloc = UnpooledByteBufAllocator.DEFAULT;
        logger.debug("-Dio.netty.allocator.type: {}", allocType);
    } else if ("pooled".equals(allocType)) {
        alloc = PooledByteBufAllocator.DEFAULT;
        logger.debug("-Dio.netty.allocator.type: {}", allocType);
    } else {
        alloc = PooledByteBufAllocator.DEFAULT;
        logger.debug("-Dio.netty.allocator.type: pooled (unknown: {})", allocType);
    }

    DEFAULT_ALLOCATOR = alloc;

    THREAD_LOCAL_BUFFER_SIZE = SystemPropertyUtil.getInt("io.netty.threadLocalDirectBufferSize", 64 * 1024);
    logger.debug("-Dio.netty.threadLocalDirectBufferSize: {}", THREAD_LOCAL_BUFFER_SIZE);

    MAX_CHAR_BUFFER_SIZE = SystemPropertyUtil.getInt("io.netty.maxThreadLocalCharBufferSize", 16 * 1024);
    logger.debug("-Dio.netty.maxThreadLocalCharBufferSize: {}", MAX_CHAR_BUFFER_SIZE);
}

# PooledByteBufAllocator.java
public static final PooledByteBufAllocator DEFAULT =
        new PooledByteBufAllocator(PlatformDependent.directBufferPreferred());

byteBuf = allocHandle.allocate(allocator);

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

# DefaultMaxMessagesRecvByteBufAllocator.MaxMessageHandle.java
/**
 * 判断一个channel是否应该继续读取数据，满足以下所有条件才继续:
 * 1. config.isAutoRead() = true
 * 2. attemptedBytesRead == lastBytesRead, 即本次读取消息填满了ByteBuf
 * 3. totalMessages < maxMessagePerRead, 连续读取消息的次数未超过最大值， 默认为16
 * 4. totalBytesRead < Integer.MAX_VALUE, 读取的消息字节数大小未超最大值
 */
public boolean continueReading() {
    return config.isAutoRead() &&
           attemptedBytesRead == lastBytesRead &&
           totalMessages < maxMessagePerRead &&
           totalBytesRead < Integer.MAX_VALUE;
}