Java NIO 中 Reactor 模式怎么实现-java教程-PHP中文网

java nio 中实现 reactor 模式的核心在于将网络 i/o 事件的处理与业务逻辑处理分离，通过一个独立的线程（reactor）负责监听和分发事件，而具体的任务则提交给工作线程池执行。具体做法是创建一个 selector 对象并注册感兴趣的 op_read 或 op_write 事件，在一个无限循环中调用 selector.select() 阻塞等待就绪事件，一旦有事件发生，遍历 selectionkey 集合并根据事件类型触发相应的回调方法，这些回调方法通常包含读取数据、解码以及将业务任务提交到 executorservice 线程池中。

Java NIO 中 Reactor 模式怎么实现

如何构建主 Reactor 线程的初始化与事件循环

构建主 Reactor 线程的关键步骤是实例化 ServerSocketChannel 并将其配置为非阻塞模式，随后创建 Selector 并将通道注册到选择器上。在事件循环中，程序会持续调用 select() 方法，该方法会阻塞直到至少有一个注册的通道准备好进行 I/O 操作。当 select() 返回大于 0 的值时，获取 selectedKeys 集合，并逐个处理其中的 SelectionKey。对于新连接请求（OP_ACCEPT），需要接受 SocketChannel 并设置其非阻塞状态，然后将其注册到同一个 Selector 上以监听读事件；对于读事件（OP_READ），则从通道中读取字节数据并封装为任务对象。这种机制确保了单个线程能够高效地管理成千上万个并发连接，而不会被某个慢速连接的 I/O 操作所阻塞。

Java NIO 中 Reactor 模式怎么实现

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

public class SimpleReactor implements Runnable {
    protected final Selector selector;
    protected final ServerSocketChannel serverSocketChannel;

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

    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                selector.select();
                Set<selectionkey> selectedKeys = selector.selectedKeys();
                Iterator<selectionkey> it = selectedKeys.iterator();
                while (it.hasNext()) {
                    SelectionKey key = it.next();
                    it.remove();
                    try {
                        handleInput(key);
                    } catch (IOException e) {
                        key.cancel();
                        key.channel().close();
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private void handleInput(SelectionKey key) throws IOException {
        if (key.isAcceptable()) {
            SocketChannel client = serverSocketChannel.accept();
            client.configureBlocking(false);
            client.register(selector, SelectionKey.OP_READ);
        } else if (key.isReadable()) {
            SocketChannel channel = (SocketChannel) key.channel();
            // 实际应用中需处理半包粘包问题
            byte[] data = new byte[1024];
            int readBytes = channel.read(data);
            if (readBytes > 0) {
                System.out.println("Received: " + new String(data, 0, readBytes));
            }
        }
    }
}</selectionkey></selectionkey>

如何将业务逻辑解耦至工作线程池

将业务逻辑解耦至工作线程池的目的是避免在 Reactor 线程中执行耗时操作，从而保证 I/O 事件的响应速度。具体做法是在接收到完整的数据后，不直接在 Reactor 线程中进行复杂的计算或数据库查询，而是创建一个实现了 Runnable 接口的任务对象，该对象持有读取到的数据和必要的上下文信息。随后，将这个任务对象提交给一个预定义的 ExecutorService 线程池。线程池中的工作线程会从任务队列中取出任务并执行，执行完毕后如果需要向客户端写回数据，可以通过 Channel.write() 方法异步写入，或者再次触发 Reactor 线程的写事件。这种多路复用与工作线程结合的方式，既利用了 NIO 的高并发能力，又充分利用了多核 CPU 的计算资源。

shopify-app-react-router

构建 Shopify 嵌入式应用（Embedded App）的开发技能：基于 @shopify/shopify-app-react-router + React Router v7 + Admin API 2026-04。覆盖 shopifyApp 配置、authenticate.admin 的 GraphQL/REST 调用、Webhook 校验、Session 存储、订阅计费、App Bridge 导航、部署，并新增：适用场景判断、安全最佳实践、错误处理与容错、最佳实践、常见问题 FAQ、高级场景。G

下载

Java NIO 中 Reactor 模式怎么实现

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class DecoupledReactor extends SimpleReactor {
    private final ExecutorService threadPool;

    public DecoupledReactor(int port, int poolSize) throws IOException {
        super(port);
        this.threadPool = Executors.newFixedThreadPool(poolSize);
    }

    @Override
    private void handleInput(SelectionKey key) throws IOException {
        if (key.isAcceptable()) {
            SocketChannel client = serverSocketChannel.accept();
            client.configureBlocking(false);
            client.register(selector, SelectionKey.OP_READ);
        } else if (key.isReadable()) {
            SocketChannel channel = (SocketChannel) key.channel();
            byte[] data = new byte[1024];
            int readBytes = channel.read(data);
            if (readBytes > 0) {
                // 提交业务处理任务到线程池
                final byte[] receivedData = data.clone();
                threadPool.submit(() -> {
                    processBusinessLogic(receivedData, channel);
                });
            }
        }
    }

    private void processBusinessLogic(byte[] data, SocketChannel channel) {
        // 模拟耗时业务处理
        String result = "Processed: " + new String(data);
        try {
            channel.write(java.nio.ByteBuffer.wrap(result.getBytes()));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

如何处理 Selector 轮询时的空转与性能优化

在处理 Selector 轮询时，如果没有任何事件发生，select() 方法会一直阻塞，这本身不是问题，但如果在高并发场景下，频繁的空转或唤醒可能导致 CPU 占用率升高。为了优化性能，可以设置 select() 的超时时间，或者使用 Java 7 引入的 Selectors.open() 默认实现中的 epoll/kqueue 后端（取决于操作系统）。此外，必须确保在每次循环结束时清理 selectedKeys 集合，否则会导致重复处理同一事件。另一个常见的优化点是将读写操作限制在 Reactor 线程中仅做数据搬运，而将编解码、业务计算完全剥离到线程池。如果某些操作必须在 Reactor 线程中完成（如直接写回小数据），应确保这些操作是非阻塞且快速的。同时，注意处理 Channel 关闭异常，及时取消 Key 并关闭 Channel，防止资源泄漏。

Java免费学习笔记：立即使用
解锁 Java 大师之旅：从入门到精通的终极指南