how to serve 1000s of concurrent connection using Java NIO

Question

My sender is sending 10000 requests per second (or even more) but my ServerSocketChannel is only able to read and process (in thread) 8000 requests (~appx).

Dummy code is like this:

public class NioReceiver {
    private int port = -1;
    private static String message = null;
    public void receive() throws IOException {
        // Get the selector
        Selector selector = Selector.open();
        // Selector is open for making connection
        // Get the server socket channel and register using selector
        ServerSocketChannel SS = ServerSocketChannel.open();
        InetSocketAddress hostAddress = new InetSocketAddress(this.port);
        SS.bind(hostAddress);
        SS.configureBlocking(false);
        int ops = SS.validOps();
        SelectionKey selectKy = SS.register(selector, ops, null);
        for (;;) {
            //Waiting for the select operation...
            int noOfKeys = selector.select();
            // The Number of selected keys are: noOfKeys
            Set selectedKeys = selector.selectedKeys();
            Iterator itr = selectedKeys.iterator();
            while (itr.hasNext()) {
                ByteBuffer buffer = ByteBuffer.allocate(1024 * 60);
                SelectionKey ky = (SelectionKey) itr.next();
                if (ky.isAcceptable()) {
                    // The new client connection is accepted
                    SocketChannel client = SS.accept();
                    client.configureBlocking(false);
                    // The new connection is added to a selector
                    client.register(selector, SelectionKey.OP_READ);
                    // The new connection is accepted from the client: client
                } else if (ky.isReadable()) {
                    // Data is read from the client
                    SocketChannel client = (SocketChannel) ky.channel();
                    String output = null;
                    buffer.clear();
                    int charRead = -1;
                    try {
                        charRead = client.read(buffer);
                    } catch (IOException e) {
                        continue;
                    }
                    if (charRead <= 0) {
                        // client closed
                        client.close();
                    } else {
                        output = new String(buffer.array());
                        message = output;
                        try {
                            new Thread(() -> {
                                processAndStore(message);
                            }).start();
                        } catch (Exception e) {
                            System.err.println("Thread exception:::" + e.getMessage());
                        }
                    } // else if of client.isConnected()
                } // else if of ky.isReadable()
                itr.remove();
            } // end of while loop
        } // end of for loop
    }

    public void processAndStore(String output) {
        String exchangeName = null;
        String dataLine = null;
        String Lines[] = output.split("\r\n");
        for (int i = 0; i < Lines.length; i++) {
            if (Lines[i].contains("Host: ")) {
                exchangeName = Lines[i].substring(6);
            }
            if (Lines[i].isEmpty()) {
                dataLine = Lines[i + 1];
            }
        }
        StringBuffer updatedLastLine = null;
        if (dataLine != null) {
            if (dataLine.contains("POST")) {
                updatedLastLine = new StringBuffer(dataLine.substring(0, dataLine.indexOf("POST")));
            } else {
                updatedLastLine = new StringBuffer(dataLine);
            }
            if (!dataLine.equals("")) {
                try {
                    if (updatedLastLine.lastIndexOf("}") != -1) {
                        updatedLastLine.replace(updatedLastLine.lastIndexOf("}"), updatedLastLine.lastIndexOf("}") + 1, ",\"name\":\"" + exchangeName
                                + "\"}");
                    } else {

                        return;
                    }
                } catch (StringIndexOutOfBoundsException e) {
                    System.out.println(updatedLastLine + "::" + dataLine);
                    System.out.println(e);
                }
                store(updatedLastLine.toString());
            }
        }
    }

    public NioReceiver(int port) {
        this.port = port;
    }
}

When I am removing processing logic it is able to receive more requests but not all.

how can I improve my code to receive all 10000s incoming requests.

Answer 1

Use a thread pool / message queue instead of creating 1000's of threads for calling processAndStore().

Starting a thread is expensive.

Starting 10000 threads per second? Yikes!

As @EJP said in a comment:

The purpose of NIO is to reduce the number of required threads. You don't seem to have got the message.

---

In addition to that, profile your code to see where the bottleneck is, rather than guessing.

But, here are some guesses anyway:

1. Don't use StringBuffer, use StringBuilder.
   Reason: See Difference between StringBuilder and StringBuffer.

2. Don't call lastIndexOf("}") three times.
   Reason: lastIndexOf() is a sequential search, so relatively slow. The JVM may or may not optimize the multiple calls away, but if performance is critical, don't rely on it. Do it yourself by assigning result to variable. See also Does Java optimize method calls via an interface which has a single implementor marked as final?