NIO 下 Reactor 多线程

有的时候博客内容会有变动，首发博客是最新的，其他博客地址可能会未同步, 认准 https://blog.zysicyj.top

全网最细面试题手册，支持艾宾浩斯记忆法。这是一份最全面、最详细、最高质量的 java 面试题，不建议你死记硬背，只要每天复习一遍，有个大概印象就行了。https://store.amazingmemo.com/chapterDetail/1685324709017001`

image.png
image.png

NIO 下的 Reactor 多线程模型

在 Java NIO 中，Reactor 模式是一种用于处理并发网络连接的设计模式。它允许系统以非阻塞的方式处理多个客户端连接，这是通过单个或多个输入输出线程来实现的。在多线程 Reactor 模型中，可以有多个 Reactor 来处理不同的任务，如连接接受、读 / 写处理等。

基本概念

在深入了解 Reactor 多线程模型之前，我们需要理解以下几个关键概念：

  • Selector（选择器）：一个 Selector 可以检测多个注册的通道上是否有事件发生，如果有事件发生，便获取事件然后进行相应的处理。
  • Channel（通道）：类似于流，但主要区别在于它可以进行非阻塞的读写。
  • Buffer（缓冲区）：数据的容器，Channel 读写数据都是通过 Buffer 进行的。

Reactor 多线程模型

在 Reactor 多线程模型中，通常包含以下几个组件：

  • Main Reactor：负责处理客户端的连接请求。
  • Sub Reactor：负责处理非阻塞的读 / 写事件。
  • Worker Pool：一组工作线程，用于执行耗时操作，如业务处理、数据库操作等。

工作流程

  1. Main Reactor 线程：当客户端发起连接请求时，Main Reactor 通过 Selector 监听 ACCEPT 事件。一旦 ACCEPT 事件到达，Main Reactor 将处理连接请求，创建一个新的 SocketChannel，并将其注册到 Sub Reactor 的 Selector 上，监听 READ 或 WRITE 事件。
  2. Sub Reactor 线程：Sub Reactor 继续监听已连接的 SocketChannel 上的 READ 或 WRITE 事件。当这些事件发生时，Sub Reactor 将相应的 Channel 分配给一个线程（通常来自线程池）进行非阻塞的读写操作。
  3. Worker Pool 线程：一旦读写操作完成，Worker Pool 中的线程可以进行进一步的处理，如业务逻辑处理或数据库操作。处理完成后，结果可以返回给 Sub Reactor 线程，由它来完成数据的最终发送。

优点

  • 高效的资源利用：由于使用非阻塞 IO，线程可以在没有 IO 操作时处理其他任务，提高资源利用率。
  • 可扩展性：可以通过增加更多的 Sub Reactor 线程来处理更多的并发连接，从而提高系统的可扩展性。
  • 快速响应：Reactor 模式可以快速响应网络事件，提高系统的响应能力。

缺点

  • 复杂性：相比传统的阻塞 IO 模型，Reactor 模式在实现上更为复杂。
  • 调试难度：由于涉及多线程和非阻塞 IO，调试可能会更加困难。

示例代码

下面是一个简化的 Reactor 多线程模型的 Java 代码示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
public class Reactor implements Runnable {
    final Selector selector;
    final ServerSocketChannel serverSocketChannel;

    Reactor(int port) throws IOException {
        selector = Selector.open();
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        serverSocketChannel.configureBlocking(false);
        SelectionKey sk = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        sk.attach(new Acceptor());
    }

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                selector.select();
                Set<SelectionKey> selected = selector.selectedKeys();
                Iterator<SelectionKey> it = selected.iterator();
                while (it.hasNext()) {
                    dispatch(it.next());
                    it.remove();
                }
            }
        } catch (IOException ex) {
            // Handle exception
        }
    }

    void dispatch(SelectionKey k) {
        Runnable r = (Runnable) (k.attachment());
        if (r != null) {
            r.run();
        }
    }

    class Acceptor implements Runnable {
        @Override
        public void run() {
            try {
                SocketChannel c = serverSocketChannel.accept();
                if (c != null) {
                    new Handler(selector, c);
                }
            } catch (IOException ex) {
                // Handle exception
            }
        }
    }
}

在这个示例中，Reactor 类负责初始化 Selector 和 ServerSocketChannel，并在运行时处理 ACCEPT 事件。Acceptor 类负责接受新的连接请求。Handler 类（未展示）将处理 READ 和 WRITE 事件，并可能将任务委托给 Worker Pool 线程。

请注意，这只是一个非常简化的示例，实际应用中的 Reactor 模式实现会更加复杂，涉及更多的错误处理和资源管理。