Java NIO非阻塞I/O实践：常见陷阱与优化策略

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;

public class MyAsyncProcessor {

    // 定义连接状态枚举
    enum States {
        Idle,
        Read,
        Write
    }

    // 线程池用于处理耗时任务
    ExecutorService pool;
    // 存储每个SocketChannel的当前状态，实际项目中推荐使用SelectionKey.attach()
    private Map<Integer, States> socketStates = new HashMap<>();

    public MyAsyncProcessor() {
    }

    // 示例任务类，实际业务逻辑在此实现
    public static class MyTask implements Runnable {
        private int secondsToRead;
        private int secondsToWrite;

        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: read time " + secondsToRead + ", write time " + secondsToWrite);
            try {
                // 模拟读取和写入操作的耗时
                Thread.sleep(secondsToRead + secondsToWrite);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            System.out.println("Task execution finished.");
        }
    }

    public static void main(String[] args) throws IOException {
        new MyAsyncProcessor().process();
    }

    public void process() throws IOException {
        // 初始化固定大小的线程池
        pool = Executors.newFixedThreadPool(2);
        InetAddress host = InetAddress.getByName("localhost");
        // 打开选择器
        Selector selector = Selector.open();
        // 打开服务器套接字通道
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        // 设置为非阻塞模式
        serverSocketChannel.configureBlocking(false);
        // 绑定到指定端口
        serverSocketChannel.bind(new InetSocketAddress(host, 9876));
        // 将服务器通道注册到选择器，监听连接接受事件
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

        System.out.println("Server started on port 9876.");

        // 主循环，监听I/O事件
        while (true) {
            // 阻塞直到至少一个通道就绪
            if (selector.select() > 0) {
                // 获取所有就绪的SelectionKey
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> i = selectedKeys.iterator();
                while (i.hasNext()) {
                    SelectionKey key = i.next();
                    i.remove(); // 移除当前键，防止重复处理

                    // 检查键是否有效
                    if (!key.isValid()) {
                        key.cancel(); // 如果无效，则取消
                        continue;
                    }

                    // 处理连接接受事件
                    if (key.isAcceptable()) {
                        SocketChannel socketChannel = serverSocketChannel.accept();
                        if (socketChannel != null) {
                            socketChannel.configureBlocking(false);
                            System.out.println("Connection accepted from: " + socketChannel.getRemoteAddress());
                            // 新连接只注册OP_READ，等待客户端发送数据
                            socketChannel.register(selector, SelectionKey.OP_READ);
                            socketStates.put(socketChannel.hashCode(), States.Idle); // 初始化状态
                        }
                    }

                    // 处理读就绪事件
                    if (key.isReadable()) {
                        SocketChannel socketChannel = (SocketChannel) key.channel();
                        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                        try {
                            int readBytes = socketChannel.read(byteBuffer);
                            if (readBytes > 0) {
                                byteBuffer.flip(); // 切换到读模式
                                String message = StandardCharsets.UTF_8.decode(byteBuffer).toString().trim();
                                System.out.println("Received message from client (" + socketChannel.getRemoteAddress() + "): " + message);

                                // 解析消息，创建任务
                                MyTask task = new MyTask();
                                String[] words = message.split(" ");
                                if (words.length >= 2) {
                                    try {
                                        int secondsToRead = Integer.parseInt(words[words.length - 2]);
                                        int secondsToWrite = Integer.parseInt(words[words.length - 1]);
                                        task.setTimeToRead(secondsToRead * 1000); // 转换为毫秒
                                        task.setTimeToWrite(secondsToWrite * 1000); // 转换为毫秒
                                    } catch (NumberFormatException e) {
                                        System.err.println("Error parsing time values: " + e.getMessage());
                                        // 默认值或错误处理
                                        task.setTimeToRead(1000);
                                        task.setTimeToWrite(1000);
                                    }
                                }

                                // 将任务提交到线程池异步执行
                                pool.execute(task);
                                socketStates.put(socketChannel.hashCode(), States.Read); // 更新状态

                                // 读操作完成后，注册OP_WRITE，准备发送响应
                                key.interestOps(SelectionKey.OP_WRITE);
                            } else if (readBytes == -1) {
                                // 客户端关闭连接
                                System.out.println("Client (" + socketChannel.getRemoteAddress() + ") closed connection.");
                                socketChannel.close();
                                key.cancel();
                                socketStates.remove(socketChannel.hashCode());
                            }
                        } catch (IOException e) {
                            System.err.println("Error reading from client (" + socketChannel.getRemoteAddress() + "): " + e.getMessage());
                            socketChannel.close();
                            key.cancel();
                            socketStates.remove(socketChannel.hashCode());
                        }
                    }

                    // 处理写就绪事件
                    if (key.isWritable()) {
                        SocketChannel socketChannel = (SocketChannel) key.channel();
                        States socketState = socketStates.get(socketChannel.hashCode());

                        // 只有在完成读操作并准备写入时才进行写入
                        if (socketState == States.Read) {
                            try {
                                String response = "Server received and processed your message. Hello from server!";
                                ByteBuffer buffer = ByteBuffer.wrap(response.getBytes(StandardCharsets.UTF_8));
                                while (buffer.hasRemaining()) { // 确保所有数据都写入
                                    socketChannel.write(buffer);
                                }
                                System.out.println("Sent response to client (" + socketChannel.getRemoteAddress() + ").");

                                socketStates.put(socketChannel.hashCode(), States.Write); // 更新状态
                                // 写操作完成后，重新注册OP_READ，等待客户端的下一个请求
                                key.interestOps(SelectionKey.OP_READ);
                            } catch (IOException e) {
                                System.