Java NIO非阻塞I/O服务器开发：常见陷阱与最佳实践

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;

public class ChannelContext {
    private ByteBuffer readBuffer = ByteBuffer.allocate(1024);
    private ByteBuffer writeBuffer = null; // 待写入的数据
    private MyTask currentTask; // 与此通道关联的业务任务

    public ByteBuffer getReadBuffer() {
        return readBuffer;
    }

    public ByteBuffer getWriteBuffer() {
        return writeBuffer;
    }

    public void setWriteBuffer(String data) {
        this.writeBuffer = ByteBuffer.wrap(data.getBytes(StandardCharsets.UTF_8));
    }

    public boolean hasDataToWrite() {
        return writeBuffer != null && writeBuffer.hasRemaining();
    }

    public MyTask getCurrentTask() {
        return currentTask;
    }

    public void setCurrentTask(MyTask currentTask) {
        this.currentTask = currentTask;
    }

    // 清理缓冲区和任务，为下一个请求做准备
    public void reset() {
        readBuffer.clear();
        writeBuffer = null;
        currentTask = null;
    }
}

public class MyTask {
    private int secondsToRead;
    private int secondsToWrite;
    private String clientMessage; // 存储客户端发送的原始消息

    public int getSecondsToRead() {
        return secondsToRead;
    }

    public void setSecondsToRead(int secondsToRead) {
        this.secondsToRead = secondsToRead;
    }

    public int getSecondsToWrite() {
        return secondsToWrite;
    }

    public void setSecondsToWrite(int secondsToWrite) {
        this.secondsToWrite = secondsToWrite;
    }

    public String getClientMessage() {
        return clientMessage;
    }

    public void setClientMessage(String clientMessage) {
        this.clientMessage = clientMessage;
    }

    @Override
    public String toString() {
        return "MyTask{" +
               "secondsToRead=" + secondsToRead +
               ", secondsToWrite=" + secondsToWrite +
               ", clientMessage='" + clientMessage + '\'' +
               '}';
    }
}

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.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class MyAsyncProcessor {

    private ExecutorService pool;
    private Selector selector;

    public MyAsyncProcessor() throws IOException {
        pool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2); // 根据CPU核心数设置线程池大小
        selector = Selector.open();
    }

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

    public void process() throws IOException {
        InetAddress host = InetAddress.getByName("localhost");
        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...");

        while (!Thread.currentThread().isInterrupted()) {
            try {
                // select()方法会阻塞，直到至少一个注册的事件发生
                if (selector.select() == 0) {
                    continue;
                }

                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> i = selectedKeys.iterator();

                while (i.hasNext()) {
                    SelectionKey key = i.next();
                    i.remove(); // 处理完一个键后必须移除

                    if (!key.isValid()) {
                        continue; // 键可能在处理过程中失效
                    }

                    try {
                        if (key.isAcceptable()) {
                            handleAccept(key);
                        }
                        if (key.isReadable()) {
                            handleRead(key);
                        }
                        if (key.isWritable()) {
                            handleWrite(key);
                        }
                    } catch (IOException e) {
                        System.err.println("Error handling channel: " + e.getMessage());
                        key.cancel(); // 发生I/O错误时取消键并关闭通道
                        key.channel().close();
                    }
                }
            } catch (Exception e) {
                System.err.println("Selector loop error: " + e.getMessage());
            }
        }
        pool.shutdown();
        selector.close();
        serverSocketChannel.close();
    }

    private void handleAccept(SelectionKey key) throws IOException {
        ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
        SocketChannel clientChannel = serverChannel.accept();
        clientChannel.configureBlocking(false);
        // 注册OP_READ事件，并附加一个ChannelContext来存储通道状态
        clientChannel.register(selector, SelectionKey.OP_READ, new ChannelContext());
        System.out.println("Connection accepted from: " + clientChannel.getRemoteAddress());
    }

    private void handleRead(SelectionKey key) throws IOException {
        SocketChannel clientChannel = (SocketChannel) key.channel();
        ChannelContext context = (ChannelContext) key.attachment();
        ByteBuffer buffer = context.getReadBuffer();
        int bytesRead;

        try {
            bytesRead = clientChannel.read(buffer);
        } catch (IOException e) {
            // 客户端强制关闭连接
            System.out.println("Client disconnected unexpectedly: " + clientChannel.getRemoteAddress());
            key.cancel();
            clientChannel.close();
            return;
        }

        if (bytesRead == -1) {
            // 客户端正常关闭连接
            System.out.println("Client closed connection: " + clientChannel.getRemoteAddress());
            key.cancel();
            clientChannel.close();
            return;
        }

        if (bytesRead > 0) {
            buffer.flip(); // 切换到读模式
            String clientMessage = StandardCharsets.UTF_8.decode(buffer).toString().trim();
            System.out.println("Received from " + clientChannel.getRemoteAddress() + ": " + clientMessage);

            // 解析消息并创建MyTask
            MyTask task = new MyTask();
            task.setClientMessage(clientMessage);
            try {
                String[] words = clientMessage.split(" ");
                // 假设消息格式稳定，获取倒数第二个和倒数第一个数字
                int secondsToRead = Integer.parseInt(words[words.length - 2]);
                int secondsToWrite = Integer.parseInt(words[words.length - 1]);
                task.setSecondsToRead(secondsToRead * 1000); // 转换为毫秒