Netty的Reactor

前言

其实这个应该是写在 netty 的 epoll 前面的，只是刚好在看 epoll，而且 reactor 也是基于 epoll 这种操作系统带来的异步实现的。所以补上

Reactor

reactor 的设计模式提出来源于论文，也是得益于 select（最早出现在 4.2BSD 版本中，大约在 1983 年）这种操作系统层对异步事件通知的实现。论文中用一个日志服务来进行表述为什么需要事件驱动模式，而不是使用多线程模式：

• 性能 Effificiency
  • 多线程下由于会出现上下文切换，同步和数据迁移造成性能弱
• 开发难度 Programming simplicity
  • 多线程可能需要更复杂的控制并发的模型
• 可移植 Portability
  • Threading is not available on all OS platforms （论文提出时间是 1995 年，我也不清楚这句话到底该怎么说）

为了避免上面的问题还需要做到下面的事情：

• 高可用 Availability
  • 服务能够尽量多的接受客户端的请求，而且新的请求不会阻塞现有正在接受的数据。（这里感觉是高吞吐）
• 高性能 Efficiency
  • 最小化延迟，最大化吞吐，减少 CPU 空闲时间
• 易编写 Programming simplicity
  • 简化设计，而且有比较适当的并发策略。
• 灵活 Adaptability
  • 如果修改了入 tcp 中的报文，对整个代码的改动较小
• 可移植性 Portability
  • 能够比较容易迁移代码

论文中提到的组件为：

• Handles
  • 操作系统提供的有唯一标识的资源，他可以是网络连接，打开的文件，定时器，同步对象等。
• Synchronous Event Demultiplexer
  • 同步事件分离器，阻塞等待事件发生 t，当某个 handle 需要处理的事件发生后会被唤醒。如 linux 下 select、epoll 等。
• Initiation Dispatcher
  • 初始化调度器接口 ，用于注册、移除和分派事件处理的 handler。同步事件分离器负责等待事件发生，当它检测到新的事件后，会通知初始化调度器，让初始化调度器调用指定的事件处理 handler。如 连接接受，数据输入和输出，定时器到期等
• Event Handler
  • 事件处理 handler，用于服务处理事件，一般是应用程序实现

他们处理关系如下：

整个的过程其实就是在主程序中首先获取到注册事件并且注册到调度器中，在调度器中执行 select，一直等到事件发生，然后通知初始调度器执行具体的 handle。

对此在 IO 多路复用的时候还应用了 Doug Lea 的NIO中的图，这里就不再啰嗦，主要是对论文中的抽象和考虑的问题进行一个简单的介绍和汇总。

Netty 的实现

Netty 是一个实现了高性能的网络 java 基础组件，所以他的事件包含的主要是网络连接中的 Accept、Connect、Read 和 write。设计中完美的实现上文中提到的需求，即编写代码容易，得益于语言特性，可移植性也高，使用 epoll 这种 IO 多路复用让他的性能也很强。

启动

首先看下他的启动

server

netty 中的 example 中有个 echo 的简单 demo,这个 demo 对了解实现就已经差不多足够了。

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
      EventLoopGroup workerGroup = new NioEventLoopGroup();
      final EventLoopGroup myGroup = new NioEventLoopGroup(10);
      final EchoServerHandler serverHandler = new EchoServerHandler();
      try {
          ServerBootstrap b = new ServerBootstrap();
          b.group(bossGroup, workerGroup)
           .channel(NioServerSocketChannel.class)
           .option(ChannelOption.SO_BACKLOG, 100)
           .handler(new LoggingHandler(LogLevel.INFO))
           .childHandler(new ChannelInitializer<SocketChannel>() {
               @Override
               public void initChannel(SocketChannel ch) throws Exception {
                   ChannelPipeline p = ch.pipeline();
                   p.addLast(myGroup,serverHandler);
               }
           });
          // Start the server.
          ChannelFuture f = b.bind(PORT).sync();
          // Wait until the server socket is closed.
          f.channel().closeFuture().sync();

启动中有 EventLoopGroup 这个类，该类就是 select 的具体实现，为什么有两个呢，主要是想要实现主从 reactor，第一个主要负责连接相关的操作，一般线程数是 1，第二个则是负责具体的读写事件，线程池默认是 CPU 核心数的 2 倍。这上面还涉及到两个 Channel

• NioServerSocketChannel
  • 主要是实现了 accept 方法的 ServerSocketChannel
• SocketChannel
  • 实现了 read ，write 等的 SocketChannel

注意的是，上面提到的两个 java 的 NIO 中的 channel 都实现了 AbstractSelectableChannel，该 channel 里面的实现就能够被 select 或者 epoll 方法进行监听。这个是 java 进行的一个抽象。从上面的两个绑定就可以看出，在 ServerBootstrap 上绑定的就是 accept 方法，在 childHandler 中绑定的就是 read 和 write 方法。ChannelOption 主要是和网络连接等相关的配置，即用来配置 channel 的。然后就是两个 handler 的区别了，在 ServerBootstrap 上绑定到 handler 主要是对 accept 的操作进行处理，即响应 accept 方法的事件。

还有一个类 ChannelPipeline，这个类就是管道，也就是说 channel 里面的数据是通过 Pipeline 进行处理的，每一个 handler 就是一个流水线上的螺丝钉。在每一个 channel，也就是网络连接创建的时候，都会新建一个 ChannelPipeline。也就是说 channel 和 ChannelPipeline 是一对一的关系，但是 handler 是自定义的对象，所以他不是一对一的关系，需要将 ChannelPipeline 和 handler 进行连接，使用的是 ChannelHandlerContext。在 ChannelPipeline 中，每一个 handler 都会被封装成 ChannelHandlerContext，然后根据 handler 加入到 ChannelPipeline 的形成一个 filterchan，此处使用的设计模式为拦截过滤器模式，三者之间画图为关系为：

初始化

上文提到的提到的 handler 包含了负责 accept 的 handler 和 read write 的 handler。他们之间是如何进行数据传递和初始化的呢？在最开始的 bind 中，最终会执行到 ServerBootstrap 中的 init 方法中，init 方法:

setChannelOptions(channel, newOptionsArray(), logger);
setAttributes(channel, newAttributesArray());
ChannelPipeline p = channel.pipeline();
final EventLoopGroup currentChildGroup = childGroup;
final ChannelHandler currentChildHandler = childHandler;
final Entry<ChannelOption<?>, Object>[] currentChildOptions = newOptionsArray(childOptions);
final Entry<AttributeKey<?>, Object>[] currentChildAttrs = newAttributesArray(childAttrs);
p.addLast(new ChannelInitializer<Channel>() {
    @Override
    public void initChannel(final Channel ch) {
        final ChannelPipeline pipeline = ch.pipeline();
        ChannelHandler handler = config.handler();
        if (handler != null) {
            pipeline.addLast(handler);
        }

        ch.eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                pipeline.addLast(new ServerBootstrapAcceptor(
                        ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
            }
        });
    }
});

这个 init 方法主要的就是给负责 accept 的 reactor 注册 ServerBootstrapAcceptor，整个方法主要还是将一些设置的参数进行配置，然后在当前的 reactor 里面注册了增加了一个 ServerBootstrapAcceptor，此时还处于被封装为一个专门用于初始化 handler 的 handler 中。初始化工作完成后，就进行 register，最后仍然会走 AbstractChannel 里面的 register0 方法，将当前的 ACCEPT 事件注册（ACCEPT 事件是在 Channel 最开始初始化的时候传入的）。主要注意的是，也是在 register0 这个方法中，我们的 eventLoop 正式开始执行线程。

那么这个 ServerBootstrapAcceptor 主要做的工作是什么呢，核心其实就是将当前的 channel 作为事件注册给 childGroup。

final Channel child = (Channel) msg;
         child.pipeline().addLast(childHandler);
         setChannelOptions(child, childOptions, logger);
         setAttributes(child, childAttrs);

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

这里我觉得 main reactor 和 child reatcor 最大的区别就是，main reactor 传入到后面的是个 channel，childgroup 往 handler 里面传入的数据是 bytebuf，也就是真实的数据。

简单的流程图和关系：

client

client 端相对于 server 最大的变化其实就是没有了 main reactor，也就是没有将现在的 channel 注册给 child reactor 的步骤。而是将 accept 的实现修改为了 connect。

总结

很简单的梳理了下 reactor 在 netty 中的逻辑，甚至可以说只是大概解释了下各个组件的使用。可以看到 netty 中的 reactor 模型其实是上文提到 NIO 的文档中的主从模式。netty 做到了 reactor 论文中提到的大部分想要的特性，他的 Channel 只会在一个 eventLoop 中被执行，而且一个 eventloop 只会绑定一个线程，这样减少了并发，得益于 java 的特性，可移植性也比较高。高性能就不用说了。在代码层面比较简洁，而且相对灵活，比如每一个 handler 可以注册自己使用的 EventLoopGroup，因为 EventLoopGroup 是继承 ScheduledExecutorService，也就是说可以自定义处理的线程池。