一文搞懂Reactor模型与实现

绝地灬酷狼 2024-03-25 14:14 76阅读 0赞

周六在公司写**Reactor**模型，

一女同事问我为啥都**2023**年了还在学习**Reactor**模型呀，

我问她为啥快**30**的年纪了，周六还在公司看我写**Reactor**呀，

一时间办公室里，男的，女的，都沉默了。

--------------------

### 前言 ###

在网络**IO**设计中，有两种高性能模型：**Reactor**模型和**Proactor**模型。**Reactor**基于同步**IO**模式，**Proactor**基于异步**IO**模式。

**Netty**网络框架，**Redis**等中间件中都有使用到**Reactor**模型。本文将对**Reactor**模型的如下三种分类进行学习和实现。

1.  单**Reactor**单线程模型；
2.  单**Reactor**多线程模型；
3.  主从**Reactor**多线程模型。

### 正文 ###

#### 一. Reactor设计模式 ####

**Reactor**翻译过来的意思是：**反应堆**，所以**Reactor**设计模式本质是基于事件驱动的。在**Reactor**设计模式中，存在如下几个角色。

1.  **Handle**（事件）。**Reactor**整体是基于**Handle**进行驱动，这里的**Handle**叫做**事件**，可以类比为**BIO**中的**Socket**，**NIO**中的**Socket**管道。比如当**Socket**管道有连接建立，或者有数据可读，那么此时就称作**事件发生**；
2.  **EventHandler**(事件处理器)。有事件发生，就需要有相应的组件来处理事件，那么这里的组件就叫做事件处理器。**EventHandler**是一个抽象概念，其会有不同的具体实现，因为事件会有不同的类型，那么不同类型的事件，肯定都需要有相应的具体处理逻辑，这里的具体处理逻辑，就由**EventHandler**的具体实现来承载；
3.  **Concrete Event Handler**（具体事件处理器）。是**EventHandler**的具体实现，用于处理不同类型的事件；
4.  **Synchronous Event Demultiplexer**（事件多路分解器）。（这里将**Synchronous Event Demultiplexer**简称为**Demultiplexer**）**Demultiplexer**用于监听事件并得到所有发生事件的集合，在监听的状态下是阻塞的，直到有事件发生为止。**Demultiplexer**有一个很好的类比，就是**NIO**中的多路复用器**Selector**，当调用**Selector**的**select()** 方法后，会进入监听状态，当从**select()** 方法返回时，会得到**SelectionKey**的一个集合，而每一个**SelectionKey**中就保存着有事件发生的**Socket**管道；
5.  **Initiation Dispatcher**（事件分发器）。现在已经有**Concrete Event Handler**（具体事件处理器）来处理不同的事件，也能通过**Synchronous Event Demultiplexer**（事件多路分解器）拿到发生的事件，那么最后需要做的事情，肯定就是**将事件分发到正确的事件处理器上进行处理**，而**Initiation Dispatcher**就是完成这个分发的事情。

**Reactor**设计模式的一个简单类图，如下所示。![7113d30b435444e599d71ca199754362.png][]

通常，**Reactor**设计模式中的**Reactor**，可以理解为上述图中的**Synchronous Event Demultiplexer + Initiation Dispatcher**。

#### 二. 单Reactor单线程模型 ####

单**Reactor**单线程模型中，只有一个**Reactor**在监听事件和分发事件，并且监听事件，分发事件和处理事件都在一个线程中完成。示意图如下所示。

![format_png][]

上述示意图中，一次完整的处理流程可以概括如下。

1.  **Reactor**监听到**ACCEPT**事件发生，表示此时有客户端建立连接；
2.  **Reactor**将**ACCEPT**事件分发给**Acceptor**处理；
3.  **Acceptor**会在服务端创建与客户端通信的**client-socket**管道，然后注册到**IO**多路复用器**selector**上，并监听**READ**事件；
4.  **Reactor**监听到**READ**事件发生，表示此时客户端数据可读；
5.  **Reactor**将**ACCEPT**事件分发给**Handler**处理，**Handler**处理**READ**事件就会基于**client-socket**管道完成客户端数据的读取。

下面将基于**Java**语言，实现一个简单的单**Reactor**单线程模型的服务端，整体代码实现完全符合上述示意图，大家可以进行参照阅读。

首先实现**Reactor**，如下所示。

public class Reactor implements Runnable {
    
        private final Selector selector;
    
        public Reactor(int port) throws IOException {
            // 开启多路复用
            selector = Selector.open();
            // 服务端创建listen-socket管道
            ServerSocketChannel listenSocketChannel = ServerSocketChannel.open();
            // 绑定端口
            listenSocketChannel.socket().bind(new InetSocketAddress(port));
            // 设置为非阻塞模式
            listenSocketChannel.configureBlocking(false);
            // ACCEPT事件的附加器是Acceptor
            listenSocketChannel.register(selector, SelectionKey.OP_ACCEPT,
                    new Acceptor(selector, listenSocketChannel));
        }
    
        @Override
        public void run() {
            while (!Thread.interrupted()) {
                try {
                    // 获取发生的事件
                    selector.select();
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterable = selectionKeys.iterator();
                    while (iterable.hasNext()) {
                        // 对事件进行分发
                        dispatch(iterable.next());
                        iterable.remove();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
                LockSupport.parkNanos(1000 * 1000 * 1000);
            }
        }
    
        private void dispatch(SelectionKey selectionKey) {
            // 获取事件的附加器
            // ACCEPT事件的附加器是Acceptor，故由Acceptor来处理ACCEPT事件
            // READ事件的附加器是Handler，故由Handler来处理READ事件
            Runnable attachment = (Runnable) selectionKey.attachment();
            if (attachment != null) {
                attachment.run();
            }
        }
    
    }
    复制代码

已知**Reactor**会监听客户端连接的**ACCEPT**事件，还已知**ACCEPT**事件由**Acceptor**处理，所以在向多路复用器注册服务端用于监听客户端连接的**listen-socket**管道时，添加了一个**Acceptor**作为附加器，那么当发生**ACCEPT**事件时，就能够获取到作为**ACCEPT**事件附加器的**Acceptor**来处理**ACCEPT**事件。

下面看一下**Acceptor**的实现，如下所示。

public class Acceptor implements Runnable {
    
        private final Selector selector;
        private final ServerSocketChannel listenSocketChannel;
    
        public Acceptor(Selector selector, ServerSocketChannel listenSocketChannel) {
            this.selector = selector;
            this.listenSocketChannel = listenSocketChannel;
        }
    
        @Override
        public void run() {
            try {
                // 为连接的客户端创建client-socket管道
                SocketChannel clientSocketChannel = listenSocketChannel.accept();
                // 设置为非阻塞
                clientSocketChannel.configureBlocking(false);
                // READ事件的附加器是Handler
                clientSocketChannel.register(selector, SelectionKey.OP_READ,
                        new Handler(clientSocketChannel));
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
    }
    复制代码

在**Acceptor**中就是在服务端创建与客户端通信的**client-socket**管道，然后注册到多路复用器上并指定监听**READ**事件，同时又因为**READ**事件由**Handler**处理，所以还添加了一个**Handler**作为附加器，当**READ**事件发生时可以获取到作为**READ**事件附加器的**Handler**来处理**READ**事件。

下面看一下**Handler**的实现，如下所示。

public class Handler implements Runnable {
    
        private final SocketChannel clientSocketChannel;
    
        public Handler(SocketChannel clientSocketChannel) {
            this.clientSocketChannel = clientSocketChannel;
        }
    
        @Override
        public void run() {
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            try {
                // 读取数据
                int read = clientSocketChannel.read(byteBuffer);
                if (read <= 0) {
                    clientSocketChannel.close();
                } else {
                    System.out.println(new String(byteBuffer.array()));
                }
            } catch (IOException e1) {
                try {
                    clientSocketChannel.close();
                } catch (IOException e2) {
                    e2.printStackTrace();
                }
                e1.printStackTrace();
            }
        }
    
    }
    复制代码

在**Handler**中就是简单的读取数据并打印，当读取数据为空或者发生异常时，需要及时将管道关闭。

最后编写一个主程序将**Reactor**运行起来，如下所示。

public class MainServer {
    
        public static void main(String[] args) throws IOException {
            Thread reactorThread = new Thread(new Reactor(8080));
            reactorThread.start();
        }
    
    }
    复制代码

现在来思考一下，单**Reactor**单线程模型有什么优点和缺点。**优点**其实就是模型简单，实现方便。缺点有两点，如下所示。

1.  一个**Reactor**同时负责监听**ACCEPT**事件和**READ**事件；
2.  只有一个线程在工作，处理效率低，无法利用多核**CPU**的优势。

但是尽管单**Reactor**单线程模型有上述的缺点，但是著名的缓存中间件**Redis**的服务端，就是使用的单**Reactor**单线程模型，示意图如下。

![format_png 1][]

那为什么以性能著称的**Redis**会采取单**Reactor**单线程模型呢，其实就是因为**Redis**的操作都在内存中，读写都非常快速，所以单**Reactor**单线程模型也能运行得很流畅，同时还避免了多线程下的各种并发问题。

#### 三. 单Reactor多线程模型 ####

在理解了单**Reactor**单线程模型后，那么肯定就能想到，假如在**Handler**中处理**READ**事件的这个事情能够使用一个线程池来完成，从而就可以实现**READ**事件的处理不会阻塞主线程。而这样的一个模型，其实就是单**Reactor**多线程模型，示意图如下所示。

![format_png 2][]

和单**Reactor**单线程模型唯一的不同，就是在**Handler**中多了一个线程池。

单**Reactor**多线程模型的代码实现，除了**Handler**以外，其余和单**Reactor**单线程模型一摸一样，所以下面就看一下单**Reactor**多线程模型中的**Handler**实现，如下所示。

public class Handler implements Runnable {
    
        private static final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(16, 32,
                60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(200));
    
        private final SocketChannel clientSocketChannel;
    
        public Handler(SocketChannel clientSocketChannel) {
            this.clientSocketChannel = clientSocketChannel;
        }
    
        @Override
        public void run() {
            threadPool.execute(() -> {
                ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                try {
                    // 读取数据
                    int read = clientSocketChannel.read(byteBuffer);
                    if (read <= 0) {
                        clientSocketChannel.close();
                    } else {
                        System.out.println(new String(byteBuffer.array()));
                    }
                    // 睡眠10S，演示任务执行耗时长也不会阻塞处理其它客户端请求
                    LockSupport.parkNanos(1000 * 1000 * 1000 * 10L);
                } catch (IOException e1) {
                    try {
                        clientSocketChannel.close();
                    } catch (IOException e2) {
                        e2.printStackTrace();
                    }
                    e1.printStackTrace();
                }
            });
        }
    
    }
    复制代码

其实就是每一个**READ**事件的处理会作为一个任务被扔到线程池中去处理。

单**Reactor**多线程模型虽然解决了只有一个线程的问题，但是可以发现，仍旧是只有一个**Reactor**在同时监听**ACCEPT**事件和**READ**事件。

那么现在思考一下，为什么一个**Reactor**同时监听**ACCEPT**事件和**READ**事件是不好的。其实就是因为通常客户端连接的建立是不频繁的，但是连接建立后数据的收发是频繁的，所以如果能够将监听**READ**事件这个动作拆分出来，让多个子**Reactor**来监听**READ**事件，而原来的主**Reactor**只监听**ACCEPT**事件，那么整体的效率，会进一步提升，而这，就是主从**Reactor**多线程模型。

#### 四. 主从Reactor多线程模型 ####

主从**Reactor**模型中，有一个主**Reactor**，专门监听**ACCEPT**事件，然后有多个从**Reactor**，专门监听**READ**事件，示意图如下所示。

![format_png 3][]

上述示意图中，一次完整的处理流程可以概括如下。

1.  主**Reactor**监听到**ACCEPT**事件发生，表示此时有客户端建立连接；
2.  主**Reactor**将**ACCEPT**事件分发给**Acceptor**处理；
3.  **Acceptor**会在服务端创建与客户端通信的**client-socket**管道，然后注册到从**Reactor**的**IO**多路复用器**selector**上，并监听**READ**事件；
4.  从**Reactor**监听到**READ**事件发生，表示此时客户端数据可读；
5.  从**Reactor**将**ACCEPT**事件分发给**Handler**处理，**Handler**处理**READ**事件就会基于**client-socket**管道完成客户端数据的读取。

下面将基于**Java**语言，实现一个简单的主从**Reactor**多线程模型的服务端，整体代码实现完全符合上述示意图，大家可以进行参照阅读。

首先是主**Reactor**的实现，如下所示。

public class MainReactor implements Runnable {
    
        private final Selector selector;
    
        public MainReactor(int port) throws IOException {
            // 开多路复用器
            selector = Selector.open();
            // 服务端创建listen-socket管道
            ServerSocketChannel listenSocketChannel = ServerSocketChannel.open();
            // 设置为非阻塞
            listenSocketChannel.configureBlocking(false);
            // 绑定监听端口
            listenSocketChannel.socket().bind(new InetSocketAddress(port));
            // 将listen-socket管道绑定到主Reactor的多路复用器上
            // 并且主Reactor上只会注册listen-socket管道，用于监听ACCEPT事件
            listenSocketChannel.register(selector, SelectionKey.OP_ACCEPT,
                    new Acceptor(listenSocketChannel));
        }
    
        @Override
        public void run() {
            while (!Thread.interrupted()) {
                try {
                    selector.select();
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterable = selectionKeys.iterator();
                    while (iterable.hasNext()) {
                        // 对事件进行分发
                        dispatch(iterable.next());
                        iterable.remove();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
                LockSupport.parkNanos(1000 * 1000 * 1000);
            }
        }
    
        private void dispatch(SelectionKey selectionKey) {
            // 获取事件附加器，只会是Acceptor
            Runnable attachment = (Runnable) selectionKey.attachment();
            if (attachment != null) {
                attachment.run();
            }
        }
    
    }
    复制代码

主**Reactor**的实现中，还是先创建服务端监听客户端连接的**listen-socket**管道，然后注册到主**Reactor**的**IO**多路复用器上，并监听**ACCEPT**事件，同时我们现在知道，主**Reactor**的**IO**多路复用器上只会注册**listen-socket**管道且只会监听**ACCEPT**事件。同样，也添加了一个**Acceptor**作为附加器，那么当发生**ACCEPT**事件时，就能够获取到作为**ACCEPT**事件附加器的**Acceptor**来处理**ACCEPT**事件。

下面是**Acceptor**的实现，如下所示。

public class Acceptor implements Runnable {
    
        // 指定从Reactor一共有16个
        private static final int TOTAL_SUBREACTOR_NUM = 16;
    
        // 服务端的listen-socket管道
        private final ServerSocketChannel listenSocketChannel;
    
        // 用于运行从Reactor
        private final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
                TOTAL_SUBREACTOR_NUM, TOTAL_SUBREACTOR_NUM * 2,
                60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(200));
    
        // 从Reactor集合
        private final List<SubReactor> subReactors = new ArrayList<>(TOTAL_SUBREACTOR_NUM);
    
        public Acceptor(ServerSocketChannel listenSocketChannel) throws IOException {
            this.listenSocketChannel = listenSocketChannel;
            // 将从Reactor初始化出来并运行
            for (int i = 0; i < TOTAL_SUBREACTOR_NUM; i++) {
                SubReactor subReactor = new SubReactor(Selector.open());
                subReactors.add(subReactor);
                threadPool.execute(subReactor);
            }
        }
    
        @Override
        public void run() {
            try {
                // 为连接的客户端创建client-socket管道
                SocketChannel clientSocketChannel = listenSocketChannel.accept();
                // 设置为非阻塞
                clientSocketChannel.configureBlocking(false);
                // 任意选择一个从Reactor，让其监听连接的客户端的READ事件
                Optional<SubReactor> anySubReactor = subReactors.stream().findAny();
                if (anySubReactor.isPresent()) {
                    SubReactor subReactor = anySubReactor.get();
                    // 从Reactor的多路复用器会阻塞在select()方法上
                    // 这里需要先唤醒多路复用器，立即从select()方法返回
                    subReactor.getSelector().wakeup();
                    // 让从Reactor负责处理客户端的READ事件
                    clientSocketChannel.register(subReactor.getSelector(), SelectionKey.OP_READ,
                            new Handler(clientSocketChannel));
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
    }
    复制代码

首先在**Acceptor**的构造函数中，会将所有从**Reactor**初始化出来，并且每一个从**Reactor**都会持有一个**IO**多路复用器。当一个从**Reactor**创建出来后就会立即运行，此时从**Reactor**的**IO**多路复用器就会开始监听，即阻塞在**select()** 方法上。

然后在**Acceptor**的主体逻辑中，会为连接的客户端创建**client-socket**管道，然后从所有从**Reactor**中基于某种策略（随机）选择一个从**Reactor**，并将**client-socket**管道注册在选择的从**Reactor**的**IO**多路复用器上，有一点需要注意，此时从**Reactor**的**IO**多路复用器可能会阻塞在**select()** 方法上，所以注册前需要先通过**wakeup()** 方法进行唤醒。

接下来继续看从**Reactor**的实现，如下所示。

public class SubReactor implements Runnable {
    
        private final Selector selector;
    
        public SubReactor(Selector selector) {
            this.selector = selector;
        }
    
        @Override
        public void run() {
            while (!Thread.interrupted()) {
                try {
                    selector.select();
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()) {
                        // 对事件进行分发
                        dispatch(iterator.next());
                        iterator.remove();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
                LockSupport.parkNanos(1000 * 1000 * 1000);
            }
        }
    
        private void dispatch(SelectionKey selectionKey) {
            // 获取事件附加器，只会是Handler
            Runnable runnable = (Runnable) selectionKey.attachment();
            if (runnable != null) {
                runnable.run();
            }
        }
    
        public Selector getSelector() {
            return selector;
        }
    
    }
    复制代码

从**Reactor**的实现中，会监听服务端为连接的客户端创建的**client-socket**管道上的**READ**事件，一旦有**READ**事件发生，就会使用作为附加器的**Handler**来处理**READ**事件。同样，从**Reactor**的**IO**多路复用器上只会注册**client-socket**管道且只会监听**READ**事件。

然后是**Handler**，因为是多线程模型，所以其实现和第三节中的**Handler**完全一样，下面再贴一下代码。

最后编写一个主程序将主**Reactor**运行起来，如下所示。

public class MainServer {
    
        public static void main(String[] args) throws IOException {
            Thread mainReactorThread = new Thread(new MainReactor(8080));
            mainReactorThread.start();
        }
    
    }

### 总结 ###

**Reactor**模型主要就是**监听事件**，**分发事件**和**处理事件**。其中**Reactor**角色会负责**监听事件** 和**分发事件**，**Handler**角色和**Acceptor**角色会负责**处理事件**。

**Reactor**模型虽然分为：单**Reactor**单线程模型，单**Reactor**多线程模型和主从**Reactor**多线程模型，但是其本质就是**NIO**的实现，是不过套了**Reactor**设计模式的外壳。

在网络通信框架**Netty**中，三种**Reactor**模型都有使用到，所以想要学习**Netty**的精髓，理解**Reactor**模型是必不可少的。

[7113d30b435444e599d71ca199754362.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/5a6103839a054adeb9cc03c29c77dbe8.png
[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/6d8f2ac4e8cb4c549b7146f310d5bce8.png
[format_png 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/ea8811b35ee448e4a78a44fca780ab47.png
[format_png 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/c25fcadd68634ed49da6fa3574d0664c.png
[format_png 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/2e00e7bec61d437daa8ad6a9015895d5.png