3 | Netty源码：什么是Reactor模型

一时失言乱红尘 2023-07-03 06:07 7阅读 0赞

> 今天是2020年2月2号，感觉是一个比较特殊的日子，今天就来一篇记录型的博客吧，哈哈

> 起初很好奇，到底什么叫Reactor模式，这个名词感觉特别高大上，然后看描述，虽然能看懂描述，但是却不是特别明白到底是什么意思。这个时候主要是没有形成一种直观的印象，直观的印象就是比如说苹果，再给你看个实物，你就能把苹果与关联起来。在学习netty的时候，也遇到了Reactor模式，于是有了机会来形成一种比较直观的印象。

## 什么是Reactor模式？ ##

定义看起来很抽象，但是其实很好理解。\*\*它是一种开发模式，模式的核心流程：注册感兴趣的事件 -> 扫描是否有感兴趣的事件发生 -> 事件发生后做出相应的处理。\*\*仅此而已。使用BIO开发的时候，每有一个新的请求过来了，都会新开一个线程，然后在新的线程里面进行业务处理，这种处理方式就是Thread-Per-Connection；  
![format_png][]  
所以对应起来，使用NIO开发的时候，也有一个模式去处理相应的请求与业务逻辑，叫做Reactor模式。至于具体怎么做，也就是前面提到的Reactor模式的核心流程。

## Reactor模式的3种版本 ##

开始这个之前我有一个疑问：**Thread-Per-Connection与Reactor单线程有什么关系？**

### Thread-Per-Connection模式 ###

示意图：  
\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-9tyrEyYU-1580657917209)(https://gitee.com/sasurai/pics/raw/master/mac/20200201181549.png)\]

伪代码：  
![format_png 1][]

### Reactor单线程模式 ###

![format_png 2][]

**从这张图里面看不懂其执行流是什么样的。待后续理解了再补上解读。**

### Reactor多线程模式 ###

![format_png 3][]

### 主从Reactor多线程模式 ###

对服务器开发来说，很重要的事情是**接收连接**，accept事件会被单独注册到另外一个reactor中。

![format_png 4][]

## 在Netty中如何实现Reactor三种模式 ##

![format_png 5][]

其中单线程和非主从reactor多线程模式的差别只在于new的时候传入的线程数量，不传的话，会默认以CPU的核数为依据来确定最终的线程数。

## Netty 如何支持主从 Reactor 模式 ##

以netty项目源代码（分支4.1）中netty-example模块的`EchoServer`为例。

### 保存 ###

它是一个主从reactor多线程模式，其中`bossGroup`负责accept事件，`workerGroup`负责逻辑处理。

![format_png 6][]

在①中，分别将两个`EventLoopGroup`传入到`ServerBootstrap`中，并将这两个`EventLoopGroup`保存起来。

![format_png 7][]

步骤②执行的保存逻辑如下：

![format_png 8][]

步骤③即已保存完毕。保存起来之后，什么时候使用呢？

### 将channel注册到`parentGroup` ###

先看`parentGroup`的使用过程，找到使用了`group`这个变量的地方`Ctrl + B`：

![format_png 9][]

进去之后，是一个类似于普通getter方法

![format_png 10][]

只有一个地方调用，名称也叫`group()`，所以还可以继续往上看调用者

![format_png 11][]

然后使用`Ctrl + Alt + H`查看该`group()`方法的调用者：

![format_png 12][]

在`initAndRegister()`中可以找到将`channel`（即`ServerSocketChannel`）注册到该`EventLoopGroup`的代码，如下：

![format_png 13][]

绑定完毕。

### 将`channel`注册到`childGroup` ###

找到使用了`childGroup`这个变量的地方`Ctrl + B`：

![format_png 14][]

只有个地方使用到了该`childGroup`，并改名成了`currentChildGroup`。

![format_png 15][]

①改名，②将`childGroup`作为一个变量，传入`ServerBootstrapAcceptor`中。`ServerBootstrapAcceptor`继承自`ChannelInboundHandlerAdapter`，其覆盖了父类的`channelRead()`方法，其中将新进来的`channel`注册到`childGroup`中。

![format_png 16][]

也就是说，新进来的连接，即`SocketChannel`，都会被注册到`childGroup`中。

### 新连接建立进行哪些初始化 ###

回到上面的`init()`方法中提出的，它是何时被调用的这个问题中。

它是`AbstractBootstrap`抽象类中的一个抽象方法，有两个类继承自`AbstractBootstrap`，分别是`Bootstrap`和`ServerBootstrap`。调用`init()`方法的地方只有一个，即`initAndRegister()`中。

![format_png 17][]

其中传入`init()`的`channel`为`ServerSocketChannel`，其大致过程：

当服务端即`EchoServer`启动的时候，会为`ServerSocketChannel`的pipeline添加一个`ServerBootstrapAcceptor`，所以每当有来自客户端的请求时，都会首先经过`ServerBootstrapAcceptor`，让它先处理，而它的处理内容就是将`SocketChannel`注册到`childGroup`中。

## 为什么说 Netty 的 main reactor 大多并不能用到一个线程组，只能线程组里面的一个？ ##

因为服务端只会启动一次，只有在启动过程中去绑定端口号时才会将`ServerSocketChannel`绑定到`main reactor`上。所以这时候要从`initAndRegister`的调用者逐级往上查看，如下；

public ChannelFuture bind(InetAddress inetHost, int inetPort) {
        
        return bind(new InetSocketAddress(inetHost, inetPort));
    }
    
    public ChannelFuture bind(SocketAddress localAddress) {
        
        validate();
        return doBind(ObjectUtil.checkNotNull(localAddress, "localAddress"));
    }
    
    private ChannelFuture doBind(final SocketAddress localAddress) {
        
        final ChannelFuture regFuture = initAndRegister();
        final Channel channel = regFuture.channel();
        ...
    }
    
    final ChannelFuture initAndRegister() {
        
        Channel channel = null;
        try {
        
            channel = channelFactory.newChannel();
            // 设置新接入连接的SocketChannel注册到sub reactor
            init(channel);
        } catch (Throwable t) {
        
            ...
        }
        // 注册ServerSocketChannel到main reactor
        ChannelFuture regFuture = config().group().register(channel);
        ...
    }

所以一个`ServerSocketChannel`只会注册到一个group中。但还是个疑问，是与EventLoopGroup相关的，留待后续再来回答。这个问题的意思是说，**只能用线程组里面的一个线程**，为什么？为什么不能多个线程？下面这个问题可以回答这个疑问！

## Netty 给 Channel 分配 NIO event loop 的规则是什么 ##

从`initAndRegister()`中的`config().group().register(channel)`代码出发，也就是`ServerSocketChannel`注册到main reactor中的那段代码（参见上面）。

// 从register方法进入
    ChannelFuture regFuture = config().group().register(channel);
    
    // 进入EventLoopGroup.java，它是一个接口。
    // NIO选MultithreadEventLoopGroup的实现
    ChannelFuture register(Channel channel);
    
    @Override
    public ChannelFuture register(Channel channel) {
        
        return next().register(channel);
    }
    
    @Override
    public EventLoop next() {
        
        return (EventLoop) super.next();
    }
    
    // 进入父类的next()实现中
    @Override
    public EventExecutor next() {
        
        return chooser.next();
    }
    
    // 进入chooser的next()方法，发现这个chooser是一个接口类型
    private final EventExecutorChooserFactory.EventExecutorChooser chooser;
    public interface EventExecutorChooserFactory {
        
        EventExecutorChooser newChooser(EventExecutor[] executors);
        @UnstableApi interface EventExecutorChooser {
        
            EventExecutor next();
        }
    }
    
    // chooser的初始化是根据传入的线程数决定的
    // 在MultithreadEventExecutorGroup的构造函数中
    children = new EventExecutor[nThreads];
    // 需要多少个线程，就有多少个EventExecutor，它初步与Thread等价
    chooser = chooserFactory.newChooser(children);

所以`chooser.next()`返回的是一个等价于Thread的对象，也就是说这个`ServerSocketChannel`只会在这个Thread中进行接收。其中的chooser就是根据线程数的个数，来选取一个线程分配给register进来的`ServerSocketChannel`。具体分配策略：

// next()是一个抽象方法，它的具体实现有两种
    public EventExecutorChooser newChooser(EventExecutor[] executors) {
        
        if (isPowerOfTwo(executors.length)) {
        
            return new PowerOfTwoEventExecutorChooser(executors);
        } else {
        
            return new GenericEventExecutorChooser(executors);
        }
    }
    
    // 判断是否为2的幂的简便方法
    private static boolean isPowerOfTwo(int val) {
        
        return (val & -val) == val;
    }
    
    private static final class PowerOfTwoEventExecutorChooser implements EventExecutorChooser {
        
        private final AtomicInteger idx = new AtomicInteger();
        private final EventExecutor[] executors;
    
        PowerOfTwoEventExecutorChooser(EventExecutor[] executors) {
        
            this.executors = executors;
        }
    
        @Override
        public EventExecutor next() {
        
            return executors[idx.getAndIncrement() & executors.length - 1];
        }
    }
    
    private static final class GenericEventExecutorChooser implements EventExecutorChooser {
        
        private final AtomicInteger idx = new AtomicInteger();
        private final EventExecutor[] executors;
    
        GenericEventExecutorChooser(EventExecutor[] executors) {
        
            this.executors = executors;
        }
    
        @Override
        public EventExecutor next() {
        
            return executors[Math.abs(idx.getAndIncrement() % executors.length)];
        }
    }

至此，上面的那个疑问算是有了一个答案。

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvbWFjLzIwMjAwMjAxMTc0OTA0LnBuZw?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvbWFjLzIwMjAwMjAxMTgxNjUwLnBuZw?x-oss-process=image/format,png
[format_png 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvbWFjLzIwMjAwMjAxMTgxODA5LnBuZw?x-oss-process=image/format,png
[format_png 3]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvbWFjLzIwMjAwMjAxMTgxODUwLnBuZw?x-oss-process=image/format,png
[format_png 4]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvbWFjLzIwMjAwMjAxMTgxOTIxLnBuZw?x-oss-process=image/format,png
[format_png 5]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvbWFjLzIwMjAwMjAxMTgyMDQxLnBuZw?x-oss-process=image/format,png
[format_png 6]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvLy8yMDIwMDIwMjE3MTkzMi5wbmc?x-oss-process=image/format,png
[format_png 7]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvLy8yMDIwMDIwMjE3MjI1OS5wbmc?x-oss-process=image/format,png
[format_png 8]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvd2luLzIwMjAwMjAyMTcyNTMxLnBuZw?x-oss-process=image/format,png
[format_png 9]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvd2luLzIwMjAwMjAyMTcyODQyLnBuZw?x-oss-process=image/format,png
[format_png 10]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvd2luLzIwMjAwMjAyMTczMDQzLnBuZw?x-oss-process=image/format,png
[format_png 11]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvd2luLzIwMjAwMjAyMTczMTM4LnBuZw?x-oss-process=image/format,png
[format_png 12]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvd2luLzIwMjAwMjAyMTczMzIxLnBuZw?x-oss-process=image/format,png
[format_png 13]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvd2luLzIwMjAwMjAyMTczNDU0LnBuZw?x-oss-process=image/format,png
[format_png 14]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvd2luLzIwMjAwMjAyMTc0NTIzLnBuZw?x-oss-process=image/format,png
[format_png 15]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvd2luLzIwMjAwMjAyMTc1MjM1LnBuZw?x-oss-process=image/format,png
[format_png 16]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvd2luLzIwMjAwMjAyMTc1ODE4LnBuZw?x-oss-process=image/format,png
[format_png 17]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9naXRlZS5jb20vc2FzdXJhaS9waWNzL3Jhdy9tYXN0ZXIvd2luLzIwMjAwMjAyMTgxNzUwLnBuZw?x-oss-process=image/format,png