java - 使用 Java NIO 的 10000 个并发连接

Question

我写了一个服务器(类似于 here )和 Client code使用 Java nio。

我正在尝试实现尽可能多的连接。根据之前的建议，我放慢了客户端创建过程，让操作系统 (Windows 8) 有足够的时间来处理请求。

我在不同的机器上运行了客户端代码，这样服务器就有了所有可用的运行空间。

当我尝试创建 10,000 个连接时，大约 8500 个正在连接，其余的连接被拒绝，并且客户端(客户端代码中的线程)连接被拒绝的情况更多，这是稍后创建的(客户端代码中的循环)。

我的 CPU 和内存使用率急剧上升。我发现大部分(占总 CPU 消耗的 48%)被选择方法消耗(其余主要由 gui 事件消耗)。是因为有这么多客户吗？我还看到一些人提示 JRE7 中的这个错误并建议使用 JRE6。

javaw.exe 进程的内存使用量为 2000+ MB。(我注意到 1 个进程内存不足但 CPU 使用率很高)。当所有 8500 左右时，我们的总体使用率约为 98%客户端已连接。系统也挂了很多次但继续服务。我看到非页面池内存使用量在此过程中从 178 MB 增加到 310 MB(最大限制是多少？)。是因为当我们写入套接字时非页面是否使用了池内存？

谁能告诉我可能达到了哪些限制，所以 10,000 次成功连接是不可能的？(每个进程的套接字限制？)(非分页内存？)(再次积压队列？)可能能够允许插入限制的调整？ (Windows 机器)

我在 4GB 系统上使用 Windows 8。

`

public class Server implements Runnable  {

public final static String ADDRESS = "192.168.2.14";

public final static int PORT = 8511;

public final static long TIMEOUT = 10000;

public int clients;

ByteBuffer readBuffer = ByteBuffer.allocate(1024);

private ServerSocketChannel serverChannel;

private Selector selector;

private Map<SocketChannel,byte[]> dataTracking = new HashMap<SocketChannel, byte[]>();

public Server(){
    init();
}

private void init(){
    System.out.println("initializing server");

    if (selector != null) return;
    if (serverChannel != null) return;

    try {
        selector = Selector.open();
        serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);
        serverChannel.socket().bind(new InetSocketAddress(ADDRESS, PORT));
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);
    } catch (IOException e) {
        e.printStackTrace();
    }
}

@Override
public void run() {
    System.out.println("Now accepting connections...");
    try{
        while (!Thread.currentThread().isInterrupted()){

            int ready = selector.select();
            if(ready==0)
                continue;
            Iterator<SelectionKey> keys = selector.selectedKeys().iterator();

            while (keys.hasNext()){
                SelectionKey key = keys.next();
                keys.remove();
                if (!key.isValid()){
                    continue;
                }

                if (key.isAcceptable()){
                    System.out.println("Accepting connection");
                    accept(key);
                }

                if (key.isWritable()){
                    System.out.println("Writing...");
                    write(key);
                }

                if (key.isReadable()){
                    System.out.println("Reading connection");
                    read(key);
                }
            }
        }
    } catch (IOException e){
        e.printStackTrace();
    } finally{
        closeConnection();
    }

}

private void write(SelectionKey key) throws IOException{

    SocketChannel channel = (SocketChannel) key.channel();
    byte[] data = dataTracking.get(channel);
    dataTracking.remove(channel);
    **int count = channel.write(ByteBuffer.wrap(data));
    if(count == 0)
    {
        key.interestOps(SelectionKey.OP_WRITE);
        return;
    }
    else if(count > 0)
    {
        key.interestOps(0);
        key.interestOps(SelectionKey.OP_READ);  
    }** 
}

private void closeConnection(){

    System.out.println("Closing server down");
    if (selector != null){
        try {
            selector.close();
            serverChannel.socket().close();
            serverChannel.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

private void accept(SelectionKey key) throws IOException{
    ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
    SocketChannel socketChannel = serverSocketChannel.accept();
    if(socketChannel == null)
    {
        throw new IOException();
    }
    socketChannel.configureBlocking(false);
     clients++;
    **//socketChannel.register(selector, SelectionKey.OP_WRITE|SelectionKey.OP_READ);
    SelectionKey skey = socketChannel.register(selector, SelectionKey.OP_READ);**

    byte[] hello = new String("Hello from server").getBytes();
    dataTracking.put(socketChannel, hello);
}

private void read(SelectionKey key) throws IOException{
    SocketChannel channel = (SocketChannel) key.channel();
    readBuffer.clear();
    int length;
    try {
        length = channel.read(readBuffer);
    } catch (IOException e) {
        System.out.println("Reading problem, closing connection");
        System.out.println("No of clients :"+clients);
        key.cancel();
        channel.close();
        return;
    }
    if (length == -1){
        System.out.println("Nothing was there to be read, closing connection");
        channel.close();
        key.cancel();
        return;
    }

    readBuffer.flip();
    byte[] data = new byte[1000];
    readBuffer.get(data, 0, length);
    String fromclient = new String(data,0,length,"UTF-8");
    System.out.println("Received: "+fromclient);
    String dat = fromclient+channel.getRemoteAddress();
    data= dat.getBytes();
    echo(key,data);
}

private void echo(SelectionKey key, byte[] data) throws IOException{
    SocketChannel socketChannel = (SocketChannel) key.channel();
    dataTracking.put(socketChannel, data);
    **//key.interestOps(SelectionKey.OP_WRITE);
    try
    {
        write(key);
    }
    catch(IOException e)
    {
        System.out.println("Problem in echo"+e);
        e.printStackTrace();
    }
}
public static void main(String [] args)
{
    Thread serv = new Thread(new Server());
    serv.start();
}

Answer 1

socketChannel.register(selector, SelectionKey.OP_WRITE|SelectionKey.OP_READ);

这是不正确的用法。您的选择器将旋转为 OP_WRITE几乎总是准备就绪，除了在套接字发送缓冲区已满的极少数情况下。这就是您不处理 OP_ACCEPT 的原因尽可能快。您正忙于处理 OP_WRITE有时你无事可写。

正确使用方法OP_WRITE如下:

• 为 OP_READ 注册一个新接受的 channel 只有
• 当你有什么要写到 channel 的时候，就写吧
• 如果该写入返回零，请注册 OP_WRITE 的 channel , 保存 ByteBuffer你试图写，然后返回到选择循环
• 何时OP_WRITE在 channel 上触发，请调用 write()使用相同的缓冲区
• 如果写入成功并且没有返回零，注册OP_READ再次，或至少删除 OP_WRITE来自 interestOps .

NB 关闭 channel 会取消其 key 。你不需要取消。