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

Question

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

我正在努力建立尽可能多的联系。 根据之前的建议，我减慢了客户端创建的过程，为操作系统(Windows 8)提供了足够的时间来处理请求。

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

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

我的 CPU 和内存使用率非常高。我分析发现大多数(占总 CPU 消耗的 48%)是由 select 方法消耗的(其余大部分是由 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，然后返回到选择循环
• 当 channel 上触发 OP_WRITE 时，使用相同的缓冲区调用 write()
• 如果写入成功并且不返回零，请再次注册 OP_READ，或者至少从 interestOps 中删除 OP_WRITE。<

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