Java NIO non-blocking read and write operation optimization and common pitfalls

This article deeply explores the common "write operation blocking" problem in Java NIO non-blocking read and write operations, and analyzes how improper `SelectionKey` management (such as incorrect use of `key.cancel()` and continuous registration of `OP_WRITE`) can cause the server to deadlock during repeated connections. The article provides optimized code examples, emphasizes the importance of dynamically adjusting the `SelectionKey` interest set, and strongly recommends using mature NIO frameworks such as Netty in production environments to avoid the complexity of native NIO.

Challenges and optimization of Java NIO non-blocking I/O operations

Java NIO (New I/O) provides an event-driven, non-blocking I/O model that implements efficient network communication through selectors and channels. However, the programming model of native NIO is relatively complex, especially when processing connection status, read and write events, and the life cycle management of SelectionKey. It is easy to introduce hard-to-find errors, causing the server to behave abnormally in certain scenarios. For example, in this example, the server is stuck in the write operation when processing the second client connection.

Problem Analysis: The root cause of the NIO server getting stuck during write operations

The original NNIO server code registers the newly accepted SocketChannel to the selector in the isAcceptable() event when handling the client connection, and pays attention to the SelectionKey.OP_READ and SelectionKey.OP_WRITE events at the same time:

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

This approach has its own potential problems. The OP_WRITE event indicates when the channel can write data. The OP_WRITE event will continue to fire as long as there is space in the send buffer. If the server has no data to write, but OP_WRITE is always being watched, the selector will continuously report the event, causing the CPU to idle or fall into an infinite loop of write events.

A more serious problem occurs in the processing logic of isWritable():

 if (key.isWritable()) {
    // ...
    Runnable h = new MyAsyncWriteThread(task);
    pool.execute(h);
    key.cancel(); // Fatal error}

After completing the write operation (or planning to perform the write operation), the code directly calls key.cancel(). The function of key.cancel() is to remove the SelectionKey from its associated selector, which means that the channel will no longer receive any event notifications (including subsequent read events), and the channel is actually "abandoned". When the first client connection is completed, its SelectionKey is canceled; when the second client tries to connect, its SocketChannel may be confused due to the previous canceled key, or its own key may be canceled after completing the write operation, so that it cannot continue to process read and write events, and ultimately the server is stuck.

In addition, using Map socketStates to manage the internal state (Idle, Read, Write) of each SocketChannel is an attempt, but when combined with SelectionKey's event-driven model, if the state transition does not match the SelectionKey's interest set (interest set) management, it can also easily lead to logic errors and state confusion.

Optimization solution: Correct SelectionKey management and event flow control

The key to solving the above problems is to accurately manage the interest set of SelectionKey, ensuring that only the events that currently need to be processed are focused, and after the event processing is completed, the interest set is updated according to the business logic. At the same time, key.cancel() should only be used when the channel is about to be closed.

Here is an example of optimized NIO server code based on the original question and answer, which demonstrates how to properly manage a SelectionKey's interest set:

 import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
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.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit; //Introduce TimeUnit

public class MyAsyncProcessor {

    // Enumeration represents the internal state of the connection, used for more complex business logic enum States {
        Idle, // Initial or waiting for reading Reading, // Reading ReadComplete, // Reading completed, ready to write Writing, // Writing WriteComplete // Writing completed}

    // Assume that MyTask contains business logic and is executed in the thread pool public static class MyTask implements Runnable {
        private int secondsToRead;
        private int secondsToWrite;
        private SocketChannel clientChannel; // Add channel reference for callback public MyTask(SocketChannel channel) {
            this.clientChannel = channel;
        }

        public void setTimeToRead(int secondsToRead) {
            this.secondsToRead = secondsToRead;
        }

        public void setTimeToWrite(int secondsToWrite) {
            this.secondsToWrite = secondsToWrite;
        }

        @Override
        public void run() {
            System.out.println("Executing task for channel: " clientChannel.hashCode()  
                    ", read delay: " secondsToRead "s, write delay: " secondsToWrite "s");
            try {
                // Simulation read operation takes timeTimeUnit.SECONDS.sleep(secondsToRead);
                System.out.println("Read task completed for channel: " clientChannel.hashCode());

                // Simulating the write operation takes timeTimeUnit.SECONDS.sleep(secondsToWrite);
                System.out.println("Write task completed for channel: " clientChannel.hashCode());

                // After the task is completed, you can consider writing the result to the channel, or re-registering OP_READ to wait for the next request // Here for demonstration purposes, it is assumed that after the task is completed, the selector can be notified to re-focus on the write event // Note: In actual NIO, the callback after the task is completed requires thread-safe operation of the Selector
                // The simple way is to put the result into a queue after the task is completed, and the main thread checks the queue in the select loop and writes // Or, as in this example, write directly in the task and re-register OP_READ (thread safety needs to be ensured)
                // Considering NIO's single-threaded model, it is usually not recommended to directly operate SelectionKey on the worker thread.
                // This is just an example. In fact, the main thread should be notified through the queue or wakeup() mechanism // Example: The task is completed and the response is ready to be written (if necessary)
                //In practice, the data should be prepared here, and then the main thread will process OP_WRITE in the next select loop.
                // To simplify the example, we do not write directly here, but assume that the task is completed and subsequent write operations can be triggered // Or, if the task execution result needs to be sent immediately, you can:
                // clientChannel.write(ByteBuffer.wrap("Task finished.".getBytes(StandardCharsets.UTF_8)));
                // Later, if the client needs to continue sending data, it can re-register OP_READ
                // if (clientChannel.isOpen()) {
                // clientChannel.register(clientChannel.selector(), SelectionKey.OP_READ);
                // }

            } catch (InterruptedException | IOException e) {
                System.err.println("Task execution error for channel " clientChannel.hashCode() ": " e.getMessage());
                try {
                    clientChannel.close();
                } catch (IOException ioException) {
                    // ignore
                }
            }
        }
    }

    private ExecutorService pool;
    // Use Map to store the current status of each SocketChannel and its corresponding business data (such as MyTask)
    private Map<socketchannel connectionstate> connectionStates = new HashMap();

    //Inner class, encapsulates the connection status and related data static class ConnectionState {
        States currentState = States.Idle;
        ByteBuffer readBuffer = ByteBuffer.allocate(1024);
        MyTask task; // Store task data related to this connection public ConnectionState(SocketChannel channel) {
            this.task = new MyTask(channel); // Each connection has its own task instance}
    }

    public MyAsyncProcessor() {
    }

    public static void main(String[] args) throws IOException {
        new MyAsyncProcessor().process();
    }

    public void process() throws IOException {
        // The thread pool is used to execute time-consuming business logic to avoid blocking the NIO main thread pool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);

        InetAddress host = InetAddress.getByName("localhost");
        Selector selector = Selector.open();
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.bind(new InetSocketAddress(host, 9876));

        //Register ServerSocketChannel and only pay attention to the OP_ACCEPT event serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

        System.out.println("Server started on port 9876.");

        while (true) {
            // Block and wait for the event to occur if (selector.select() &gt; 0) {
                Set<selectionkey> selectedKeys = selector.selectedKeys();
                Iterator<selectionkey> i = selectedKeys.iterator();

                while (i.hasNext()) {
                    SelectionKey key = i.next();
                    i.remove(); // Each time an event is processed, it must be removed from selectedKeys // Check whether the key is still valid to prevent processing of canceled or closed channels if (!key.isValid()) {
                        continue;
                    }

                    try {
                        if (key.isAcceptable()) {
                            ServerSocketChannel server = (ServerSocketChannel) key.channel();
                            SocketChannel socketChannel = server.accept();
                            socketChannel.configureBlocking(false);
                            System.out.println("Connection accepted from: " socketChannel.getRemoteAddress());

                            // New connections are registered to the selector and only focus on the OP_READ event socketChannel.register(selector, SelectionKey.OP_READ);
                            connectionStates.put(socketChannel, new ConnectionState(socketChannel));

                        } else if (key.isReadable()) {
                            SocketChannel socketChannel = (SocketChannel) key.channel();
                            ConnectionState state = connectionStates.get(socketChannel);
                            if (state == null) { // The connection may have been closed, but the event is still in the queue socketChannel.close();
                                key.cancel();
                                continue;
                            }

                            state.readBuffer.clear(); // Clear the buffer for reading int bytesRead = socketChannel.read(state.readBuffer);

                            if (bytesRead &gt; 0) {
                                state.readBuffer.flip(); // Switch to read mode String clientMessage = StandardCharsets.UTF_8.decode(state.readBuffer).toString().trim();
                                System.out.println("Received from " socketChannel.getRemoteAddress() ": " clientMessage);

                                // Parse the message and update the task data String[] words = clientMessage.split(" ");
                                if (words. length &gt;= 2) {
                                    int secondsToRead = Integer.parseInt(words[words.length - 2]);
                                    int secondsToWrite = Integer.parseInt(words[words.length - 1]);
                                    state.task.setTimeToRead(secondsToRead);
                                    state.task.setTimeToWrite(secondsToWrite);
                                } else {
                                    System.out.println("Invalid message format, using default task times.");
                                    state.task.setTimeToRead(1);
                                    state.task.setTimeToWrite(1);
                                }

                                // Submit time-consuming tasks to the thread pool pool.execute(state.task);
                                state.currentState = States.Reading; // Mark as processing task // After reading the data, cancel OP_READ, register OP_WRITE, and prepare to send a response // Note: It is assumed that a response needs to be sent immediately after the business logic processing is completed key.interestOps(SelectionKey.OP_WRITE); // Only focus on write events state.currentState = States.ReadComplete; // Status update} else if (bytesRead == -1) {
                                //The client closes the connection System.out.println("Client " socketChannel.getRemoteAddress() " disconnected.");
                                closeConnection(socketChannel, key);
                            } else {
                                // bytesRead == 0, there is no data to read at the moment, waiting for the next event System.out.println("No data to read from " socketChannel.getRemoteAddress());
                            }

                        } else if (key.isWritable()) {
                            SocketChannel socketChannel = (SocketChannel) key.channel();
                            ConnectionState state = connectionStates.get(socketChannel);
                            if (state == null) {
                                socketChannel.close();
                                key.cancel();
                                continue;
                            }

                            // Only when the business logic processing is completed and there is data to be written, the writing is actually performed if (state.currentState == States.ReadComplete) { // Confirm that the data is ready String response = "Server received and processed: " state.task.secondsToRead " " state.task.secondsToWrite "\n";
                                ByteBuffer writeBuffer = ByteBuffer.wrap(response.getBytes(StandardCharsets.UTF_8));
                                int bytesWritten = socketChannel.write(writeBuffer);

                                if (!writeBuffer.hasRemaining()) { // All data has been written System.out.println("Sent response to " socketChannel.getRemoteAddress() ": " response.trim());
                                    // After writing, if you expect the client to continue sending data, re-register OP_READ
                                    key.interestOps(SelectionKey.OP_READ);
                                    state.currentState = States.WriteComplete; //State update} else {
                                    // Partial data has been written, waiting for the next OP_WRITE event to continue writing System.out.println("Partial write, " writeBuffer.remaining() " bytes remaining.");
                                }
                            } else {
                                // The business logic is not completed, or there is no data to be written, cancel the OP_WRITE interest // Avoid continuously triggering OP_WRITE when there is no data to write
                                key.interestOps(SelectionKey.OP_READ); // Assume that we should wait for the next request from the client}
                        }
                    } catch (IOException e) {
                        System.err.println("Error processing channel " key.channel() ": " e.getMessage());
                        closeConnection((SocketChannel) key.channel(), key);
                    }
                }
            }
        }
    }

    private void closeConnection(SocketChannel channel, SelectionKey key) {
        try {
            channel.close();
            key.cancel();
            connectionStates.remove(channel);
            System.out.println("Connection closed for " channel.getRemoteAddress());
        } catch (IOException e) {
            System.err.println("Error closing channel " channel.getRemoteAddress() ": " e.getMessage());
        }
    }
}</selectionkey></selectionkey></socketchannel>

Key improvements:

1. Dynamically adjust interest sets:
   • The newly accepted SocketChannel only registers SelectionKey.OP_READ. The server only pays attention to read events when it needs to read data from the client.
   • After successfully reading the client data and submitting it to the business thread pool for processing, switch the SelectionKey's interest set from OP_READ to OP_WRITE (key.interestOps(SelectionKey.OP_WRITE)). This indicates that the server is now ready to send a response to the client.
   • When the data is successfully written to the client, if the client is expected to continue sending data, switch the interest set back to OP_READ. If the session ends, the connection can be closed.
2. Avoid key.cancel() abuse: key.cancel() is only called when the channel is closed to ensure that the channel life cycle is managed correctly.
3. Connection state management: Use Map to store the more detailed state of each connection (such as States enumeration), as well as the ByteBuffer and MyTask instances related to the connection. This makes the context of each connection independent and easy to manage.
4. Asynchronous business logic: Submit time-consuming business logic (such as MyTask's run() method) to an independent thread pool for execution to avoid blocking the NIO main thread, thereby ensuring that the selector can continue to process I/O events of other connections.
5. Complete read and write handling: Ensure ByteBuffer flip() and clear() correctly between read and write operations. Handle the situation where read() returns -1 (client closed).
6. key.isValid() check: Before handling any events, check whether the SelectionKey is still valid to avoid operating on canceled keys.

Client code (no need to modify, but understand the behavior):

The client code remains the same, it sends a message to the server containing the read and write times.

 import java.io.IOException;
import java.io.PrintWriter;
import java.net.Socket;
import java.util.Random;

public class MyClient {

    public static void main(String [] args) {

        Random rand = new Random();
        int secondsToRead = rand.nextInt(5) 1; // 1-5 seconds int secondsToWrite = rand.nextInt(5) 1; // 1-5 seconds String message = "Seconds for the task to be read and written: " secondsToRead " " secondsToWrite;
        System.out.println("Client sending message: " message);
        Socket socket = null;
        try {
            socket = new Socket("127.0.0.1", 9876);
            PrintWriter printWriter = new PrintWriter(socket.getOutputStream(), true);
            printWriter.println(message);
            System.out.println("Message sent. Waiting for response...");

            //Read the server response java.io.BufferedReader reader = new java.io.BufferedReader(new java.io.InputStreamReader(socket.getInputStream()));
            String serverResponse = reader.readLine();
            System.out.println("Server response: " serverResponse);

        } catch (IOException e) {
            System.out.println("Error in Socket: " e.getMessage());
            System.exit(-1);
        } finally {
            if (socket != null) {
                try {
                    socket.close();
                } catch (IOException e) {
                    // ignore
                }
            }
        }
    }
}

Supplementary explanation for the client code: In order to better demonstrate the server's response, the client code adds logic to read the server's response so that it can receive the "Server received and processed..." message sent by the server.

Notes and Summary

1. Complexity of native NIO: Although Java native NIO provides high-performance non-blocking I/O capabilities, its API design is relatively low-level, requiring developers to have an in-depth understanding of the I/O event loop, SelectionKey management, ByteBuffer operations, and threading models. Negligence in any aspect may lead to performance problems, deadlocks or connection abnormalities.
2. Features of OP_WRITE: The OP_WRITE event will always be triggered as long as there is space in the channel's send buffer. Therefore, you should only pay attention to OP_WRITE when there is actually data that needs to be written. Once the data writing is completed, OP_WRITE should be immediately canceled or switched to other events (such as OP_READ) to avoid unnecessary event triggering and CPU overhead.
3. Recommended to use NIO framework: For production

The above is the detailed content of Java NIO non-blocking read and write operation optimization and common pitfalls. For more information, please follow other related articles on the PHP Chinese website!