摘自《Netty 即时聊天实战与底层原理》

本章，我们来分析每个新连接在接入过程中，Netty 底层的机制是如何实现的。先来简要回顾一下：

首先是 Netty 中的 Reactor 线程模型。

Netty 中最核心的东西莫过于两种类型的 Reactor 线程。这两种类型的 Reactor 线程可以看作 Netty 中的两组发动机，驱动着 Netty 整个框架的运转。

一种是 boss 线程，专门用来接收新请求，然后封装成 Channel 对象传递给 worker 线程；还有一种类型是 worker 线程，专门用来处理连接上数据的读写。

不管是 boss 线程还是 worker 线程，所做的事情均分为以下 3 个步骤：

1. 轮询注册在 Selector 上的 IO 事件；
2. 处理 IO 事件；
3. 执行异步 Task。

对于 boss 线程来说，第一步轮询出来的基本都是 ACCEPT 事件，表示有新的连接；而 worker 线程轮询出来的基本都是 read 或 write 事件，表示网络的读写事件。

其次是服务端启动流程。

服务端是在用户线程中开启的，通过 bind 方法，在第一次添加异步任务的时候启动 boss 线程（[08]#6）。启动之后，当前服务器就可以开始监听了。

1. 新连接接入的总体流程

简单来说，新连接的接入流程可以分为 3 个过程：

1. 检测到有新连接；
2. 将新连接注册到 worker 线程；
3. 注册新连接的读事件。

2. 检测到有新连接

我们已经知道，当调用 bind 方法启动服务端之后，服务端的 Channel —— NioServerSocketChannel 已经注册到 boss Reactor 线程，Reactor 线程不断检测是否有新的连接，直到检测出有 ACCEPT 事件发生。

NioEventLoop

private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
  final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
  if (!k.isValid()) {
      final EventLoop eventLoop;
      eventLoop = ch.eventLoop();

      // Only close ch if ch is still registered to this EventLoop. ch could have deregistered
      // from the event loop and thus the SelectionKey could be cancelled as part of the
      // deregistration process, but the channel is still healthy and should not be closed.
      if (eventLoop != this || eventLoop == null) { return; }

      // close the channel if the key is not valid anymore
      unsafe.close(unsafe.voidPromise());
      return;
  }

  int readyOps = k.readyOps();
  // 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
      int ops = k.interestOps();
      ops &= ~SelectionKey.OP_CONNECT;
      k.interestOps(ops);

      unsafe.finishConnect();
  }

  // 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();
  }

  // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
  // to a spin loop
  if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
      unsafe.read();
  }
}

上面这段代码是 Reactor 线程在第二个过程做的事情，最后一个 if 表示 boss Reactor 线程已经轮循到 SelectionKey.OP_ACCEPT 事件，即表明有新连接进入，此时将调用 Channel 的 Unsafe 来进行实际的操作。

在服务端启动流程解析章节中，我们已经知道，服务端对应的 Channel 的 Unsafe 是 NioMessageUnsafe，我们进入它的 read 方法，进入新连接处理的第二步。

3. 注册 Reactor 线程

NioMessageUnsafe

private final List<Object> readBuf = new ArrayList<Object>();

@Override
public void read() {
    assert eventLoop().inEventLoop();
    final ChannelConfig config = config();
    final ChannelPipeline pipeline = pipeline();
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    allocHandle.reset(config);

    boolean closed = false;
    Throwable exception = null;

    do {
        // 1. 创建 NioSocketChannel
        int localRead = doReadMessages(readBuf);
        if (localRead == 0) {
            break;
        }
        if (localRead < 0) {
            closed = true;
            break;
        }
        allocHandle.incMessagesRead(localRead);
    } while (allocHandle.continueReading());

    // 2. 设置并绑定 NioSocketChannel
    int size = readBuf.size();
    for (int i = 0; i < size; i ++) {
        readPending = false;
        pipeline.fireChannelRead(readBuf.get(i));
    }
    readBuf.clear();
    allocHandle.readComplete();
    pipeline.fireChannelReadComplete();

    // ...
}

笔者省去了非关键部分的代码，可以看到，一上来，就用一条断言确定该 read 方法必须来自 Reactor 线程调用，然后获得 Channel 对应的 Pipeline 和 RecvByteBufAllocator.Handle（先暂时不展开）。

接下来，调用 doReadMessages() 不断地读取消息，用 readBuf 作为容器。其实读者可以猜到这里读取的是一个个连接，然后使用 for 循环调用 pipeline.fireChannelRead()，将每个新连接都经过一层服务端 Channel 的 Pipeline 逻辑处理，之后清理容器，触发 pipeline.fireChannelReadComplete()。整个过程还是比较清晰的，下面我们具体分析这两个方法。

1. 创建 NioSocketChannel：doReadMessages(List)；
2. 设置并绑定 NioSocketChannel：pipeline.fireChannelRead(NioSocketChannel)。

3.1 创建 NioSocketChannel

NioMessageUnsafe

private final List<Object> readBuf = new ArrayList<Object>();

@Override
public void read() {
    assert eventLoop().inEventLoop();
    final ChannelConfig config = config();
    final ChannelPipeline pipeline = pipeline();
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    allocHandle.reset(config);

    boolean closed = false;
    Throwable exception = null;

    do {
        // 1. 创建 NioSocketChannel
        int localRead = doReadMessages(readBuf);
        if (localRead == 0) {
            break;
        }
        if (localRead < 0) {
            closed = true;
            break;
        }
        allocHandle.incMessagesRead(localRead);
    } while (allocHandle.continueReading());

    // 2. 设置并绑定 NioSocketChannel
    // ...
}

doReadMessages 的方法体在 NioServerSocketChannel 类中，下面进入这个方法体来分析（代码稍作精简）。

NioServerSocketChannel

@Override
protected int doReadMessages(List<Object> buf) throws Exception {
    // 1. 创建 JDK 领域的 Channel
    SocketChannel ch = SocketUtils.accept(javaChannel());
    // 2. 封装为 Netty 领域的 Channel
    if (ch != null) {
        buf.add(new NioSocketChannel(this, ch));
        return 1;
    }
    return 0;
}

在这里，我们终于窥探到 Netty 调用 JDK NIO 的边界：SocketUtils.accept(javaChannel())。由于 Netty 中 Reactor 线程第一步就扫描有 ACCEPT 事件发生，因此，这里的 accept 方法是立即返回的，返回 JDK 底层 NIO 创建的一条 JDK 层面的 Channel。

接下来，Netty 将 JDK 的 SocketChannel 封装成自定义的 NioSocketChannel，加入 List，这样外层就可以遍历该 List，做后续处理。

我们已经知道，服务端启动过程中会创建一个 NioServerSocketChannel，而创建 NioServerSocketChannel 的过程中又会创建 Netty 的一系列核心组件，包括 Pipeline、Unsafe 等，那么，创建 NioSocketChanel 的时候是否也会创建这一系列组件呢？

NioSocketChannel

public NioSocketChannel(Channel parent, SocketChannel socket) {
    super(parent, socket);
    config = new NioSocketChannelConfig(this, socket.socket());
}

AbstractNioByteChannel

protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
    super(parent, ch, SelectionKey.OP_READ);
}

这里，我们看到 JDK NIO 里熟悉的影子 SelectionKey.OP_READ，一般在原生的 JDK NIO 编程中，我们也会注册这样一个事件，表示对 Channel 的读事件感兴趣。

继续向上追踪，追踪到 AbstractNioByteChannel 的父类 AbstractNioChannel。这里，相信大家应该了解了 NioServerSocketChannel 最终的父类也是 AbstractNioChannel。所以，创建 NioSocketChannel 的模板和创建 NioServerSocketChannel 保持一致。

protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
    super(parent);
    this.ch = ch;
    this.readInterestOp = readInterestOp;
    ch.configureBlocking(false);
}

这里的 readInterestOp 表示该 Channel 关心的事件是 SelectionKey.OP_READ，后续会将该事件注册到 Selector，之后设置该通道为非阻塞模式。

AbstractNioChannel 构造方法的第一行代码调用 super(parent)，便是在 AbstractChannel 构造方法中创建一系列与该 Channel 绑定的组件。

protected AbstractChannel(Channel parent) {
    this.parent = parent;
    id = newId();
    unsafe = newUnsafe();
    pipeline = newChannelPipeline();
}

分析到这里，是时候了解一下 Netty 中最常用的 Channel 的结构了，如下图所示。

这里的继承关系有所简化，当前，我们只需要了解这么多。

1. Channel 继承 Comparable 表示 Channel 是一个可以比较的对象；
2. Channel 继承 AttributeMap 表示 Channel 是可以绑定属性的对象，在用户代码中，我们经常使用 channel.attr(...) 来给 Channel 绑定属性，其实就是把属性设置到 AttributeMap 中；
3. ChannelOutboundInvoker 是 4.1.x 版本新加的抽象，表示用户代码可以在 Channel 上进行哪些操作；
4. DefaultAttributeMap 为 AttributeMap 的默认实现，后面的 Channel 继承了它，可以直接使用；
5. AbstractChannel 用于实现 Channel 的大部分方法，其中我们最熟悉的就是在其构造方法中，创建一些 Channel 的基本组件，这里的 Channel 通常包括 SocketChannel 和 ServerSocketChannel；
6. AbstractNioChannel 基于 AbstractChannel 做了 NIO 相关的一些操作，保存 JDK 底层的 SelectableChannel 的引用，并且在构造方法中设置 Channel 为非阻塞（设置非阻塞这一点对于 NIO 编程是必不可少的）；
7. 最后，就是两大 Channel —— NioSocketChannel 和 NioServerSocketChannel，分别对应则会服务端接收新连接过程和新连接读写过程。

我们继续之前的源码分析，在创建一条 NioSocketChannel 并放置在 List 容器里后，就开始 for 循环进行下一步操作。

3.2 设置并绑定 NioSocketChannel

创建完 NioSocketChannel 之后，接下来要对 NioSocketChannel 做一些设置，并且需要将它绑定到一个执行的 Reactor 线程中。

NioMessageUnsafe

private final List<Object> readBuf = new ArrayList<Object>();

@Override
public void read() {
    assert eventLoop().inEventLoop();
    final ChannelConfig config = config();
    final ChannelPipeline pipeline = pipeline();
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    allocHandle.reset(config);

    // 1. 创建 NioSocketChannel
    // ...

    // 2. 设置并绑定 NioSocketChannel
    int size = readBuf.size();
    for (int i = 0; i < size; i ++) {
        readPending = false;
        pipeline.fireChannelRead(readBuf.get(i));
    }
    readBuf.clear();
    allocHandle.readComplete();
    pipeline.fireChannelReadComplete();

    // ...
}

readBuf 中承载着所有新建的连接，如果某个时刻，Netty 轮询到多个连接，那么使用 for 循环就可以批量处理这些连接，即 NioSocketChannel。

处理每一个 NioSocketChannel 是通过调用 NioServerSocketChannel 的 pipeline.fireChannelRead(...) 来执行的，在后面章节正式介绍 Pipeline 之前，先简单介绍一下 Pipeline 组件。

在 Netty 的各种类型的 Channel 中，都会包含一个 Pipeline。Pipeline 的字面意思是“管道”，我们可以理解为一条流水线。流水线有起点、有结束，中间还有各种各样的流水线关卡。一件物品，在流水线起点开始处理，经过各种流水线关卡的加工，最终到流水线结束。

对应到 Netty 里，流水线的开始是 HeadContext，流水线的结束是 TailContext。HeadContext 中调用 Unsafe 做具体的操作，TailContext 中用于向用户抛出 Pipeline 中未处理异常以及对未处理消息的警告。我们暂时先了解这么多，关于 Pipeline 的具体分析，后面再说。

在服务端的启动过程中，Netty 给服务端 Channel 自动添加了一个 Pipeline 处理器 ServerBootstrapAcceptor，并已经将用户代码中设置的一系列参数传入了 ServerBootstrapAcceptor 构造方法。接下来，我们来分析 ServerBootstrapAcceptor。

private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter {

    private final EventLoopGroup childGroup;
    private final ChannelHandler childHandler;
    private final Entry<ChannelOption<?>, Object>[] childOptions;
    private final Entry<AttributeKey<?>, Object>[] childAttrs;
    private final Runnable enableAutoReadTask;

    ServerBootstrapAcceptor(final Channel channel, EventLoopGroup childGroup,
            ChannelHandler childHandler, Entry<ChannelOption<?>, Object>[] childOptions,
            Entry<AttributeKey<?>, Object>[] childAttrs) {
        this.childGroup = childGroup;
        this.childHandler = childHandler;
        this.childOptions = childOptions;
        this.childAttrs = childAttrs;

        // Task which is scheduled to re-enable auto-read.
        // It's important to create this Runnable before we try to submit
        // it as otherwise the URLClassLoader may not be able to
        // load the class because of the file limit it already reached.
        enableAutoReadTask = new Runnable() {
            @Override
            public void run() {
                channel.config().setAutoRead(true);
            }
        };
    }

    /**
     * 在新连接接入时被调用
     */
    @Override
    @SuppressWarnings("unchecked")
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        final Channel child = (Channel) msg;

        // 1. 给新连接的 Channel 添加用户自定义的 Handler 处理器，这其实是一个
        // 特殊的 ChannelHandler: ChannelInitializer (联系下个代码块)
        child.pipeline().addLast(childHandler);

        // 2. 设置 ChannelOption，主要和 TCP 连接一些底层参数及 Netty 自身对一个连接的参数有关
        setChannelOptions(child, childOptions, logger);

        // 3. 设置新连接 Channel 的属性
        for (Entry<AttributeKey<?>, Object> e: childAttrs) {
            child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
        }

        // 4. 绑定 Reactor 线程
        childGroup.register(child).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    forceClose(child, future.cause());
                }
            }
        });
    }

    // ...

}

a. 添加用户自定义 Handler

pipeline.fireChannelRead(NioSocketChannel) 最终调用这里的 ServerBootstrapAcceptor 的 channelRead 方法，而 channelRead() 一上来就把这里的 msg 强制转换成 Channel，为什么这里可以强制转换？读者可以思考一下。

拿到该 Channel，也就是拿到了该 Channel 对应的 Pipeline，这个 Pipeline 其实就是在 #3.1 中调用 AbstractChannel 的构造方法时创建的。然后，将用户代码中的 childHandler，添加到 Pipeline 中。

serverBootstrap
        .group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new NettyServerHandler());
            }
        })
        // ...

childHandler 对应的就是上述用户代码中的 ChannelInitializer。到了这里，NioSocketChannel 中 Pipeline 对应的 Handler 为 head → ChannelInitializer → tail。

b. 设置 ChannelOption

这里的 ChannelOption 也在用户代码中设置，最终传递到 ServerBootstrapAcceptor。ChannelOption 主要是和 TCP 底层参数相关的一些配置以及 Netty 对一条连接的配置。

c. 设置 ChannelAttr

设置 NioSocketChannel 对应的 ChannelAttr，ChannelAttr 和 ChannelOption 一样，也在用户代码中设置，最终传递到 ServerBootstrapAcceptor，一般情况下用不着 ChannelAttr。

d. 绑定 Reactor 线程

对于 childGroup.register(child)，这里的 childGroup 就是我们在用户代码里创建的 worker NioEventLoopGroup，我们进入 NioEventLoopGroup 的 register 方法，register 首先调用 next() 方法获取一个 EventLoop 对象 MultithreadEventLoopGroup。

@Override
public ChannelFuture register(Channel channel) {
    return next().register(channel);
}
@Override
public EventLoop next() {
    return (EventLoop) super.next();
}

调用其父类 MultithreadEventExecutorGroup。

@Override
public EventExecutor next() {
    return chooser.next();
}

我们发现，MultithreadEventExecutorGroup 中的 next() 方法调用了 chooser 对象的 next() 方法，而这个对象正是我们在 [09] 分析的 EventExecutorChooser，它的作用是从 NioEventLoopGroup 中，选择一个 NioEventLoop，所以，最终 childGroup.register(child) 会调用 NioEventLoop 的 register 方法，由其父类 SingleThreadEventLoop 来实现。

@Override
public ChannelFuture register(Channel channel) {
    return register(new DefaultChannelPromise(channel, this));
}

到这里，读者应该会比较眼熟了，这里和服务端启动流程中的注册 Channel 的模板一样，都由 AbstractUnsafe 来执行。下面我们来分析，这一套逻辑应该如何执行。最终，register() 方法会调用如下方法。

AbstractUnsafe

private void register0(ChannelPromise promise) {

    // 1. 注册 Selector
    doRegister();
    neverRegistered = false;
    registered = true;

    // 2. 配置自定义 Handler
    pipeline.invokeHandlerAddedIfNeeded();

    safeSetSuccess(promise);

    // 3. 传播 ChannelRegistered 事件
    pipeline.fireChannelRegistered();

    // 4. 注册读事件
    if (isActive()) {
        if (firstRegistration) {
            pipeline.fireChannelActive();
        } else if (config().isAutoRead()) {
            beginRead();
        }
    }
}

（1）注册 Selector。 和服务端启动过程一样，先调用 doRegister() 进行真正的注册过程。

AbstractNioChannel

@Override
protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
            selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
            return;
        } catch (CancelledKeyException e) {
            if (!selected) {
                // Force the Selector to select now as the "canceled" SelectionKey may still be
                // cached and not removed because no Select.select(..) operation was called yet.
                eventLoop().selectNow();
                selected = true;
            } else {
                // We forced a select operation on the selector before but the
                // SelectionKey is still cached for whatever reason. JDK bug ?
                throw e;
            }
        }
    }
}

javaChannel().register(...) 将 NioSocketChannel 所有的事件都由绑定的 Reactor 线程的 Selector 来轮询。

（2）配置自定义 Handler。 到目前为止，NioSocketChannel 的 Pipeline 中有 3 个 Handler：head → ChannelInitializer → tail。接下来，invokeHandlerAddedIfNeeded 最终会调用 ChannelInitializer 的 handlerAdded(...) 方法。

@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    if (ctx.channel().isRegistered()) {
        // This should always be true with our current DefaultChannelPipeline implementation.
        // The good thing about calling initChannel(...) in handlerAdded(...) is that
        // there will be no ordering surprises if a ChannelInitializer will add another
        // ChannelInitializer. This is as all handlers will be added in the expected order.
        initChannel(ctx);
    }
}

private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
     // Guard against re-entrance.
    if (initMap.putIfAbsent(ctx, Boolean.TRUE) == null) {
        try {
            initChannel((C) ctx.channel());
        } catch (Throwable cause) {
            // Explicitly call exceptionCaught(...) as
            // we removed the handler before calling initChannel(...).
            // We do so to prevent multiple calls to initChannel(...).
            exceptionCaught(ctx, cause);
        } finally {
            // 将自身删除
            remove(ctx);
        }
        return true;
    }
    return false;
}

这里的 initChannel() 方法又是什么呢？让我们回到服务端启动代码，比如下面这段用户代码。

serverBootStrap
    .group(bossGroup, workerGroup)
    .channel(NioServerSocketChannel.class)
    .handler(new LoggingHandler(LogLevel.INFO))
    .childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline()
                .addLast(new IdleStateHandler(...))
                .addLast(new MyServerHandler());
        }
    });

对照前面的分析，原来，最终 initChannel(...) 会调用用户代码，而一般在用户代码里，我们会添加自定义的一系列 Handler。所以，这个过程其实就是给 NioSocketChannel 配置自定义 Handler，NioSocketChannel 中的 Handler 包括 head → IdleStateHandler → MyServerHandler → tail。

（3）传播 ChannelRegistered 事件。 pipeline.fireChannelRegistered() 其实没有干特别的事情，最终只是把连接注册时间往下传播，调用了每一个 Handler 的 channelRegistered 方法。

（4）注册读事件。 现在，我们还剩下这些代码没有分析。

AbstractUnsafe

private void register0(ChannelPromise promise) {

    // ...

    // 4. 注册读事件
    if (isActive()) {
        if (firstRegistration) {
            pipeline.fireChannelActive();
        } else if (config().isAutoRead()) {
            beginRead();
        }
    }
}

isActive() 在连接已经建立的情况下返回 true，所以进入方法块，即进入 pipeline.fireChannelActive()。接下来的调用过程和服务端启动流程的分析过程一样，最终都会调用如下代码。

AbstractNioChannel

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

读者应该还记得前面分析 register0 方法的时候，向 Selector 注册的事件代码是 0，而 readInterestOp 对应的事件代码是 SelectionKey.OP_READ。参考前文中创建 NioSocketChannel 的过程，稍加推理，就会知道，这里其实就是将 SelectionKey.OP_READ 事件注册到 Selector，表示这条管道已经可以开始处理读事件。至此，新连接接入的流程就算结束了。

3.3 小结

当 boss Reactor 线程在检测到有 ACCEPT 事件之后，创建 JDK 底层的 Channel，然后使用一个 NioSocketChannel 包装 JDK 底层的 Channel，把用户设置的 ChannelOpotion、ChannelAttr、ChannelHandler 都设置到 NioSocketChannel 中。

接着，从 worker Reactor 线程组，也就是 worker NioEventLoopGroup 选择一个 NioEventLoop，把 NioSocketChannel 包装的 JDK 的 Channel 当作 key，自身当作 attachement，注册到 NioEventLoop 对应的 Selector。这样，后续有读写事件发生时，就可以直接获得 attachement，也就是 NioSocketChannel，来处理读写数据逻辑。

最后，对本章再做个总结。

1. boss Reactor 线程轮询到有新连接接入；
2. 通过封装 JDK 底层的 Channel 创建 NioSocketChannel 及一系列 Netty 核心组件；
3. 通过 chooser 选择一个 worker Reactor 线程将该连接绑定上去；
4. 注册读事件，开始新连接的读写。

标签：10,pipeline,ch,NioSocketChannel,线程,解析,final,Channel,客户端
来源： https://www.cnblogs.com/liujiaqi1101/p/16392044.html