本文是Netty文集中“Netty 源码解析”系列的文章。主要对Netty的重要流程以及类进行源码解析，以使得我们更好的去使用Netty。Netty是一个非常优秀的网络框架，对其源码解读的过程也是不断学习的过程。

预备知识

首先，我们知道JDK NIO的Selector实现了I/O多路复用。可以通过一个线程来管理多个Socket。我们可以将多个Channel(一个Channel代表了一个Socket)注册到一个Selector上，并且设置其感兴趣的事件。这样一来，在Selector.select操作时，若发现Channel有我们所感兴趣的事件发生时Selector就会将其记录下来(即，SelectedKeys)，然后我们就可以对事件进行相应的处理了。更多关于JDK NIO的知识请参阅关于 NIO 你不得不知道的一些“地雷”

ServerSocketChannel的有效事件为OP_ACCEPT。
SocketChannel的有效事件为OP_CONNECT、OP_READ、OP_WRITE

SelectionKey.OP_ACCEPT 事件处理流程

当服务端收到客户端的一个连接请求时，‘SelectionKey.OP_ACCEPT’将会触发。在NioEventLoop的事件循环中会对该事件进行处理：

if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
    unsafe.read();
}

我们来看看unsafe.read()的实现，在NioServerSocketChannel中unsafe是一个NioMessageUnsafe实例：

① 对于ACCEPT事件，每次读循环执行一次读操作(但并没有读取任何字节数据，totalBytesRead > 0 为false)这也是符合NIO规范的，因为每次ACCEPT事件被触发时，仅表示有一个客户端向服务器端发起了连接请求。
② doReadMessages(readBuf)：这步主要是通过serverSocket.accpet()来接受生成和客户端通信的socket，并将其放入到readBuf集合中。

        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            final Channel child = (Channel) msg;

            child.pipeline().addLast(childHandler);

            setChannelOptions(child, childOptions, logger);

            for (Entry<AttributeKey<?>, Object> e: childAttrs) {
                child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
            }

            try {
                childGroup.register(child).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (!future.isSuccess()) {
                            forceClose(child, future.cause());
                        }
                    }
                });
            } catch (Throwable t) {
                forceClose(child, t);
            }
        }

a) 将我们在启动流程中设置的childHandler(如，「serverBootstrap.childHandler(new MyServerInitializer())」)以及childOptions(如，「serverBootstrap.childOption(ChannelOption.ALLOW_HALF_CLOSURE, true)」)、childAttrs(如，「serverBootstrap.childAttr(AttributeKey.valueOf("userID"), UUID.randomUUID().toString())」)设置到这个NioSocketChannel中。
b) 将配置好的NioSocketChannel注册到childGroup中，并注册了一个监听器，因为注册是一个异步操作，该监听器会得以调用在注册操作完成时(完成包括，成功的完成、失败的完成、取消的完成)。该监听器实现：如果发现该注册操作失败了，则会强制关闭当前这个NioSocketChannel。
而将NioSocketChannel注册到childGroup的操作和将NioServerSocketChannel注册到parentGroup的流程也是极其类似的。详细的说明请参阅Netty 源码解析 ——— 服务端启动流程 (下)。这里做一个简单的概述。
整个注册流程：
a) 首先会通过轮询的方式从childGroup中获取一个NioEventLoop，将当前的NioSocketChannel注册到这个NioEventLoop上。
b) 将当前的SocketChannel注册到NioEventLoop中的Selector上，并将NioSocketChannel作为附加属性设置到SelectionKey中。
c) 回调我们自定义的ChannelInitializer的initChannel方法，将我们定义的一个个ChannelHandler添加到当前NioSocketChannel所关联的ChannelPipeline上，然后将ChannelInitializer本身从ChannelPipeline中移除。
d) 标记注册这个异步操作标志为成功完成。
e) 触发ChannelRegistered事件，该事件会在ChannelPipeline中得以传播。
f) 触发ChannelActive事件，该事件也是一个入站事件，它会从ChannelPipeline中的head开始传播，而head的channelActive方法除了将ChannelActive事件传播给下一个ChannelInboundHandler之外，还调用一个readIfIsAutoRead()方法。
而readIfIsAutoRead()最终会调用到「AbstractNioChannel.doBeginRead()」方法：

    protected void doBeginRead() throws Exception {
        // Channel.read() or ChannelHandlerContext.read() was called
        final SelectionKey selectionKey = this.selectionKey;
        if (!selectionKey.isValid()) {
            return;
        }

        readPending = true;

        final int interestOps = selectionKey.interestOps();
        if ((interestOps & readInterestOp) == 0) {
            selectionKey.interestOps(interestOps | readInterestOp);
        }
    }

而这个方法中就完成了将readInterestOp(即，上面在说明NioSocketChannel的构建过程中已经提及，该成员变量值为SelectionKey.OP_READ)设置为感兴趣的事件。这样一来，Selector就会监听该SocketChannel的读事件了。

到目前为止，NioServerSocketChannel通过accept接受了一个客户端的连接请求的整个流程就完成了。这里NioSocketChannel的创建是由NioServerSocketChannel所在的NioEventLoop( 实际上是NioEventLoop所在的线程上 )完成的。然后将创建好的NioSocketChannel注册到childGroup中，也就是通过轮询的方式的方式从childGroup中获取一个NioEventLoop，然后将NioSocketChannel注册的其上。在注册的过程中也完成了将SocketChannel注册到Selector，并设置SelectionKey.OP_READ为感兴趣的事件，这样Selector就会开始监听这个SocketChannel的读事件了。

SelectionKey.OP_CONNECT 事件处理流程

当SelectionKey.OP_CONNECT(连接事件)准备就绪时，我们执行如下操作：

            // We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
            // the NIO JDK channel implementation may throw a NotYetConnectedException.
            if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
                // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
                // See https://github.com/netty/netty/issues/924
                int ops = k.interestOps();
                ops &= ~SelectionKey.OP_CONNECT;
                k.interestOps(ops);

                unsafe.finishConnect();
            }

① 将SelectionKey.OP_CONNECT事件从SelectionKey所感兴趣的事件中移除，这样Selector就不会再去监听该连接的SelectionKey.OP_CONNECT事件了。而SelectionKey.OP_CONNECT连接事件是只需要处理一次的事件，一旦连接建立完成，就可以进行读、写操作了。

a) boolean wasActive = isActive()：

    public boolean isActive() {
        SocketChannel ch = javaChannel();
        return ch.isOpen() && ch.isConnected();
    }

因为此时「SocketChannel.finishConnect()」还没调用，所以「ch.isConnected()」将返回false，因此isActive()的结果为false。
b) doFinishConnect()：

    protected void doFinishConnect() throws Exception {
        if (!javaChannel().finishConnect()) {
            throw new Error();
        }
    }

该方法会调用SocketChannel.finishConnect()来标识连接的完成，如果我们不调用该方法，就去调用read/write方法，则会抛出一个NotYetConnectedException异常。
注意，无论如何在connect后finishConnect()方法都是需要被调用的。调用finishConnect()的三种返回：

1. 如果你在connect()后直接调用了finishConnect()( 并非在CONNECT事件中调用 )，则若finishConnect()返回了true，则表示channel连接已经建立，而且CONNECT事件也不会被触发了。
2. 如果finishConnect()方法返回false，则表示连接还未建立好。那么就可以通过CONNECT事件来监听连接的完成。

3. 如果finishConnect()方法抛出了一个IOException异常，则表示连接操作失败。
   

   c) fulfillConnectPromise(connectPromise, wasActive)：
   I. 当异步的“连接尝试”操作通过取消来关闭了，那么则直接返回。因为当“连接尝试”操作被取消时，connectPromise会被置为null。
   
   
   

   II. isActive()：获取当前SocketChannel的状态，因为此时「SocketChannel.finishConnect()」已经被调用过了，因此该方法会返回true。
   III. boolean promiseSet = promise.trySuccess()：
   将当前的异步的“连接尝试”操作尝试标记为成功。如果用户取消了“连接尝试(即，调用connect操作后，用户调用了cancel来取消该操作)”的操作的话，该方法将返回false；否则返回true，并且如果有ChannelFutureListener注册到了这个ChannelFuture(即，ChannelPromise)上，那么监听器的operationComplete方法将得以回调。
   IV. 无论当前的“连接尝试”操作是否被取消，channelActive()事件都将被触发，因为此时channel确实已经处于active状态了。
   channelActive是一个入站事件，该事件会从ChannelPipeline的head开始传播至tail间的所有ChannelInboundHandler，并回调它们的channelActive方法。
   

   我们来深入看看head对channelActive事件的处理：
   它除了将channelActive事件传播给下一个ChannelInboundHandler外，还会进行一个「readIfIsAutoRead()」操作：

        private void readIfIsAutoRead() {
            if (channel.config().isAutoRead()) {
                channel.read();
            }
        }

因为NioSocketChannel的NioSocketChannelConfig对象的autoRead属性默认就为1，因此isAutoRead()为true。那么就会调用channel.read()操作，这将触发一个read事件在ChannelPipeline中，而read是一个出站操作。它会从ChannelPipeline的尾部开始传播至head间的每个ChannelOutboundHandler。
我们接着来看head对read事件的处理：

        // DefaultChannelPipeline#HeadContext#read()
        public void read(ChannelHandlerContext ctx) {
            unsafe.beginRead();
        }

        // AbstractChannel#AbstractUnsafe#beginRead()
        public final void beginRead() {
            assertEventLoop();

            if (!isActive()) {
                return;
            }

            try {
                doBeginRead();
            } catch (final Exception e) {
                invokeLater(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.fireExceptionCaught(e);
                    }
                });
                close(voidPromise());
            }
        }

        // AbstractNioChannel#doBeginRead()
        protected void doBeginRead() throws Exception {
            // Channel.read() or ChannelHandlerContext.read() was called
            final SelectionKey selectionKey = this.selectionKey;
            if (!selectionKey.isValid()) {
                return;
            }

            readPending = true;

            final int interestOps = selectionKey.interestOps();
            if ((interestOps & readInterestOp) == 0) {
                selectionKey.interestOps(interestOps | readInterestOp);
            }
        }

这里的readInterestOp是SelectionKey.OP_READ，是在构建NioSocketChannel对象时传进来的。因此，我们可以知道，此时的read事件主要是完成了，在Selector中对已经注册到其上的NioSocketChannel的OP_READ标识为感兴趣的事件。这样Selector就监控该SocketChannel的读操作了。
V. 如果用户执行了取消“连接尝试”的操作，那么就关闭channel，并触发channelInactive事件。
d) 如果成员变量connectTimeoutFuture非空，则说明该“连接尝试”操作设置了一个连接超时时间。那么，此时连接已经完成了，我们就可以取消这个连接超时检测的定时任务了。超时任务会记录本次“连接尝试”操作为失败状态，并且会将connectTimeoutFuture成员变量置为null。
比如，可能存在这样一种情况：也就是当程序执行完fulfillConnectPromise方法中的「promise.trySuccess()」之后，以及在执行finally代码块之前，“连接尝试”的已经完成，并且ChannelPromise已标记为了true。但是此时设置的连接超时时间已到并且连接超时任务被得以执行，此时超时任务发现ChannelPromise的状态已经被标识过了也就不会进行关闭channel的操作。
因此如上流程我们知道，『connectTimeoutFuture == null』有两种情况：1，如果没有设置连接超时；2，设置了连接超时，并且只因为超时或其他原因已经执行了 NioSocketChannel 的 close() 操作。『connectTimeoutFuture#operationComplete』也有两种情况：1，在超时时间内连接还没建立，则执行相应的close操作；2，连接已经建立了，但是还未执行到connectTimeoutFuture#cancel(false)操作，connectTimeoutFuture 就触发了，此时因为 promise的状态已经被表示过了，也不会进行close操作。

总的来说，OP_CONNECT事件的触发时，表示当前的socket处于了可连接的状态了，需要调用SocketChannel.finishConnect()来完成连接的后续事件。同时会触发ChannelActive事件，该事件为一个入站事件，它会在NioSocketChannel所关联的ChannelPipeline管道得以传播，即，回调head到tail之间所有的ChannelInboundHandler的channelActive方法。而head的channelActive方法中又会触发一个channel的read操作，该操作最终会在NioSocketChannel所注册的Selector中标识OP_READ为感兴趣的事件，这使得Selector会监听NioSocketChannel上是否有可读的数据准备好被读取了。

SelectionKey.OP_READ 事件处理流程

当有可读数据准备被读取时，‘SelectionKey.OP_READ’将会触发。在NioEventLoop的事件循环中会对该事件进行处理：

if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
    unsafe.read();
}

我们来看看unsafe.read()的实现，在NioSocketChannel中unsafe是一个NioByteUnsafe实例：

PS：注意，如果在当前端主动调用「channel.shutdownInput()」方法时，需要在处理’ChannelInputShutdownReadComplete’这个用户自定义的事件时调用「channel().config().setAutoRead(false);」来将autoRead置为false。不然，OP_READ事件会一直被触发，而上的步骤’III’会一直被调用，这会导致一些问题，比如不必要的CPU消耗。
调用方式类似：

public class MyServerHandler extends SimpleChannelInboundHandler<String> {
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        Channel serverChannel = ctx.channel();
        if(serverChannel instanceof NioSocketChannel) {
            System.out.println("server shutdownInput...");
            ((NioSocketChannel) serverChannel).shutdownInput();
        }
    }

    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if(evt instanceof ChannelInputShutdownReadComplete) {
            System.out.println("detail ChannelInputShutdownEvent event......");
            ctx.channel().config().setAutoRead(false);
        }
        super.userEventTriggered(ctx, evt);
    }
}

SelectionKey.OP_WRITE 事件处理流程

在NioEventLoop的事件循环中’SelectionKey.OP_WRITE’事件的处理流程如下：

    // Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
     if ((readyOps & SelectionKey.OP_WRITE) != 0) {
        // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
        ch.unsafe().forceFlush();
    }

关于OP_WRITE事件：
OP_WRITE事件的就绪条件并不是发生在调用channel的write方法之后，而是在当底层缓冲区有空闲空间的情况下。因为写缓冲区在绝大部分时候都是有空闲空间的，所以如果你注册了写事件，这会使得写事件一直处于就就绪，选择处理线程就会一直占用着CPU资源。所以，只有当你确实有数据要写时再注册写操作，并在写完以后马上取消注册。
SocketChannel会在写数据时，若发现当buffer还有数据，但写缓冲区已经满的情况下，socketChannel.write(buffer)会返回已经写出去的字节数，此时为0。那么这个时候我们就需要注册OP_WRITE事件，这样当写缓冲区又有空闲空间的时候就会触发OP_WRITE事件，这样我们就可以继续将没写完的数据继续写出了。而且在写完后，一定要记得将OP_WRITE事件注销。
比如，来看看NioSocketChannel的doWrite()操作(「 ch.unsafe().forceFlush()」方法最终也就是会调用到这里)：

后记

本文主要对NioEventLoop中涉及到的四种NIO事件的处理流程进行了分析。四个看似简单的处理流程，深入探索后发现其实并不简单，其实可以展开的点还有很多，特别是关于写事件涉及到ChannelOutboundBuffer以及Netty默认使用的PooledByteBufAllocator实现了jemalloc算法来完成高效的内存分配等等，希望在后面的文章中能继续和大家分享我的分析以及想法。
若文章有任何错误，望大家不吝指教:)