[toc]

Netty 线程模型分析

阅读本文之前，需要了解 NIO 相关知识，可参阅我的文章：NIO 与 epoll 知识详解

Reactor 线程模型

Netty 的线程模型 是基于 Reactor 线程模型的，Reactor 线程模型分为三种模型：

  • 单线程模型：注册所有感兴趣的事件，一个线程管理多个 Channel，单线程轮询 IO 事件是否发生，若事件发生，在同一个线程中调用处理程序，属于 一对一 的模型。

Netty 源码 — 线程模型的分析 - 图1

  • 多线程模型：多线程模型中，一个线程管理多个 Channel，仍然是单线程轮询 IO 事件是否发生，但将事件发生的处理程序交由线程池运行，属于 一对多 的模型。

Netty 源码 — 线程模型的分析 - 图2

  • 主从线程模型：在这种模型中，有多个线程负责 IO 事件的连接，每个负责 IO 事件连接的线程又可以指派多个处理线程，属于 多对多 的模型。
    这种模式通常有两种实现，严格意义上的实现应该是一个 Channel 可由多个线程管理，一个线程又可管理多个 Channel，这种实现需要在线程间共享 Channel，Netty 并没有采用这种方式。另一种不太规范的实现是一个线程管理多个 Channel，但创建多个线程，形象的说，就像多个多线程模型一起启动一样。

Netty 源码 — 线程模型的分析 - 图3

Netty 中可根据所调参数选择对应的模型，Netty 中有一个专门的线程组 BossGroup 用于接收发送 IO 事件，还有一个线程组 WorkerGroup 用以调用 handler。

Netty 源码 — 线程模型的分析 - 图4

在 Netty 中，即使设置 bossGroup 为多个线程，一个 Channel 仍然只属于一个线程管理，并不会由多个线程管理，因此如果服务端只有一个 ServerChannel，设置多线程 bossGroup 是没有意义的，除非有多个 Channel。

Netty 初始化线程池

Netty 中的模型就是可创建 BossGroup 和 WorkerGroup 两个线程组，你可以指定对应的线程数。

Netty 源码 — 线程模型的分析 - 图5

在 Netty 中，每一个 EventLoop 其实就是一个线程，Netty 中的 IO 事件就是一个任务，每个任务都会被提交到 EventLoop 中，由 EventLoop 在事件循环中执行。

在 Netty 中，创建一个 EventLoopGroup 时就会自动创建多个 EventLoop 线程，数目由 nThreads 指定。

  1. NioEventLoopGroup group = new NioEventLoopGroup(nThreads);

如果没有指定线程数，在 MultithreadEventLoopGroup 的静态方法中初始化默认的线程数为 CPU 核心数的两倍：

  1. static {
  2.     DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
  3.         "io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
  4. }

Netty 使用 EventExecutor 数组来保存 EventLoop，EventLoop 继承于 EventExecutor，在 new NioEventLoopGroup 中，通过不断向上传递，在父类 MultithreadEventExecutorGroup 的构造方法中，EventExecutor 数组被创建：

  1. private final EventExecutor[] children;
  2. private final Set<EventExecutor> readonlyChildren;
  4. protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
  5.                                             EventExecutorChooserFactory chooserFactory, Object... args) {
  6.     children = new EventExecutor[nThreads];
  7.     for (int i = 0; i < nThreads; i ++) {
  8.         // 创建 NioEventLoop，此方法其实就是 new NioEventLoop
  9.         children[i] = newChild(executor, args);
  10.     }
  12.     for (EventExecutor e: children) {
  13.         // 添加监听器
  14.         e.terminationFuture().addListener(terminationListener);
  15.     }
  16.     // 去重
  17.     Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
  18.     Collections.addAll(childrenSet, children);
  19.     readonlyChildren = Collections.unmodifiableSet(childrenSet);
  20. }

所以，在创建 Group 的时候主要就是初始化了 EventLoop 线程组。

BossGroup 工作原理

Netty 源码 — 线程模型的分析 - 图6

绑定 Channel

当调用了 bootstrap.bind() 方法时，开始正式进入核心环节，我们主要关注 Channel 是如何被绑定到 EventLoop 中的，调用此方法时，会经过如下调用链：

  1. AbstractBootstrap.bind()
  2.     -> doBind()
  3.         -> initAndRegister()
  4.             -> config().group().register(channel)
  5.                 -> MultithreadEventLoopGroup.register(channel)
  6.                     -> SingleThreadEventLoop.register(channel)
  7.                         -> AbstractChannel.AbstractUnsafe.register()

从 MultithreadEventLoopGroup.register(channel) 方法看起：

  1. @Override
  2. public ChannelFuture register(Channel channel) {
  3.     return next().register(channel);
  4. }

这个方法调用了 EventExecutorChooserFactory#next() 方法，next() 方法其实就是获取一个 EventLoop，跟进 next() 方法，发现 Netty 中只提供了轮询的方式获取 EventLoop：

  1. @Override
  2. public EventExecutor next() {
  3.     return executors[Math.abs(idx.getAndIncrement() % executors.length)];
  4. }

还记得 executors 数组吗？这里 executors 数组其实就是 Group 初始化的 EventLoop 数组，因此通过调用 next() 方法就轮询获得了一个 EventLoop，现在开始将 Channel 绑定到 EventLoop 中。

将目光移到调用链中的 AbstractUnsafe.register:

  1. public final void register(EventLoop eventLoop, final ChannelPromise promise) {
  2.     // 省略, promise 是 channel 的包装, eventLoop 是调用 next() 得到的线程
  3.     AbstractChannel.this.eventLoop = eventLoop;
  4.     // 如果给定的线程已经被启动运行了，那么直接注册
  5.     if (eventLoop.inEventLoop()) {
  6.         register0(promise);
  7.     } else {
  8.         // 否则，要先启动一下，再注册，这里其实是向 Netty 提交了一个任务
  9.         eventLoop.execute(new Runnable() {
  10.             @Override
  11.             public void run() {
  12.                 register0(promise);
  13.             }
  14.         });
  15.     }
  16. }

execute 方法是事件循环的开始方法，但这里我们先关注 register0(promise) 方法，register0 方法内调用了 AbstractNioChannel#doRegister() 方法，这个方法其实是 Netty NIO 与 JDK NIO 交互的地方：

  1. protected void doRegister() throws Exception {
  2.     boolean selected = false;
  3.     for (;;) {
  4.         try {
  5.             selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
  6.             return;
  7.         } catch (CancelledKeyException e) {
  8.             if (!selected) {
  9.                 eventLoop().selectNow();
  10.                 selected = true;
  11.             } else {
  12.                 throw e;
  13.             }
  14.         }
  15.     }
  16. }

javaChannel() 方法返回了已被绑定的、 jdk 原生的 Channel，此 Channel 由传递的地址注册，eventLoop().unwrappedSelector() 是对应线程的选择器，也是 jdk 原生的 Selector，doRegister() 方法还向通道附加了 this(AbstractNioChannel) 对象，在这里 Channel 正式向选择器注册了，而 selectionKey 被 AbstractNioChannel 所保存，用于后续 select。

事件循环

现在来看看 execute(SingleThreadEventExecutor类下) 方法，这个方法将任务(doRegister方法)添加到任务队列后，然后调用了 startThread 方法，startThread 方法会判断线程是否已被启动：

  1. private void startThread() {
  2.     if (state == ST_NOT_STARTED) {
  3.         if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
  4.             try {
  5.                 doStartThread();
  6.             } catch (Throwable cause) {
  7.                 STATE_UPDATER.set(this, ST_NOT_STARTED);
  8.                 PlatformDependent.throwException(cause);
  9.             }
  10.         }
  11.     }
  12. }

如果未启动则调用 dotartThread 方法：

  1. private void doStartThread() {
  2.     executor.execute(new Runnable() {
  3.         public void run() {
  4.             try {
  5.                 SingleThreadEventExecutor.this.run();
  6.             } catch (Throwable t) {
  7.                 logger.warn("Unexpected exception from an event executor: ", t);
  8.             }
  9.         }
  10.     }
  11. }

此方法提交任务到 Netty，执行 SingleThreadEventExecutor.this.run() 方法，这会新建线程执行：

  1. public void execute(Runnable command) {
  2.     threadFactory.newThread(command).start();
  3. }

也就是说到了这里才新建了线程，来看看 run 方法：

  1. protected void run() {
  2.     for (;;) {
  3.         switch(selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
  4.             case SelectStrategy.CONTINUE:
  5.                 continue;
  6.             case SelectStrategy.SELECT:
  7.                 // 堵塞等待 IO 事件到来，在这里面有解决 NIO 空轮询的 BUG
  8.                 select(false);
  9.             default:
  10.         }
  11.         // IO 事件与 时间调度任务执行的时间比例，默认是 50%
  12.         final int ioRatio = this.ioRatio;
  13.         final long ioStartTime = System.nanoTime();
  14.         try {
  15.             // 处理 IO 事件
  16.             processSelectedKeys();
  17.         } finally {
  18.             final long ioTime = System.nanoTime() - ioStartTime;
  19.             // 处理任务，根据 ioRatio 与 IO 事件执行的时间计算出执行任务的最大允许时间
  20.             // 这不是很准确的，因为无论如何 Netty 必须会等待一个任务完成
  21.             // Netty 会等待一个任务完成再计算时间，如果超时，在下个任务开始前返回
  22.             runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
  23.         }
  24.     }
  25. }

在 select(wakenUp.getAndSet(false)) 方法中，有效解决了 Java NIO 空轮询的 BUG，出现这个 BUG 的原因是 某个连接出现异常，操作系统返回异常类型，因此会唤醒阻塞在selector.select上的线程，但由于 Java NIO 事件设计中并没有此异常事件，也没有对应的解决方案，因此被唤醒的线程将不停的运行，因为 select() 总是会返回(问题没有解决，在 Java 中会返回 0)，以至于占满 CPU。

在部分Linux的2.6的kernel中，poll和epoll对于突然中断的连接socket会对返回的eventSet事件集合置为POLLHUP，也可能是POLLERR，eventSet事件集合发生了变化，这就可能导致Selector会被唤醒。

在 Netty 的解决方案中，Netty 会统计空轮询出现的次数，一旦达到阈值时，Netty 会重新新建一个选择器，将原先选择器上有效的连接迁移至新的选择器上，重新运行。

当 IO 事件到来时，Netty 会根据 ioRatio 配置执行 IO 事件 和 任务，执行任务的时间与执行 IO 事件时间所占比例不高于 ioRatio，这么做是为了防止任务事件长时间堵塞而导致 IO 事件无法执行。

要想提交一条任务，只需执行如下代码，但记住，千万不要执行长时间堵塞代码，这会导致 IO 事件的执行也被堵塞：

  1. EventLoop eventLoop = channel.eventLoop();
  2. eventLoop.execute(new Runnable() {
  3.     @Override
  4.     public void run() {
  5.         System.out.println("Hello, Netty!");
  6.     }
  7. });

执行 IO 事件

processSelectedKeys 方法其实调用了 processSelectedKeysOptimized 方法：

  1. private void processSelectedKeysOptimized() {
  2.     for (int i = 0; i < selectedKeys.size; ++i) {
  3.         final SelectionKey k = selectedKeys.keys[i];
  4.         // 为了 GC，将其回收
  5.         selectedKeys.keys[i] = null;
  6.         processSelectedKey(k, (AbstractNioChannel) a);
  7.     }
  8. }

在这个方法内，遍历 select() 得到的 key，Netty 中准备就绪的 key 集合被封装成一个数组对象，而在 jdk 中的实现是一个 HashSet，这么做的好处是为了提高性能，操作数组比操作哈希快得多，当然，这里 Netty 更多的考虑是考虑到在实践中 HashSet.add() 发送哈希冲突的概率并不小，一旦发生哈希冲突，add 将是 O(N) 级别的。

随后调用 processSelectedKey 方法：

  1. private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
  2.     final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
  3.     try {
  4.         int readyOps = k.readyOps();
  5.         if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
  6.             int ops = k.interestOps();
  7.             ops &= ~SelectionKey.OP_CONNECT;
  8.             k.interestOps(ops);
  10.             unsafe.finishConnect();
  11.         }
  13.         if ((readyOps & SelectionKey.OP_WRITE) != 0) {
  14.             ch.unsafe().forceFlush();
  15.         }
  16.         // 这里会触发连接事件
  17.         if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
  18.             unsafe.read();
  19.         }
  20.     } catch (CancelledKeyException ignored) {
  21.         unsafe.close(unsafe.voidPromise());
  22.     }
  23. }

这完全就是 jdk 的那一套嘛！根据 readyOps 判断事件的类型，进而调用不同的 unsafe 方法，注意这里 unsafe 是与对应通道绑定的，这在代码 final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe() 中有体现，所以通过 unsafe 是可以获得到对应通道的配置和数据的，

这里分析下 unsafe.read() 方法，：

  1. @Override
  2. public void read() {
  3.     assert eventLoop().inEventLoop();
  4.     final ChannelConfig config = config();
  5.     final ChannelPipeline pipeline = pipeline();
  6.     final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
  7.     Throwable exception = null;
  8.     try {
  9.         try {
  10.             do {
  11.                 // 将消息写入缓存
  12.                 int localRead = doReadMessages(readBuf);
  13.                 allocHandle.incMessagesRead(localRead);
  14.             } while (allocHandle.continueReading());
  15.         } catch (Throwable t) {
  16.             exception = t;
  17.         }
  19.         int size = readBuf.size();
  20.         for (int i = 0; i < size; i ++) {
  21.             readPending = false;
  22.             // 将每个消息交由 pipeline, Netty 会帮我们整合
  23.             // readBuf 是一个 Object 数组
  24.             pipeline.fireChannelRead(readBuf.get(i));
  25.         }
  26.         readBuf.clear();
  27.         allocHandle.readComplete();
  28.         pipeline.fireChannelReadComplete();
  29.     } finally {
  30.         if (!readPending && !config.isAutoRead()) {
  31.             removeReadOp();
  32.         }
  33.     }
  34. }
  35. }

简单来说，此方法就是将数据读入缓存中，默认是使用池化技术的 ByteBuf，然后调用 pipeline.fireChannelRead(byteBuf) 方法进行转发。

一个 pipeline 对应一个 Channel，pipeline 其实就是一个双端链表，链表的节点由 Handler 组成，入站消息由 head 向 tail 依次遍历 Handler，出战事件由 tail 向 head 依次遍历 Handler。

来看看 readBuf.get(i) 到底是什么：

Netty 源码 — 线程模型的分析 - 图7

消息竟然是一个 NioSocketChannel，此 Channel 已经是对应 IO的通道了，将用于 WorkerGroup 工作！

NioSocketChannel 到底是如何产生的呢？一切都在 int localRead = doReadMessages(readBuf); 这一行代码中：

  1. protected int doReadMessages(List<Object> buf) throws Exception {
  2.     // 调用 javaChannel() 获取包装后的服务器 channel，然后调用 accept 获取与客户端的连接
  3.     SocketChannel ch = SocketUtils.accept(javaChannel());
  4.     try {
  5.         if (ch != null) {
  6.             // 将包装后的 channel 添加至缓存
  7.             buf.add(new NioSocketChannel(this, ch));
  8.             return 1;
  9.         }
  10.     } catch (Throwable t) {
  11.         logger.warn("Failed to create a new channel from an accepted socket.", t);
  12.     }
  14.     return 0;
  15. }

现在，一个崭新的 NioSocketChannel 终于诞生了，它将作为参数，调用fireChannelRead。

WorkerGroup 工作

上述仅仅只是在 BossGroup 中工作，IO 事件已经就绪，ByteBuf 也已经准备好了，现在终于开始调用pipeline.fireChannelRead(byteBuf) 方法。

在一系列调用链中，会调用 ChannelInboundHandler#channelRead 方法，此方法仅当当前Channel已从对等方读取消息时调用。

这是个接口类，经过调试，这会调用 ServerBootstrap 下 channelRead 方法：

  1. public void channelRead(ChannelHandlerContext ctx, Object msg) {
  2.     final Channel child = (Channel) msg;
  3.     child.pipeline().addLast(childHandler);
  5.     try {
  6.         childGroup.register(child).addListener(new ChannelFutureListener() {
  7.             @Override
  8.             public void operationComplete(ChannelFuture future) throws Exception {
  9.                 if (!future.isSuccess()) {
  10.                     forceClose(child, future.cause());
  11.                 }
  12.             }
  13.         });
  14.     } catch (Throwable t) {
  15.         forceClose(child, t);
  16.     }
  17. }

这个方法向 pipeline 中添加了我们在启动程序中配置的处理者，然后调用了 childGroup.register，这与我们之前分析的 register 方法是一样的，只不过之前默认是采用父类的 BossGroup 的方法，而现在指明调用 childGroup.register 方法，也就是向 WorkerGroup 注册一个通道！不过现在这个通道已经是对应 IO 的 NioSocketChannel，而不再是服务端接收请求的通道了。

然后这又会经历：轮询得到 EventLoop -> 执行 register0 方法绑定 Channel -> 执行doStratThread 方法启动新线程 -> 事件循环 -> 执行 select()，select 必然会立即返回 -> 遍历结果 key 集合。

不过这一次，由于 Channel 已经是和客户端的连接，此次事件将会直接读取数据！

总结

Netty 线程模型其实核心就是 boss 和 worker 两个线程组，通过构造这两个线程组，Netty 完美的贴切了 Reactor 模型，并且通过调整参数，我们可以任意的选择不同的 Reactor 模式。

在 Netty 中，我们可以提交自己的任务， 但 Netty 中的 任务 和 IO 事件处理是在同一个事件循环中运行的，长时间的任务会堵塞 IO 事件的处理，Netty 中的 ioRatio 适当缓解了这个问题，但没有根治，我们仍需小心提交任务。

Netty 中使用优化的 SelectedSelectionKeySet，在 Netty 中，存储 key 集合不再是 HashSet，而是一个数组，这使得 add 十分高效，Netty 通过反射的方式巧妙的替换掉了原生 Selector 的字段。

Netty 自定义的 select 方法相较于原生方法更加高效，并且也解决了一些问题，例如空轮询 Bug，Netty 通过检测空轮询次数，一旦到达阈值，则重建 Selector。

一旦消息经 boss 到达 worker，并由 worker 经过相同 select 步骤后，消息就正式进入管道，在双端链表中传递。