java nio 中实现 reactor 模式的核心在于将网络 i/o 事件的处理与业务逻辑处理分离，通过一个独立的线程（reactor）负责监听和分发事件，而具体的任务则提交给工作线程池执行。具体做法是创建一个 selector 对象并注册感兴趣的 op_read 或 op_write 事件，在一个无限循环中调用 selector.select() 阻塞等待就绪事件，一旦有事件发生，遍历 selectionkey 集合并根据事件类型触发相应的回调方法，这些回调方法通常包含读取数据、解码以及将业务任务提交到 executorservice 线程池中。
如何构建主 Reactor 线程的初始化与事件循环
构建主 Reactor 线程的关键步骤是实例化 ServerSocketChannel 并将其配置为非阻塞模式，随后创建 Selector 并将通道注册到选择器上。在事件循环中，程序会持续调用 select() 方法，该方法会阻塞直到至少有一个注册的通道准备好进行 I/O 操作。当 select() 返回大于 0 的值时，获取 selectedKeys 集合，并逐个处理其中的 SelectionKey。对于新连接请求（OP_ACCEPT），需要接受 SocketChannel 并设置其非阻塞状态，然后将其注册到同一个 Selector 上以监听读事件；对于读事件（OP_READ），则从通道中读取字节数据并封装为任务对象。这种机制确保了单个线程能够高效地管理成千上万个并发连接，而不会被某个慢速连接的 I/O 操作所阻塞。
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));
}
}
}
}
如何将业务逻辑解耦至工作线程池
将业务逻辑解耦至工作线程池的目的是避免在 Reactor 线程中执行耗时操作，从而保证 I/O 事件的响应速度。具体做法是在接收到完整的数据后，不直接在 Reactor 线程中进行复杂的计算或数据库查询，而是创建一个实现了 Runnable 接口的任务对象，该对象持有读取到的数据和必要的上下文信息。随后，将这个任务对象提交给一个预定义的 ExecutorService 线程池。线程池中的工作线程会从任务队列中取出任务并执行，执行完毕后如果需要向客户端写回数据，可以通过 Channel.write() 方法异步写入，或者再次触发 Reactor 线程的写事件。这种多路复用与工作线程结合的方式，既利用了 NIO 的高并发能力，又充分利用了多核 CPU 的计算资源。
构建 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免费学习笔记（深入）”；
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，防止资源泄漏。
