Reator 模式 + Netty 线程模型 + 最佳实践建议

Reactor 模式

大部分网络框架的设计都基于 Reactor 模式。
这种模式基于事件驱动，特别适合处理大量的 IO 事件。
 

根据线程数量，我们可以将 Reactor 模式大致分为以下3种（以服务端实现为例）：
 

单线程 Reactor

单个 Reactor 线程负责对TCP链路读写数据和编解码（包括执行业务逻辑）。
（很多人把该模式称为 “1 - 1”。其实这种称呼并不贴切。）
 

适用场景

适合 并发度低、请求处理快 的小应用。

 

缺陷

不适合 高并发、高负载的场景。因为：

• 单线程处理大量并发链路时性能不高，也无法发挥多核计算机的优势。无法满足大量消息的编解码和读写需求。
• 单线程负载过高后，处理速度变慢，可能导致大量客户端连接超时。超时会引起消息重发，线程负载更重。最终大量消息积压、超时，成为系统的性能瓶颈。
• 单线程可靠性不足。单个线程意外终止或陷入死循环，会导致整个系统通信瘫痪，无法接受和处理外部请求。

 

 

多线程 Reactor

与单线程Reactor模式不同，此模式用一个 NIO线程池 代替原来的单个Reactor线程。
 

通常，Reactor 线程池中，每个线程可以同时处理 N条链路；
但是一个链路只能由1个线程处理，以防止线程安全问题。
 

缺陷

这种模式可以满足绝大多数场景的性能需求。
但是 单线程Acceptor 也可能成为性能瓶颈。客户端连接非常多时，处理客户端请求连接/安全认证等操作也会非常耗性能。

 
 

 

主从 多线程 Reactor

为了解决 单线程Acceptor 的性能问题，又衍生出了第三种模式 —— “主从 多线程 Reactor”。
即，用一个 NIO线程池 专门处理那些与耗时的非业务性操作。
大致过程如下：

1. 开始监听连接：由一个线程作为 Acceptor，绑定监听端口接收客户端连接。
   具体实现时，这个Acceptor线程可以是从“主Reactor线程池”中随机选定的一个线程
2. 收到连接 并 分配处理线程：Acceptor 接收到客户端连接，并创建 SocketChannel，并将其注册到 主线程池 的 Reactor 线程上。
3. 执行非业务性操作：主线程池的 Reactor 线程负责 接入认证、IP黑白名单过滤、握手等非业务性操作。
4. 执行业务逻辑：执行完上一步，业务层链路正式建立。SocketChannel 从 主线程池 中线程的多路复用器上摘除，重新注册到 从线程池 的线程上，执行处理后续业务操作。

 

 

Netty 线程模型

Netty 的线程模型与上述三种 Reactor 线程模型相似。
Netty 提供了便捷的API来实现相关线程配置。通过 NioEventLoopGroup 的构造方法 和 ServerBootstrap.group() 方法就可以实现相应的线程配置。
 

可以简单地理解为：
一个 EventLoop 对应一个 Reactor 线程；
EvenLoopGroup 则对应 线程池（ExecutorService）。
 

Netty 官方示例

 

 

Netty 线程开发最佳实践

时间可控的简单业务 直接在 IO线程 上处理

如果业务非常简单，执行时间非常短，不需要访问外部资源（如，网络、数据库、磁盘等）时，可以直接在 ChannelHandler 中执行业务。
这样实现简单，避免线程上下文切换，也不会有线程安全问题。

 

复杂、时间不可控的业务 投递到后端业务线池处理

可以将这类业务封装成 Task，投递到后端的业务线程池处理。
因为过多的 业务ChannelHandler 会降低开发效率，增加维护成本。
不要把 Netty 当作业务容器。

 

业务线程 不要直接操作 ChannelHandler

（从某种角度而言，这算是上一条的延伸。）
业务通常是多线程模型处理的，如果业务线程直接操作 ChannelHandler，就需要处理线程安全问题。
可以参照Netty自身的做法，将操作封装成独立的 Task 由 NioEventLoopGroup 统一调度。
 

我们可以在Netty源码中找很多类似如下的代码：

Netty ChunkedWriteHandler 中的 resumeTransfer() 方法：

public void resumeTransfer() {
  ...
  if (ctx.executor.inEventLoop()) {
    resumeTransfer0(ctx);
  } else {
    // let the transfer resume on the next event loop round
    ctx.executor.execute(() -> resumeTransfer0(ctx));
  }
}