使用Java NIO的10000并发连接

Question

我使用Java nio编写了一个服务器（类似于这里的一个）和客户机代码。

我正在努力实现尽可能多的联系。根据之前的建议，我减缓了客户端创建过程，给操作系统（Windows 8）足够的时间来处理请求。

我在不同的机器上运行客户端代码，以便服务器拥有所有可用的运行空间。

当我尝试创建10000个连接时，大约8500个连接被连接，其余的连接被拒绝，客户端（客户端代码中的线程）的连接被拒绝的情况发生得更多，这是后来创建的（客户端代码中的循环）。

我的CPU和内存使用率非常高。我分析发现，大多数（总CPU消耗的48%）是由select方法消耗的（其余大部分由gui事件消耗）。是因为这么多客户吗？我还看到一些人抱怨JRE7中的这个bug，并建议使用JRE6。

javaw.exe进程的内存使用量为2000 MB。（我注意到1个进程使用低内存，但CPU使用量很大）。当所有8500个左右的客户端都连接时，总体使用量约为98%。系统挂起了很多次，但仍在继续服务。我看到非页池内存使用量在这个过程中从178 MB增加到310 MB（最大限制是多少？）。是因为当我们写入套接字时使用了非页池内存吗？

有谁能告诉我，我可能达到了哪些极限，所以10000个成功的连接是不可能的？（每个过程的插座限制？）（非分页内存？）（又是积压队列？）可能允许推动限制的调整？（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注册一个新接受的频道
• 当你有东西要写到频道时，就写吧
• 如果写入返回零，为OP_WRITE注册通道，保存您尝试写入的ByteBuffer，然后返回选择循环
• 当OP_WRITE在通道上触发时，使用相同的缓冲区调用write（）
• 如果写入成功并且没有返回零，请再次注册OP_READ，或者至少从interest stOps中删除OP_WRITE。

NB关闭通道会取消其键。你不需要取消。