TCP Server using NIO to save data from IoT clients

Question

I've built a small single threaded TCP server using NIO. This server is used by small client devices to report things like temperature, when the device has been switched on, when it switches off, and various other parameters. The server then saves those parameters to a database.

We've implemented a very simple protocol to handle the process. The clients report a string, which the server can parse as JSON. Once the string has been converted to a JSONObject it is processed by a database handler class, which saves the various parameters to various tables.

This server will eventually need to handle a lot of connections. After looking at the code with fresh eyes this morning I realised there are quite a few problems with it. What I've spotted so far is the following (tell me if I'm wrong here):

1. I handle saving the data to the database in the server main thread. This will slow down the server because it's blocking while it saves data to the database. In order to rectify this I should perhaps make a thread pool for saving to the database, and share some sort of list with the threads, possibly a Queue or LinkedList. The server can then simply push the JSONObject onto the LinkedList and notify the thread pool that there is data ready to be processed. Would this be a reasonable thing to do?
2. Right now I'm not using OP_WRITE. I'm reading the data from the channel and appending it to a string, then once I see the 'end' message in the string I simply write to the channel. This is also blocking the server. How should I implement the OP_WRITE functionality. Most examples of NIO servers that I've come across write to the channel after they have finished reading. This seems wrong.
3. Some parts of my if/else statement seem redundant. But I'm unsure how to fix it. It strikes me that once I read all the bytes and locate the 'end' message I should switch the channel to OP_WRITE and handle the writing somewhere else. But a previous attempt to implement that failed.

Once I understand how this should work properly I'm looking at rewriting the server using Netty, or perhaps vert.x, because I'm really not experienced enough to build this right now, not on the time scale that's needed anyway. If I could comfortably handle a few thousand connections I'd be quite happy. But given the blocking nature of what I've built so far that's not the case.

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;

import com.serverapp.deviceserver.data.DataDAO;
import com.serverapp.deviceserver.data.JSONKeys;

public class ServerRunnable implements Runnable {

    private static final Logger EVENT_LOGGER = Logger
            .getLogger(DeviceServer.EVENT_LOGGER_NAME);

    private static final String JSON_START_WRAPPER = "{\"DATA\" :[";
    private static final String JSON_END_WRAPPER = "]}";
    private static final String END_JSON_MSG = "End\" }";
    private int serverPort;
    @SuppressWarnings("unused")
    private int queueLength;
    private boolean testMode = false;
    DataDAO database;

    public ServerRunnable(int serverPort, int queueLength, boolean testMode) {
        this.serverPort = serverPort;
        this.queueLength = queueLength;
        this.testMode = testMode;
        database = (!testMode) ? new DataDAO() : null;
    }

    private boolean running;
    private ServerSocketChannel serverSocketChannel;
    private ByteBuffer readBuffer = ByteBuffer.allocate(1024);
    private ByteBuffer writeBuffer = ByteBuffer.allocate(2048);

    @Override
    public void run() {
        try {
            Selector selector = Selector.open();

            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            ServerSocket serverSocket = serverSocketChannel.socket();
            serverSocket.bind(new InetSocketAddress(serverPort));
            System.out.println(System.getProperty("java.version"));
            EVENT_LOGGER.log(Level.INFO,
                    "DeviceServerV0.2 Runnable- Listening for connections on port: "
                            + serverSocket.getLocalPort());
            running = true;

            @SuppressWarnings("unused")
            SelectionKey serverAcceptKey = serverSocketChannel.register(
                    selector, SelectionKey.OP_ACCEPT);

            while (running) {

                @SuppressWarnings("unused")
                int keyCount = selector.select();
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> keyIterator = selectedKeys.iterator();

                while (keyIterator.hasNext()) {

                    SelectionKey key = (SelectionKey) keyIterator.next();
                    if ((key.readyOps() & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT) {

                        ServerSocketChannel ssc = (ServerSocketChannel) key
                                .channel();
                        SocketChannel sc = ssc.accept();
                        sc.configureBlocking(false);

                        // Here we add the new connection to the selector
                        SelectionKey newKey = sc.register(selector,
                                SelectionKey.OP_READ);
                        newKey.attach(new StringBuilder(JSON_START_WRAPPER));
                        keyIterator.remove();

                    } else if ((key.readyOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ) {

                        SocketChannel sc = (SocketChannel) key.channel();

                        readBuffer.clear();
                        int read = sc.read(readBuffer);

                        if (read < 0) {

                            EVENT_LOGGER.log(Level.INFO,
                                    "Read < 0; building JSON");
                            StringBuilder stringBuilder = (StringBuilder) key
                                    .attachment();
                            stringBuilder.append(JSON_END_WRAPPER);
                            String received = stringBuilder.toString();

                            try {

                                JSONObject json = new JSONObject(received);
                                processJsonObject(json);

                            } catch (JSONException e) {

                                EVENT_LOGGER.log(
                                        Level.WARNING,
                                        "A JSONException occurred: "
                                                + e.getMessage());

                            }finally{

                                key.cancel();

                            }


                        } else if (read == 0) {

                            EVENT_LOGGER.log(Level.INFO,
                                    "Read == 0; building JSON");
                            StringBuilder stringBuilder = (StringBuilder) key
                                    .attachment();
                            stringBuilder.append(JSON_END_WRAPPER);
                            String received = stringBuilder.toString();

                            try {

                                JSONObject json = new JSONObject(received);
                                processJsonObject(json);

                            } catch (JSONException e) {

                                EVENT_LOGGER.log(
                                        Level.WARNING,
                                        "A JSONException occurred: "
                                                + e.getMessage());
                                key.cancel();
                            }

                            writeToChannel(sc);
                            // TODO: when switching to OP_WRITE do we remove the
                            // key from keyIterator? I'm not sure
                            // key.interestOps(SelectionKey.OP_WRITE);
                        } else {

                            EVENT_LOGGER.log(Level.INFO, "Read == " + read);
                            readBuffer.flip();
                            StringBuilder stringBuilder = (StringBuilder) key
                                    .attachment();

                            byte[] receivedBytes = new byte[readBuffer.limit()];
                            readBuffer.get(receivedBytes);

                            String result = new String(receivedBytes);

                            stringBuilder.append(result);
                            if (result.contains(END_JSON_MSG)) {
                                EVENT_LOGGER.log(Level.INFO,
                                        "Writing end message to "
                                                + sc.getRemoteAddress()
                                                        .toString() + "...");
                                writeToChannel(sc);
                                stringBuilder.append(JSON_END_WRAPPER);
                                String received = stringBuilder.toString();

                                try {
                                    JSONObject json = new JSONObject(received);
                                    processJsonObject(json);
                                } catch (JSONException e) {
                                    EVENT_LOGGER.log(
                                            Level.WARNING,
                                            "A JSONException occurred: "
                                                    + e.getMessage());

                                    // TODO: Removing key from keyIterator in
                                    // JSONException, is this correct?
                                    key.cancel();
                                }

                            }
                            key.cancel();

                        }

                        keyIterator.remove();

                    }

                }

            }

        } catch (Exception e) {
            EVENT_LOGGER.log(Level.SEVERE, e.getMessage());

        }

    }

    /**
     * @param sc
     * @throws UnsupportedEncodingException
     * @throws IOException
     */
    private void writeToChannel(SocketChannel sc)
            throws UnsupportedEncodingException, IOException {

        String configTest = "{ \"Event\" : \"End\" }\\r\\n";
        byte[] bytes = configTest.getBytes("US-ASCII");
        writeBuffer.clear();
        writeBuffer.put(bytes);
        writeBuffer.flip();

        while (writeBuffer.hasRemaining()) {

            int written = sc.write(writeBuffer);
            EVENT_LOGGER.log(Level.INFO, "Bytes Written: " + written);
        }

        EVENT_LOGGER.log(Level.INFO, "Finished writing to "
                + sc.getRemoteAddress().toString());
    }

    /**
     * Analyse the JSONObject and process it accordingly
     * 
     * @param json
     * @throws JSONException
     */
    private void processJsonObject(JSONObject json) throws JSONException {

        JSONArray jsonA = json.getJSONArray("DATA");
        EVENT_LOGGER.log(Level.INFO, jsonA.toString());
        // System.out.println("JSON Received: " + jsonA.toString());
        if (!testMode) {

            String deviceId = "error";
            for (int i = 0; i < jsonA.length(); i++) {

                JSONObject msg = jsonA.getJSONObject(i);

                if (msg.has(JSONKeys.MesageType.ID)) {
                    deviceId = msg.getString(JSONKeys.ParamsId.ID);

                } else if (msg.has(JSONKeys.MesageType.EVENT)) {
                    @SuppressWarnings("unused")
                    int result;

                    switch (msg.getString(JSONKeys.MesageType.EVENT)) {

                    case JSONKeys.Events.BOOT:

                        result = database.insertBootEvent(msg, deviceId);
                        break;

                    case JSONKeys.Events.CONNECT:

                        result = database.insertConnectEvent(msg, deviceId);
                        break;

                    case JSONKeys.Events.END:

                        break;

                    case JSONKeys.Events.FAILED_TRANSMISSION:

                        result = database.insertFailedTransmissionEvent(msg,
                                deviceId);
                        break;

                    case JSONKeys.Events.IDLE:

                        result = database.insertIdleEvent(msg, deviceId);
                        break;

                    }
                }

            }
        }
    }

}

Answer 1

This will need to be a little bit of a history lesson, and tutorial.....

History

In the good old days, Java server programmers would have one Server Socket that listened for connections, and for each incoming connection, would create a thread to handle it. The threads would block, waiting for IO from the socket, and that's OK, but not great:

1. Threads were expensive to create in terms of time
2. scheduling many threads was hard
3. concurrency is hard

Then, two things happened at about the same time:

1. Java introduced NIO
2. Thread handling in Java became much more efficient

The NIO concepts were fantastic because:

1. You have a single thread handling all network processing. A CPU is faster than a network connection, so, just so long as you had fast processing, a single thread is more than enough for an interface.
2. You can reduce the number of active threads from 1-per-client to be 1-per-CPU (or so).

Instead of having each socket handled by a thread, you have a small-ish pool of threads ready, and waiting, to handle the few sockets that actually do something.

At the same time (actually, shortly after) as NIO was introduced, the cost factor of creating, and maintaining many threads was significantly reduced, in Java, on the core platforms (Intel, AMD, etc.). As a result, it became effectively just as reasonable to have 1 thread per client again, even if there are thousands of clients.

The issue is now mostly confined to places where thread handling is still hard, slow, and expensive.

NIO is still relevant today where there are many, many client connections, or where the infrastructure still has expensive context-switches, or thread startup times.

Tutorial

For NIO Selectors to make sense in Java, you should follow particular pattern in your code:

1. set up your selector and the thread that manages it.
2. set up a limited thread pool (about as many threads as you have CPU's/Cores)
3. plan your protocol for communication
4. set up a state machine for each client

Now, with the above, the important part is the state machine for each client. The state machine must have a very small 'footprint' on the network side. All it does is read the channel when it is told that data is available. If the data is a complete message, it flips state to 'Processing'. When the processing is complete, it switches to 'Writing', and when writing is complete it switches back to 'Reading'.

If you set it up right, Client states will be:

1. created when you accept a socket.
2. switch to Reading if the client is a valid client (using your protocol).
3. be attached to a SelectionKey.OP_READ
4. when data is available, the handler is updated with the new data
5. if the data constitutes a full message, the state changes to Processing and:
   1. the OP_READ is deregistered.
   2. the client handler is scheduled to run on a different thread
   3. the main thread goes back to monitoring the selector
6. you identify that a client handler has completed a task (using a blocking queue, and a separate thread?)
7. you register the socket for OP_WRITE with the client handler attached.
8. you get the Write-ready notification, and copy the data on to the socket output.
9. you return the state to Reading

Review

Your code is doing far too much work in the selector thread.

you are:

1. selecting channels (fine)
2. reading socket content (fine)
3. processing JSON content (not fine)
4. building output responses (not fine)
5. outputting socket content.

In addition I have the following observations:

• you are doing too much in that single handler loop. You should at least have functions for the various routines (read, write, accept).
• you do not have a protocol for the data on the stream - how do you identify when the client is a valid client? What if someone browses to your socket with a web browser, what then? What about ssh? You should expect a client to identify itself before handling it.
• you expect all messages to be received 'in full' on each loop. What if only the first 20 bytes of the message are available when you read?
• you don't specify the character encoding for strings read from the socket. Use a specific encoding (always UTF-8?) so that if your client is running on a different host than your server, and they have different default encodings, it will always work.