Netty Channel源碼分析

時間 2019-11-24

原文原文鏈接

原文連接：wangwei.one/posts/netty…html

前面，咱們大體瞭解了Netty中的幾個核心組件。今天咱們就來先來介紹Netty的網絡通訊組件，用於執行網絡I/O操做 —— Channel。java

Netty版本：4.1.30ios

概述

數據在網絡中老是以字節的形式進行流通。咱們在進行網絡編程時選用何種傳輸方式編碼（OIO、NIO等）決定了這些字節的傳輸方式。編程

在沒有Netty以前，爲了提高系統的併發能力，從OIO切換到NIO時，須要對代碼進行大量的重構，由於相應的Java NIO 與 IO API大不相同。而Netty在這些Java原生API的基礎上作了一層封裝，對用戶提供了高度抽象而又統一的API，從而讓傳輸方式的切換不在變得困難，只須要直接使用便可，而不須要對整個代碼進行重構。segmentfault

Netty Channel UML

netty channel族以下：api

整個族羣中，AbstractChannel 是最爲關鍵的一個抽象類，從它繼承出了AbstractNioChannel、AbstractOioChannel、AbstractEpollChannel、LocalChannel、EmbeddedChannel等類，每一個類表明了不一樣的協議以及相應的IO模型。除了 TCP 協議之外，Netty 還支持不少其餘的鏈接協議，而且每種協議還有 NIO(異步 IO) 和 OIO(Old-IO，即傳統的阻塞 IO) 版本的區別. 不一樣協議不一樣的阻塞類型的鏈接都有不一樣的 Channel 類型與之對應。下面是一些經常使用的 Channel 類型：數組

NioSocketChannel：表明異步的客戶端 TCP Socket 鏈接
NioServerSocketChannel：異步的服務器端 TCP Socket 鏈接
NioDatagramChannel：異步的 UDP 鏈接
NioSctpChannel：異步的客戶端 Sctp 鏈接
NioSctpServerChannel：異步的 Sctp 服務器端鏈接
OioSocketChannel：同步的客戶端 TCP Socket 鏈接
OioServerSocketChannel：同步的服務器端 TCP Socket 鏈接
OioDatagramChannel：同步的 UDP 鏈接
OioSctpChannel：同步的 Sctp 服務器端鏈接
OioSctpServerChannel：同步的客戶端 TCP Socket 鏈接

Channel API

咱們先來看下最頂層接口 channel 主要的API，經常使用的以下：promise

接口名	描述
eventLoop()	Channel須要註冊到EventLoop的多路複用器上，用於處理I/O事件，經過eventLoop()方法能夠獲取到Channel註冊的EventLoop。EventLoop本質上就是處理網絡讀寫事件的Reactor線程。在Netty中，它不只僅用來處理網絡事件，也能夠用來執行定時任務和用戶自定義NioTask等任務。
pipeline()	返回channel分配的ChannelPipeline
isActive()	判斷channel是否激活。激活的意義取決於底層的傳輸類型。例如，一個Socket傳輸一旦鏈接到了遠程節點即是活動的，而一個Datagram傳輸一旦被打開即是活動的
localAddress()	返回本地的socket地址
remoteAddress()	返回遠程的socket地址
flush()	將以前已寫的數據沖刷到底層Channel上去
write(Object msg)	請求將當前的msg經過ChannelPipeline寫入到目標Channel中。注意，write操做只是將消息存入到消息發送環形數組中，並無真正被髮送，只有調用flush操做纔會被寫入到Channel中，發送給對方。
writeAndFlush()	等同於調用write()並接着調用flush()
metadate()	熟悉TCP協議的讀者可能知道，當建立Socket的時候須要指定TCP參數，例如接收和發送的TCP緩衝區大小，TCP的超時時間。是否重用地址等。在Netty中，每一個Channel對應一個物理連接，每一個鏈接都有本身的TCP參數配置。因此，Channel會聚合一個ChannelMetadata用來對TCP參數提供元數據描述信息，經過metadata()方法就能夠獲取當前Channel的TCP參數配置。
read()	從當前的Channel中讀取數據到第一個inbound緩衝區中，若是數據被成功讀取，觸發ChannelHandler.channelRead(ChannelHandlerContext,Object)事件。讀取操做API調用完成後，緊接着會觸發ChannelHander.channelReadComplete（ChannelHandlerContext）事件，這樣業務的ChannelHandler能夠決定是否須要繼續讀取數據。若是已經有操做請求被掛起，則後續的讀操做會被忽略。
close(ChannelPromise promise)	主動關閉當前鏈接，經過ChannelPromise設置操做結果並進行結果通知，不管操做是否成功，均可以經過ChannelPromise獲取操做結果。該操做會級聯觸發ChannelPipeline中全部ChannelHandler的ChannelHandler.close(ChannelHandlerContext，ChannelPromise)事件。
parent()	對於服務端Channel而言，它的父Channel爲空；對於客戶端Channel，它的父Channel就是建立它的ServerSocketChannel。
id()	返回ChannelId對象，ChannelId是Channel的惟一標識。

Channel建立

對Netty Channel API以及相關的類有了一個初步瞭解以後，接下來咱們來詳細瞭解一下在Netty的啓動過程當中Channel是如何建立的。服務端Channel的建立過程，主要分爲四個步驟：1）Channel建立；2）Channel初始化；3）Channel註冊；4）Channel綁定。服務器

咱們如下面的代碼爲例進行解析：網絡

// 建立兩個線程組,專門用於網絡事件的處理，Reactor線程組
// 用來接收客戶端的鏈接，
EventLoopGroup bossGroup = new NioEventLoopGroup();
// 用來進行SocketChannel的網絡讀寫
EventLoopGroup workGroup = new NioEventLoopGroup();

// 建立輔助啓動類ServerBootstrap，並設置相關配置：
ServerBootstrap b = new ServerBootstrap();
// 設置處理Accept事件和讀寫操做的事件循環組
b.group(bossGroup, workGroup)
         // 配置Channel類型
        .channel(NioServerSocketChannel.class)
         // 配置監聽地址
        .localAddress(new InetSocketAddress(port))
         // 設置服務器通道的選項，設置TCP屬性
        .option(ChannelOption.SO_KEEPALIVE, Boolean.TRUE)
         // 設置創建鏈接後的客戶端通道的選項
        .childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
         // channel屬性，便於保存用戶自定義數據
        .attr(AttributeKey.newInstance("UserId"), "60293")
    	.handler(new LoggingHandler(LogLevel.INFO))
        // 設置子處理器，主要是用戶的自定義處理器，用於處理IO網絡事件
        .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(serverHandler);
            }
        });

// 調用bind()方法綁定端口，sync()會阻塞等待處理請求。這是由於bind()方法是一個異步過程，會當即返回一個ChannelFuture對象，調用sync()會等待執行完成
ChannelFuture f = b.bind().sync();
// 得到Channel的closeFuture阻塞等待關閉，服務器Channel關閉時closeFuture會完成
f.channel().closeFuture().sync();
複製代碼

調用channel()接口設置 AbstractBootstrap 的成員變量 channelFactory，該變量顧名思義就是用於建立channel的工廠類。源碼以下：

...

public B channel(Class<? extends C> channelClass) {
    if (channelClass == null) {
        throw new NullPointerException("channelClass");
    }
    // 建立 channelFactory
    return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
}

...

public B channelFactory(ChannelFactory<? extends C> channelFactory) {
    if (channelFactory == null) {
        throw new NullPointerException("channelFactory");
    }
    
    if (this.channelFactory != null) {
        throw new IllegalStateException("channelFactory set already");
    }
    this.channelFactory = channelFactory;
    return (B) this;
}

...

複製代碼

channelFactory 設置爲 ReflectiveChannelFactory ，在咱們這個例子中 clazz 爲 NioServerSocketChannel ，咱們能夠看到其中有個 newChannel() 接口，經過反射的方式來調用，這個接口的調用處咱們後面會介紹到。源碼以下：

// Channel工廠類
public class ReflectiveChannelFactory<T extends Channel> implements ChannelFactory<T> {

    private final Class<? extends T> clazz;

    public ReflectiveChannelFactory(Class<? extends T> clazz) {
        if (clazz == null) {
            throw new NullPointerException("clazz");
        }
        this.clazz = clazz;
    }
	
    @Override
    public T newChannel() {
        try {
            // 經過反射來進行常見Channel實例
            return clazz.newInstance();
        } catch (Throwable t) {
            throw new ChannelException("Unable to create Channel from class " + clazz, t);
        }
    }

    @Override
    public String toString() {
        return StringUtil.simpleClassName(clazz) + ".class";
    }
}
複製代碼

接下來咱們來看下 NioServerSocketChannel 的構造函數，主要就是：

生成ServerSocketChannel對象。NioServerSocketChannel建立時，首先使用SelectorProvider的openServerSocketChannel打開服務器套接字通道。SelectorProvider是Java的NIO提供的抽象類，是選擇器和可選擇通道的服務提供者。具體的實現類有SelectorProviderImpl，EPollSelectorProvide，PollSelectorProvider。選擇器的主要工做是根據操做系統類型和版本選擇合適的Provider：若是LInux內核版本>=2.6則，具體的SelectorProvider爲EPollSelectorProvider，不然爲默認的PollSelectorProvider。
設置 ServerSocketChannelConfig 成員變量。

private static ServerSocketChannel newSocket(SelectorProvider provider) {
    try {
        // 調用JDK底層API生成 ServerSocketChannel 對象實例
        return provider.openServerSocketChannel();
    } catch (IOException e) {
        throw new ChannelException("Failed to open a server socket.", e);
    }
}

private final ServerSocketChannelConfig config;

public NioServerSocketChannel() {
    this(newSocket(DEFAULT_SELECTOR_PROVIDER));
}

public NioServerSocketChannel(SelectorProvider provider) {
    this(newSocket(provider));
}

public NioServerSocketChannel(ServerSocketChannel channel) {
    // 調用 AbstractNioChannel 構造器，建立 NioServerSocketChannel，設置SelectionKey爲ACCEPT
    super(null, channel, SelectionKey.OP_ACCEPT);
    // 建立ChannleConfig對象,主要是TCP參數配置類
    config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}
複製代碼

AbstractNioChannel 的構造器以下：

protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
    // 調用 AbstractChannel 構造器
    super(parent);
    this.ch = ch;
    // 從上一步過來，這裏設置爲 SelectionKey.OP_ACCEPT
    this.readInterestOp = readInterestOp;
    try {
        // 設置爲非阻塞狀態
        ch.configureBlocking(false);
    } catch (IOException e) {
        try {
            ch.close();
        } catch (IOException e2) {
            if (logger.isWarnEnabled()) {
                logger.warn("Failed to close a partially initialized socket.", e2);
            }
        }
        throw new ChannelException("Failed to enter non-blocking mode.", e);
    }
}
複製代碼

在 AbstractChannel 構造器中，會設Channel關聯的三個核心對象：ChannelId、ChannelPipeline、Unsafe。

初始化ChannelId，ChannelId是一個全局惟一的值；
建立 NioMessageUnsafe 實例，該類爲Channel提供了用於完成網絡通信相關的底層操做，如connect(),read(),register(),bind(),close()等；
爲Channel建立DefaultChannelPipeline，初始事件傳播管道。關於Pipeline的分析，請看後文的分析。

protected AbstractChannel(Channel parent) {
    this.parent = parent;
    // 設置ChannelId
    id = newId();
    // 設置Unsafe
    unsafe = newUnsafe();
    // 設置Pipeline
    pipeline = newChannelPipeline();
}
複製代碼

從 NioServerSocketChannelConfig 的構造函數追溯下去，在 DefaultChannelConfig 會設置channel成員變量。

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

protected DefaultChannelConfig(Channel channel, RecvByteBufAllocator allocator) {
    setRecvByteBufAllocator(allocator, channel.metadata());
    // 綁定channel
    this.channel = channel;
}
複製代碼

以上就是channel建立的過程，總結一下:

經過 ReflectiveChannelFactory 工廠類，以反射的方式對channel進行建立；
channel建立的過程當中，會建立四個重要的對象：ChannelId、ChannelConfig、ChannelPipeline、Unsafe。

Channel初始化

主要分爲如下兩步：

將啓動器（Bootstrap）設置的選項和屬性設置到NettyChannel上面
向Pipeline添加初始化Handler，供註冊後使用

咱們從 AbstractBootstrap 的 bind() 接口進去，調用鏈：bind() —> doBind(localAddress) —> initAndRegister() —> init(Channel channel)，咱們看下 ServerBootstrap 中 init() 接口的實現：

final ChannelFuture initAndRegister() {
    Channel channel = null;
    try {
        // 調用Channel工程類的newChannel()接口，建立channel，就是前面咱們講的部份內容
        channel = channelFactory.newChannel();
        // 初始化channel
        init(channel);
    } catch (Throwable t) {
        ....
}
複製代碼

初始化Channel，咱們來重點看下 init(channel) 接口：

void init(Channel channel) throws Exception {
    // 獲取啓動器 啓動時配置的option參數，主要是TCP的一些屬性
    final Map<ChannelOption<?>, Object> options = options0();
    // 將得到到 options 配置到 ChannelConfig 中去
    synchronized (options) {
        setChannelOptions(channel, options, logger);
    }

    // 獲取 ServerBootstrap 啓動時配置的 attr 參數
    final Map<AttributeKey<?>, Object> attrs = attrs0();
    // 配置 Channel attr，主要是設置用戶自定義的一些參數
    synchronized (attrs) {
        for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
            @SuppressWarnings("unchecked")
            AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
            channel.attr(key).set(e.getValue());
        }
    }
	
    // 獲取channel中的 pipeline，這個pipeline使咱們前面在channel建立過程當中設置的 pipeline
    ChannelPipeline p = channel.pipeline();

    // 將啓動器中配置的 childGroup 保存到局部變量 currentChildGroup
    final EventLoopGroup currentChildGroup = childGroup;
    // 將啓動器中配置的 childHandler 保存到局部變量 currentChildHandler
    final ChannelHandler currentChildHandler = childHandler;
    final Entry<ChannelOption<?>, Object>[] currentChildOptions;
    final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
    // 保存用戶設置的 childOptions 到局部變量 currentChildOptions
    synchronized (childOptions) {
        currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));
    }
    // 保存用戶設置的 childAttrs 到局部變量 currentChildAttrs
    synchronized (childAttrs) {
        currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
    }

    p.addLast(new ChannelInitializer<Channel>() {
        @Override
        public void initChannel(final Channel ch) throws Exception {
            final ChannelPipeline pipeline = ch.pipeline();
            // 獲取啓動器上配置的handler
            ChannelHandler handler = config.handler();
            if (handler != null) {
                // 添加 handler 到 pipeline 中
                pipeline.addLast(handler);
            }
            ch.eventLoop().execute(new Runnable() {
                @Override
                public void run() {
                    // 用child相關的參數建立出一個新鏈接接入器ServerBootstrapAcceptor
                    // 經過 ServerBootstrapAcceptor 能夠將一個新鏈接綁定到一個線程上去
                    // 每次有新的鏈接進來 ServerBootstrapAcceptor 都會用child相關的屬性對它們進行配置，並註冊到ChaildGroup上去
                    pipeline.addLast(new ServerBootstrapAcceptor(
                            ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                }
            });
        }
    });
}
複製代碼

對於新鏈接接入器 ServerBootstrapAcceptor 的分析，請查看後文

Channel註冊

在channel完成建立和初始化以後，接下來就須要將其註冊到事件輪循器Selector上去。咱們回到 initAndRegister 接口上去：

final ChannelFuture initAndRegister() {

    ...

    // 獲取 EventLoopGroup ，並調用它的 register 方法來註冊 channel
    ChannelFuture regFuture = config().group().register(channel);
    if (regFuture.cause() != null) {
        if (channel.isRegistered()) {
            channel.close();
        } else {
            channel.unsafe().closeForcibly();
        }
    }
    return regFuture;
}
複製代碼

最終會向下調用到 SingleThreadEventLoop 中的 register 接口：

如何調用到這裏，裏面的細節須要等到後面文章講到 MultithreadEventExecutorGroup 再詳細說明

@Override
public ChannelFuture register(final ChannelPromise promise) {
    ObjectUtil.checkNotNull(promise, "promise");
    // 調用unsafe的register接口
    promise.channel().unsafe().register(this, promise);
    return promise;
}
複製代碼

代碼跟蹤下去，直到 AbstractChannel 中的 AbstractUnsafe 這個類中的 register 接口。

@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
    if (eventLoop == null) {
        throw new NullPointerException("eventLoop");
    }
    if (isRegistered()) {
        promise.setFailure(new IllegalStateException("registered to an event loop already"));
        return;
    }
    if (!isCompatible(eventLoop)) {
        promise.setFailure(
                new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
        return;
    }

    // 將該Channel與eventLoop 進行綁定，後續與該channel相關的IO操做都由eventLoop來處理
    AbstractChannel.this.eventLoop = eventLoop;
	// 初次註冊時 eventLoop.inEventLoop() 返回false
    if (eventLoop.inEventLoop()) {
        // 調用實際的註冊接口register0
        register0(promise);
    } else {
        try {
            eventLoop.execute(new Runnable() {
                @Override
                public void run() {
                    // 調用實際的註冊接口register0
                    register0(promise);
                }
            });
        } catch (Throwable t) {
            logger.warn(
                    "Force-closing a channel whose registration task was not accepted by an event loop: {}",
                    AbstractChannel.this, t);
            closeForcibly();
            closeFuture.setClosed();
            safeSetFailure(promise, t);
        }
    }
}
複製代碼

register0接口主要分爲如下三段邏輯：

doRegister();
pipeline.invokeHandlerAddedIfNeeded();
pipeline.fireChannelRegistered();

private void register0(ChannelPromise promise) {
    try {
        if (!promise.setUncancellable() || !ensureOpen(promise)) {
            return;
        }
        boolean firstRegistration = neverRegistered;
        // 調用 doRegister() 接口
        doRegister();
        neverRegistered = false;
        registered = true;
        
       	// 經過pipeline的傳播機制，觸發handlerAdded事件
        pipeline.invokeHandlerAddedIfNeeded();
        safeSetSuccess(promise);
        // 經過pipeline的傳播機制，觸發channelRegistered事件
        pipeline.fireChannelRegistered();
        // 尚未綁定，因此這裏的 isActive() 返回false.
        if (isActive()) {
            if (firstRegistration) {
                pipeline.fireChannelActive();
            } else if (config().isAutoRead()) {
                beginRead();
            }
        }
    } catch (Throwable t) {
        closeForcibly();
        closeFuture.setClosed();
        safeSetFailure(promise, t);
    }
}
複製代碼

咱們來看 AbstractNioChannel 中的 doRegister()接口，最終調用的就是Java JDK底層的NIO API來註冊。

@Override
protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
            // eventLoop().unwrappedSelector()：獲取selector，將在後面介紹 EventLoop 建立時會講到
            // 將selector註冊到Java NIO Channel上
            // ops 設置爲 0，表示不關心任何事件
            // att 設置爲 channel自身，表示後面還會將channel取出來用做它用(後面文章會講到)
            selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
            return;
        } catch (CancelledKeyException e) {
            if (!selected) {
                eventLoop().selectNow();
                selected = true;
            } else {
                throw e;
            }
        }
    }
}
複製代碼

Channel綁定

在完成建立、初始化以及註冊以後，接下來就是Channel綁定操做。

本小節涉及到的pipeline事件傳播機制，咱們放到後面的文章中去講解。

從啓動器的bind()接口開始，往下調用 doBind() 方法：

private ChannelFuture doBind(final SocketAddress localAddress) {
    // 初始化及註冊
    final ChannelFuture regFuture = initAndRegister();
    final Channel channel = regFuture.channel();
    if (regFuture.cause() != null) {
        return regFuture;
    }

    if (regFuture.isDone()) {
        // At this point we know that the registration was complete and successful.
        ChannelPromise promise = channel.newPromise();
        // 調用 doBind0
        doBind0(regFuture, channel, localAddress, promise);
        return promise;
    } else {
        ....
    }
}
複製代碼

doBind 方法又會調用 doBind0() 方法，在doBind0()方法中會經過EventLoop去執行channel的bind()任務，關於EventLoop的execute接口的分析，請看後面的文章。

private static void doBind0( final ChannelFuture regFuture, final Channel channel, final SocketAddress localAddress, final ChannelPromise promise) {

    channel.eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            if (regFuture.isSuccess()) {
                // 調用channel.bind接口
                channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            } else {
                promise.setFailure(regFuture.cause());
            }
        }
    });
}
複製代碼

doBind0() 方法往下會條用到 pipeline.bind(localAddress, promise); 方法，經過pipeline的傳播機制，最終會調用到 AbstractChannel.AbstractUnsafe.bind() 方法，這個方法主要作兩件事情：

調用doBind()：調用底層JDK API進行Channel的端口綁定。
調用pipeline.fireChannelActive()：

關於Pipeline的傳播機制，請看後文

@Override
public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
    
    ....
    
    // wasActive 在綁定成功前爲 false
    boolean wasActive = isActive();
    try {
        // 調用doBind()調用JDK底層API進行端口綁定
        doBind(localAddress);
    } catch (Throwable t) {
        safeSetFailure(promise, t);
        closeIfClosed();
        return;
    }
	// 完成綁定以後，isActive() 返回true
    if (!wasActive && isActive()) {
        invokeLater(new Runnable() {
            @Override
            public void run() {
                // 觸發channelActive事件
                pipeline.fireChannelActive();
            }
        });
    }
    safeSetSuccess(promise);
}
複製代碼

咱們這裏看服務端 NioServerSocketChannel 實現的 doBind方法，最終會調用JDK 底層 NIO Channel的bind方法：

@Override
protected void doBind(SocketAddress localAddress) throws Exception {
    if (PlatformDependent.javaVersion() >= 7) {
        javaChannel().bind(localAddress, config.getBacklog());
    } else {
        javaChannel().socket().bind(localAddress, config.getBacklog());
    }
}
複製代碼

調用 pipeline.fireChannelActive()，開始傳播active事件，pipeline首先就會調用HeadContext節點進行事件傳播，會調用到 DefaultChannelPipeline.HeadContext.channelActive() 方法：

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    // 觸發heanlder 的 ChannelActive 方法
    ctx.fireChannelActive();
    // 調用接口readIfIsAutoRead
    readIfIsAutoRead();
}

private void readIfIsAutoRead() {
    if (channel.config().isAutoRead()) {
        // 調用channel.read()
        channel.read();
    }
}
複製代碼

channel.read() 方法往下會調用到 AbstractChannelHandlerContext.read() 方法：

@Override
public ChannelHandlerContext read() {
    // 獲取下一個ChannelHandlerContext節點
    final AbstractChannelHandlerContext next = findContextOutbound();
    // 獲取EventExecutor
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        // 調用下一個節點的invokeRead接口
        next.invokeRead();
    } else {
        Runnable task = next.invokeReadTask;
        if (task == null) {
            next.invokeReadTask = task = new Runnable() {
                @Override
                public void run() {
                    next.invokeRead();
                }
            };
        }
        executor.execute(task);
    }
    return this;
}
複製代碼

經過pipeline的事件傳播機制，最終會調用到 AbstractChannel.AbstractUnsafe.beginRead() 方法：

@Override
public final void beginRead() {
    assertEventLoop();
    if (!isActive()) {
        return;
    }
    try {
        // 調用 doBeginRead();
        doBeginRead();
    } catch (final Exception e) {
        invokeLater(new Runnable() {
            @Override
            public void run() {
                pipeline.fireExceptionCaught(e);
            }
        });
        close(voidPromise());
    }
}
複製代碼

咱們看下 AbstractNioChannel 對doBeginRead接口的實現邏輯：

// 註冊一個OP_ACCEPT
@Override
protected void doBeginRead() throws Exception {
    // Channel.read() or ChannelHandlerContext.read() was called
    // 獲取channel註冊是的設置的 selectionKey
    final SelectionKey selectionKey = this.selectionKey;
    // selectionKey無效則返回
    if (!selectionKey.isValid()) {
        return;
    }
	
    readPending = true;
	// 前面講到channel在註冊的時候，這是 interestOps 設置的是 0
    final int interestOps = selectionKey.interestOps();
    // readInterestOp 在前面講到channel建立的時候，設置值爲 SelectionKey.OP_ACCEPT
    if ((interestOps & readInterestOp) == 0) {
        // 最終 selectionKey 的興趣集就會設置爲 SelectionKey.OP_ACCEPT
        // 表示隨時能夠接收新鏈接的接入
        selectionKey.interestOps(interestOps | readInterestOp);
    }
}
複製代碼