• 1 概述
• 2 如何触发Read调用
• 3 具体实现

1 概述

本文介绍Netty客户端read实现原理（主要介绍NIO通道的Read方法），内容比较简单，因为不涉及Netty内存池等相关知识点，Netty内存池原理后续会有专文介绍，但是本文集已经介绍过了Netty源码-PoolThreadCache、Netty源码-PooledByteBufAllocator静态变量初始化等和Netty内存分配相关内容，可结合着理解本文内容。

2 如何触发Read调用

在NIO编程中，如果应用程序需要和远程交换数据，有读数据的需求，我们一般会将通道注册到Selecotr上，并注册OP_READ事件，然后在Selector.select返回的SelectedKey中有读事件继续时，会从通道读取数据（关于Selector事件的说明可看笔者文章NIO SelectionKey事件理解）。Netty作为NIO编程框架也不例外，在NioEventLoop.run方法中会通过Selector.select方法等待感兴趣的事件就绪，如果通道有数据到来，即读事件继续，会调用Unsafe.read方法。本文主要介绍Netty客户端Read方法实现。

3 具体实现

因为我们主要介绍Java NIO实现，所以我们看Netty基于NIO的客户端read实现，其实现源码在NioByteUnsafe.read方法中：

//NioByteUnsafe
@Override
public final void read() {
    //获取config配置
    final ChannelConfig config = config();
    //判断是否因配置或者异常、错误等中断本次读操作
    if (shouldBreakReadReady(config)) {
        //这个函数会取消该通道对OP_READ事件的注册
        //取消之后即不再响应通道read准备好事件
        clearReadPending();
        return;
    }
    final ChannelPipeline pipeline = pipeline();
    //获取ByteBuf分配器
    final ByteBufAllocator allocator = config.getAllocator();
    //获取RecvByteBuf分配器，这个笔者有文章介绍过，RecvByteBuf分配器
    //主要告诉ByteBuf分配器分配ByteBuf的大小，默认配置的是
    //AdaptiveRecvByteBufAllocator
    final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
    //每次read时，重置RecvByteBuf统计的一些信息
    allocHandle.reset(config);

    ByteBuf byteBuf = null;
    boolean close = false;
    try {
        do {
            //分配本次读取操作的ByteBuf
            byteBuf = allocHandle.allocate(allocator);
            //进行实际读取操作，doReadBytes我们后面列出源码
            allocHandle.lastBytesRead(doReadBytes(byteBuf));
            //如果没有读取到数据，直接返回，结束本次读取
            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;
            }

            allocHandle.incMessagesRead(1);
            readPending = false;
            //触发Pipeline的channelRead方法，让
            //用户定义的handler处理读到的数据
            //比如对数据进行解码，然后交给业务层handler处理等
            pipeline.fireChannelRead(byteBuf);
            byteBuf = null;
            //判断是否需要继续读
        } while (allocHandle.continueReading());

        allocHandle.readComplete();
        //触发channelReadComplete方法
        pipeline.fireChannelReadComplete();

        if (close) {
            //上面如果最后读取的字节数小于0，则关闭读取操作
            //但是不关闭通道
            closeOnRead(pipeline);
        }
    } catch (Throwable t) {
        //异常处理
        handleReadException(pipeline, byteBuf, t, close, allocHandle);
    } finally {
        // Check if there is a readPending which was not processed yet.
        // This could be for two reasons:
        // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
        // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
        //
        // See https://github.com/netty/netty/issues/2254
        if (!readPending && !config.isAutoRead()) {
            removeReadOp();
        }
    }
}

private void handleReadException(ChannelPipeline pipeline, ByteBuf byteBuf, Throwable cause, boolean close,
        RecvByteBufAllocator.Handle allocHandle) {
    if (byteBuf != null) {
        if (byteBuf.isReadable()) {
            readPending = false;
            pipeline.fireChannelRead(byteBuf);
        } else {
            byteBuf.release();
        }
    }
    allocHandle.readComplete();
    pipeline.fireChannelReadComplete();
    //看到了熟悉的方法fireExceptionCaught
    //这个是handler定义的，可供应用处理异常
    pipeline.fireExceptionCaught(cause);
    if (close || cause instanceof IOException) {
        closeOnRead(pipeline);
    }
}

因为我们这里介绍客户端读取操作，所以我们看NioSocketChannel.doReadBytes方法

//NioSocketChannel
@Override
protected int doReadBytes(ByteBuf byteBuf) throws Exception {
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    allocHandle.attemptedBytesRead(byteBuf.writableBytes());
    //其实主要是从java nio channel中读取数据
    return byteBuf.writeBytes(javaChannel(), allocHandle.attemptedBytesRead());
}

上面关于RecvByteBufAllocator相关内容可参考笔者文章Netty源码-接收缓冲区分配器AdaptiveRecvByteBufAllocator，关于ChannelPipeline、ChannelHandler的介绍可参考笔者文章Netty源码-ChannelPipeline和ChannelHandler。