java - 使用 Executors.newFixedThreadPool 的异步 Java NIO 需要帮助

Question

嘿，大家好，我正在开发一个服务器程序，该程序可以很好地扩展并为潜在的数千个客户提供服务。问题是，我觉得 Apache MINA 太重量级了，所以我决定不使用它，而是编写自己的客户端监听器。我从未真正在 Java 中执行过异步套接字操作(C# 使这变得更加容易，但我真的更喜欢用 Java 编写这个项目，因为除了套接字读取之外，我对它的所有内容都更熟悉)，因此尝试了解如何使用正确的线程池对我来说很难。我使用 Apache MINA 文档来了解应该如何完成工作。我有两个问题:

1. 线程池使用正确吗？ Apache MINA 的默认线程大小是 CPU 核心数 + 1，但我真的应该为我的 Core 2 Duo 使用 3 线程线程池才能接受数千个客户端吗？
2. 我知道为从客户端收到的每条消息重新分配缓冲区两次(每条消息是两个数据包，一个是 4 字节常量的 header ，另一个是 header 中指定长度的内容数据包)。有没有一种简单的方法来使用固定大小的缓冲区来检查缓冲区溢出，以便行为仍然相同，但不必不断重新分配缓冲区？

这是我启动监听器的方法:

ClientListener cl = new ClientListener(1234);
cl.init();
new Thread(cl).start();

这是ClientListener的相关代码:

private static final int THREADS = Runtime.getRuntime().availableProcessors() + 1;
private ServerSocket socket;
private ExecutorService threadPool;
private int port;

public ClientListener(int port) {
    this.port = port;
    threadPool = Executors.newFixedThreadPool(THREADS);
}

public void init() {
    try {
        socket = new ServerSocket(port);
    } catch (IOException ex) {
    }
}

public void run() {
    while (true) {
        try {
            ClientSession s = new ClientSession(socket.accept());
            threadPool.execute(s);
        } catch (IOException ex) {
        }
    }
}

ClientSession相关代码:

private Socket socket;
private byte[] buffer;
private boolean isHeader;

public ClientSession(Socket socket) {
    this.socket = socket;
    this.buffer = new byte[4];
    this.isHeader = true;
}

public void run() {
    InputStream in;
    try {
        in = socket.getInputStream();
        out = socket.getOutputStream();
    } catch (IOException ex) {
        return;
    }
    while (!socket.isClosed()) {
        try {
            int read = in.read(buffer);
            if (read == -1)
                break;
            receive(read);
        } catch (IOException ex) {
            break;
        }
    }
}

private void receive(int readBytes) {
    if (isHeader) {
        if (readBytes >= 4) {
            buffer = new byte[getPacketLength(buffer)];
            isHeader = false;
        } else {
            System.out.println("Not enough data received from client " + socket.getInetAddress() + " to decode packet.");
        }
    } else {
        if (readBytes >= buffer.length) {
            processMessage(new LittleEndianByteArrayReader(decryptData(buffer)), this);
            buffer = new byte[4];
            isHeader = true;
        } else {
            System.out.println("Not enough data received from client " + socket.getInetAddress() + " to decode packet (needed " + buffer.length + ", received " + readBytes + ").");
        }
    }
}

您不需要了解 getPacketLength、processMessage、decryptData 和 LittleEndianByteArrayReader 类的代码，但我很确定这些方法/类的用途是显而易见的。

Answer 1

阻塞IO场景中的线程数量必须根据客户端数量和每个客户端连接打开的时间来计算。每个用户的每个连接都需要一个线程。

只有三个线程，用户只需打开三个 TCP 连接而不向服务器发送任何数据，就可以简单地阻塞您的服务器，直到连接超时。