Java NIO Reactor 示例代码
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
public class NioReactorHttpServer {
private static final int PORT = 8080;
private static final int BUFFER_SIZE = 1024;
public static void main(String[] args) throws IOException {
// 1. 创建Selector
Selector selector = Selector.open();
// 2. 创建ServerSocketChannel
ServerSocketChannel serverChannel = ServerSocketChannel.open();
serverChannel.bind(new InetSocketAddress(PORT));
serverChannel.configureBlocking(false);
// 3. 注册Accept事件
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("NIO Reactor HTTP Server started on port " + PORT);
// 4. Reactor事件循环
while (true) {
selector.select(); // 阻塞等待事件
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> it = keys.iterator();
while (it.hasNext()) {
SelectionKey key = it.next();
it.remove();
try {
if (key.isAcceptable()) {
handleAccept(key, selector);
} else if (key.isReadable()) {
handleRead(key);
} else if (key.isWritable()) {
handleWrite(key);
}
} catch (IOException e) {
key.cancel();
key.channel().close();
e.printStackTrace();
}
}
}
}
private static void handleAccept(SelectionKey key, Selector selector) throws IOException {
ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
SocketChannel clientChannel = serverChannel.accept();
clientChannel.configureBlocking(false);
// 给客户端注册可读事件
clientChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(BUFFER_SIZE));
System.out.println("Accepted connection from " + clientChannel.getRemoteAddress());
}
private static void handleRead(SelectionKey key) throws IOException {
SocketChannel channel = (SocketChannel) key.channel();
ByteBuffer buffer = (ByteBuffer) key.attachment();
int read = channel.read(buffer);
if (read == -1) {
channel.close();
key.cancel();
System.out.println("Client disconnected.");
return;
}
String request = new String(buffer.array(), 0, buffer.position());
if (request.contains("\r\n\r\n")) { // 请求头结束
System.out.println("Received request:\n" + request);
// 简单处理HTTP请求
String body = "<html><body><h1>Hello NIO Reactor!</h1></body></html>";
byte[] bodyBytes = body.getBytes();
String response = "HTTP/1.1 200 OK\r\n" +
"Content-Length: " + bodyBytes.length + "\r\n" +
"Content-Type: text/html\r\n" +
"Connection: keep-alive\r\n" + // 支持长连接
"\r\n";
ByteBuffer responseBuffer = ByteBuffer.allocate(response.getBytes().length + bodyBytes.length);
responseBuffer.put(response.getBytes());
responseBuffer.put(bodyBytes);
responseBuffer.flip();
// 改为可写事件
key.interestOps(SelectionKey.OP_WRITE);
key.attach(responseBuffer);
}
}
private static void handleWrite(SelectionKey key) throws IOException {
SocketChannel channel = (SocketChannel) key.channel();
ByteBuffer buffer = (ByteBuffer) key.attachment();
channel.write(buffer);
if (!buffer.hasRemaining()) {
// 写完了，切换回读模式，保持长连接
buffer.clear();
key.attach(ByteBuffer.allocate(BUFFER_SIZE));
key.interestOps(SelectionKey.OP_READ);
}
}
}
核心说明
Selector + Channel
Selector 用于监听多个通道事件。
ServerSocketChannel 监听客户端连接。
SocketChannel 非阻塞模式。
事件循环
select() 阻塞等待事件。
遍历 selectedKeys() 处理事件。
非阻塞读写
读：channel.read(buffer)。
写：channel.write(buffer)。
使用 ByteBuffer 存储数据。
长连接处理
Connection: keep-alive。
写完响应后，保持通道注册 OP_READ，继续等待下一次请求。
并发高效
单线程 Reactor 可以同时处理成百上千连接。
可拓展多线程 Reactor，处理 CPU 密集型业务。
测试方法
curl -v --http1.1 http://localhost:8080
curl -v --http1.1 -H "Connection: keep-alive" http://localhost:8080
你会看到同一 TCP 连接可以发送多个请求，服务器继续响应。
来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/70050070/viewspace-3117738/ ，如需转载，请注明出处，否则将追究法律责任。
