本文深入探讨java nio非阻塞读写操作中常见的“写操作阻塞”问题，分析了不当的`selectionkey`管理（如错误地使用`key.cancel()`和持续注册`op_write`）如何导致服务器在重复连接时陷入僵局。文章提供了优化后的代码示例，强调了动态调整`selectionkey`兴趣集的重要性，并强烈建议在生产环境中使用netty等成熟的nio框架以规避原生nio的复杂性。
Java NIO（New I/O）提供了一种基于事件驱动、非阻塞I/O模型，它通过选择器（Selector）和通道（Channel）实现高效的网络通信。然而，原生NIO的编程模型较为复杂，尤其是在处理连接状态、读写事件以及SelectionKey的生命周期管理时，极易引入难以发现的错误，导致服务器在特定场景下表现异常，例如本例中服务器在处理第二个客户端连接时卡死在写操作环节。
原始NNIO服务器代码在处理客户端连接时，在isAcceptable()事件中将新接受的SocketChannel注册到选择器，并同时关注SelectionKey.OP_READ和SelectionKey.OP_WRITE事件：
socketChannel.register(selector, SelectionKey.OP_READ + SelectionKey.OP_WRITE);
这种做法本身就存在潜在问题。OP_WRITE事件表示通道何时可以写入数据，只要发送缓冲区有空间，OP_WRITE事件就会持续触发。如果服务器没有数据需要写入，但OP_WRITE一直被关注，那么选择器将不断报告该事件，导致CPU空转或陷入写事件的无限循环。
更严重的问题出现在isWritable()的处理逻辑中：
立即学习“Java免费学习笔记（深入）”；
if (key.isWritable()) {
// ...
Runnable h = new MyAsyncWriteThread(task);
pool.execute(h);
key.cancel(); // 致命错误
}在完成写操作（或计划执行写操作）后，代码直接调用了key.cancel()。key.cancel()的作用是将该SelectionKey从其关联的选择器中移除，这意味着该通道将不再接收任何事件通知（包括后续的读事件），并且该通道实际上被“遗弃”了。当第一个客户端连接完成后，其SelectionKey被取消；当第二个客户端尝试连接时，其SocketChannel可能因为前一个被取消的键导致的状态混乱，或者其自身的键在完成写操作后也被取消，从而无法继续处理读写事件，最终表现为服务器卡死。
此外，使用Map<Integer, States> socketStates来管理每个SocketChannel的内部状态（Idle, Read, Write）虽然是一种尝试，但在与SelectionKey的事件驱动模型结合时，如果状态转换和SelectionKey的兴趣集（interest set）管理不匹配，同样容易导致逻辑错误和状态混乱。
解决上述问题的关键在于精确地管理SelectionKey的兴趣集，确保只关注当前需要处理的事件，并在事件处理完成后，根据业务逻辑更新兴趣集。同时，key.cancel()应仅在通道即将关闭时使用。
以下是基于原始问题和答案进行优化的NIO服务器代码示例，它演示了如何正确地管理SelectionKey的兴趣集：
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit; // 引入TimeUnit
public class MyAsyncProcessor {
// 枚举表示连接的内部状态，用于更复杂的业务逻辑
enum States {
Idle, // 初始或等待读
Reading, // 正在读
ReadComplete, // 读完成，准备写
Writing, // 正在写
WriteComplete // 写完成
}
// 假设MyTask包含业务逻辑，并在线程池中执行
public static class MyTask implements Runnable {
private int secondsToRead;
private int secondsToWrite;
private SocketChannel clientChannel; // 添加通道引用以便回调
public MyTask(SocketChannel channel) {
this.clientChannel = channel;
}
public void setTimeToRead(int secondsToRead) {
this.secondsToRead = secondsToRead;
}
public void setTimeToWrite(int secondsToWrite) {
this.secondsToWrite = secondsToWrite;
}
@Override
public void run() {
System.out.println("Executing task for channel: " + clientChannel.hashCode() +
", read delay: " + secondsToRead + "s, write delay: " + secondsToWrite + "s");
try {
// 模拟读操作耗时
TimeUnit.SECONDS.sleep(secondsToRead);
System.out.println("Read task completed for channel: " + clientChannel.hashCode());
// 模拟写操作耗时
TimeUnit.SECONDS.sleep(secondsToWrite);
System.out.println("Write task completed for channel: " + clientChannel.hashCode());
// 任务完成后，可以考虑将结果写入通道，或者重新注册OP_READ等待下一个请求
// 这里为了演示，假设任务执行完毕后，可以通知选择器重新关注写事件
// 注意：在实际NIO中，任务完成后的回调需要线程安全地操作Selector
// 简单的做法是，任务完成后，将结果放入一个队列，主线程在select循环中检查队列并写入
// 或者，如本例，直接在任务中写入并重新注册OP_READ (需要确保线程安全)
// 考虑到NIO的单线程模型，通常不建议在工作线程直接操作SelectionKey
// 这里仅作示例，实际应通过队列或wakeup()机制通知主线程
// 示例：任务完成，准备写入响应（如果需要）
// 实际中，这里应该将数据准备好，然后由主线程在下一次select循环中处理OP_WRITE
// 为简化示例，这里不直接写入，而是假设任务完成，可以触发后续写操作
// 或者，如果任务执行结果需要立即发送，可以：
// clientChannel.write(ByteBuffer.wrap("Task finished.".getBytes(StandardCharsets.UTF_8)));
// 之后，如果客户端需要继续发送数据，可以重新注册OP_READ
// if (clientChannel.isOpen()) {
// clientChannel.register(clientChannel.selector(), SelectionKey.OP_READ);
// }
} catch (InterruptedException | IOException e) {
System.err.println("Task execution error for channel " + clientChannel.hashCode() + ": " + e.getMessage());
try {
clientChannel.close();
} catch (IOException ioException) {
// ignore
}
}
}
}
private ExecutorService pool;
// 使用Map来存储每个SocketChannel的当前状态，以及其对应的业务数据（如MyTask）
private Map<SocketChannel, ConnectionState> connectionStates = new HashMap<>();
// 内部类，封装连接的状态和相关数据
static class ConnectionState {
States currentState = States.Idle;
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
MyTask task; // 存储与此连接相关的任务数据
public ConnectionState(SocketChannel channel) {
this.task = new MyTask(channel); // 每个连接有自己的任务实例
}
}
public MyAsyncProcessor() {
}
public static void main(String[] args) throws IOException {
new MyAsyncProcessor().process();
}
public void process() throws IOException {
// 线程池用于执行耗时业务逻辑，避免阻塞NIO主线程
pool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
InetAddress host = InetAddress.getByName("localhost");
Selector selector = Selector.open();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.bind(new InetSocketAddress(host, 9876));
// 注册ServerSocketChannel，只关注OP_ACCEPT事件
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("Server started on port 9876.");
while (true) {
// 阻塞等待事件发生
if (selector.select() > 0) {
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> i = selectedKeys.iterator();
while (i.hasNext()) {
SelectionKey key = i.next();
i.remove(); // 每次处理完一个事件后，必须将其从selectedKeys中移除
// 检查键是否仍然有效，防止处理已取消或已关闭的通道
if (!key.isValid()) {
continue;
}
try {
if (key.isAcceptable()) {
ServerSocketChannel server = (ServerSocketChannel) key.channel();
SocketChannel socketChannel = server.accept();
socketChannel.configureBlocking(false);
System.out.println("Connection accepted from: " + socketChannel.getRemoteAddress());
// 新连接注册到选择器，只关注OP_READ事件
socketChannel.register(selector, SelectionKey.OP_READ);
connectionStates.put(socketChannel, new ConnectionState(socketChannel));
} else if (key.isReadable()) {
SocketChannel socketChannel = (SocketChannel) key.channel();
ConnectionState state = connectionStates.get(socketChannel);
if (state == null) { // 可能连接已关闭，但事件仍在队列中
socketChannel.close();
key.cancel();
continue;
}
state.readBuffer.clear(); // 清空缓冲区以备读取
int bytesRead = socketChannel.read(state.readBuffer);
if (bytesRead > 0) {
state.readBuffer.flip(); // 切换到读模式
String clientMessage = StandardCharsets.UTF_8.decode(state.readBuffer).toString().trim();
System.out.println("Received from " + socketChannel.getRemoteAddress() + ": " + clientMessage);
// 解析消息，更新任务数据
String[] words = clientMessage.split(" ");
if (words.length >= 2) {
int secondsToRead = Integer.parseInt(words[words.length - 2]);
int secondsToWrite = Integer.parseInt(words[words.length - 1]);
state.task.setTimeToRead(secondsToRead);
state.task.setTimeToWrite(secondsToWrite);
} else {
System.out.println("Invalid message format, using default task times.");
state.task.setTimeToRead(1);
state.task.setTimeToWrite(1);
}
// 将耗时任务提交到线程池
pool.execute(state.task);
state.currentState = States.Reading; // 标记为正在处理任务
// 读完数据后，取消OP_READ，注册OP_WRITE，准备发送响应
// 注意：这里假设业务逻辑处理完成后需要立即发送响应
key.interestOps(SelectionKey.OP_WRITE); // 仅关注写事件
state.currentState = States.ReadComplete; // 状态更新
} else if (bytesRead == -1) {
// 客户端关闭连接
System.out.println("Client " + socketChannel.getRemoteAddress() + " disconnected.");
closeConnection(socketChannel, key);
} else {
// bytesRead == 0, 暂时没有数据可读，等待下次事件
System.out.println("No data to read from " + socketChannel.getRemoteAddress());
}
} else if (key.isWritable()) {
SocketChannel socketChannel = (SocketChannel) key.channel();
ConnectionState state = connectionStates.get(socketChannel);
if (state == null) {
socketChannel.close();
key.cancel();
continue;
}
// 只有当业务逻辑处理完成，并且有数据需要写入时才真正执行写入
if (state.currentState == States.ReadComplete) { // 确认数据已准备好
String response = "Server received and processed: " + state.task.secondsToRead + " " + state.task.secondsToWrite + "\n";
ByteBuffer writeBuffer = ByteBuffer.wrap(response.getBytes(StandardCharsets.UTF_8));
int bytesWritten = socketChannel.write(writeBuffer);
if (!writeBuffer.hasRemaining()) { // 所有数据都已写入
System.out.println("Sent response to " + socketChannel.getRemoteAddress() + ": " + response.trim());
// 写完后，如果期望客户端继续发送数据，则重新注册OP_READ
key.interestOps(SelectionKey.OP_READ);
state.currentState = States.WriteComplete; // 状态更新
} else {
// 部分数据已写入，等待下一次OP_WRITE事件继续写入
System.out.println("Partial write, " + writeBuffer.remaining() + " bytes remaining.");
}
} else {
// 业务逻辑未完成，或者没有数据需要写入，取消OP_WRITE兴趣
// 避免在没有数据可写时持续触发OP_WRITE
key.interestOps(SelectionKey.OP_READ); // 假设此时应该等待客户端的下一个请求
}
}
} catch (IOException e) {
System.err.println("Error processing channel " + key.channel() + ": " + e.getMessage());
closeConnection((SocketChannel) key.channel(), key);
}
}
}
}
}
private void closeConnection(SocketChannel channel, SelectionKey key) {
try {
channel.close();
key.cancel();
connectionStates.remove(channel);
System.out.println("Connection closed for " + channel.getRemoteAddress());
} catch (IOException e) {
System.err.println("Error closing channel " + channel.getRemoteAddress() + ": " + e.getMessage());
}
}
}关键改进点：
客户端代码保持不变，它发送一个包含读写时间的消息到服务器。
import java.io.IOException;
import java.io.PrintWriter;
import java.net.Socket;
import java.util.Random;
public class MyClient {
public static void main(String [] args) {
Random rand = new Random();
int secondsToRead = rand.nextInt(5) + 1; // 1-5秒
int secondsToWrite = rand.nextInt(5) + 1; // 1-5秒
String message = "Seconds for the task to be read and written: " + secondsToRead + " " + secondsToWrite;
System.out.println("Client sending message: " + message);
Socket socket = null;
try {
socket = new Socket("127.0.0.1", 9876);
PrintWriter printWriter = new PrintWriter(socket.getOutputStream(), true);
printWriter.println(message);
System.out.println("Message sent. Waiting for response...");
// 读取服务器响应
java.io.BufferedReader reader = new java.io.BufferedReader(new java.io.InputStreamReader(socket.getInputStream()));
String serverResponse = reader.readLine();
System.out.println("Server response: " + serverResponse);
} catch (IOException e) {
System.out.println("Error in Socket: " + e.getMessage());
System.exit(-1);
} finally {
if (socket != null) {
try {
socket.close();
} catch (IOException e) {
// ignore
}
}
}
}
}客户端代码补充说明： 为了更好地演示服务器的响应，客户端代码增加了读取服务器响应的逻辑，使其能接收到服务器发送的“Server received and processed...”消息。
以上就是Java NIO非阻塞读写操作优化与常见陷阱的详细内容，更多请关注php中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
广告Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
323
收藏
