深入解析Java NIO在高并发场景下的性能优化实践指南

serverSocketChannel.socket().setReuseAddress(true);
socketChannel.socket().setTcpNoDelay(true);
socketChannel.socket().setReceiveBufferSize(4 * 1024 * 1024);

public int select(long timeout) throws IOException {
    // 底层调用 epoll_wait 或者 kqueue
    int n = Impl.poll(fd, events, nevents, timeout);
    if (n > 0) {
        // 填充 readyKeys
        for (int i = 0; i < n; i++) {
            SelectionKeyImpl k = (SelectionKeyImpl) findKey(events[i]);
            k.nioReadyOps = mapReadyOps(events[i]);
            selectedKeys.add(k);
        }
    }
    return n;
}

public ByteBuffer allocateDirect(int capacity) {
    return new DirectByteBuffer(capacity);
}

// DirectByteBuffer内部维护一个Cleaner用于回收堆外内存
private static class DirectByteBuffer implements ByteBuffer {
    private final long address;
    private final int capacity;
    private final Cleaner cleaner;
    DirectByteBuffer(int cap) {
        address = unsafe.allocateMemory(cap);
        cleaner = Cleaner.create(this, new Deallocator(address));
        capacity = cap;
    }
}

public class MainReactor implements Runnable {
    private final Selector selector;
    private final ServerSocketChannel serverChannel;
    private final WorkerReactor[] workers;
    private int workerIndex = 0;

    public MainReactor(int port, int workerCount) throws IOException {
        selector = Selector.open();
        serverChannel = ServerSocketChannel.open();
        serverChannel.socket().bind(new InetSocketAddress(port));
        serverChannel.configureBlocking(false);
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);

        workers = new WorkerReactor[workerCount];
        for (int i = 0; i < workerCount; i++) {
            workers[i] = new WorkerReactor();
            new Thread(workers[i], "Worker-" + i).start();
        }
    }

    @Override
    public void run() {
        while (true) {
            selector.select();
            Iterator<SelectionKey> it = selector.selectedKeys().iterator();
            while (it.hasNext()) {
                SelectionKey key = it.next(); it.remove();
                if (key.isAcceptable()) {
                    SocketChannel client = ((ServerSocketChannel) key.channel()).accept();
                    client.configureBlocking(false);
                    // 轮询分发给Worker
                    WorkerReactor worker = workers[(workerIndex++) % workers.length];
                    worker.register(client);
                }
            }
        }
    }
    public static void main(String[] args) throws IOException {
        new Thread(new MainReactor(9090, Runtime.getRuntime().availableProcessors())).start();
        System.out.println("Echo Server started on port 9090");
    }
}

public class WorkerReactor implements Runnable {
    private Selector selector;
    private final Queue<SocketChannel> queue = new ConcurrentLinkedQueue<>();

    public WorkerReactor() throws IOException {
        selector = Selector.open();
    }

    public void register(SocketChannel channel) throws ClosedChannelException {
        queue.offer(channel);
        selector.wakeup();
    }

    @Override
    public void run() {
        while (true) {
            try {
                selector.select();
                SocketChannel client;
                while ((client = queue.poll()) != null) {
                    client.register(selector, SelectionKey.OP_READ, ByteBuffer.allocateDirect(1024));
                }
                Iterator<SelectionKey> it = selector.selectedKeys().iterator();
                while (it.hasNext()) {
                    SelectionKey key = it.next(); it.remove();
                    if (key.isReadable()) {
                        ByteBuffer buffer = (ByteBuffer) key.attachment();
                        SocketChannel ch = (SocketChannel) key.channel();
                        int len = ch.read(buffer);
                        if (len > 0) {
                            buffer.flip(); ch.write(buffer); buffer.clear();
                        } else if (len < 0) {
                            key.cancel(); ch.close();
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}