Inner World: Java NIO

• Channel – parent class for the entire family

• ReadableByteChannel – a channel that reads bytes from a data source.

• ScatteringByteChannel – a channel that reads bytes from a data source into an array of buffers.

• WritableByteChannel – channel that writes bytes to the data receiver.

• GatheringByteChannel – a channel that writes bytes to the data receiver from the buffer array.

• ByteChannel – a channel that can both read and write data. Interface that connects ReadableByteChannel And WritableByteChannel.

• SeekableByteChannel – a channel that remembers the current reading position and has the ability to change it. In other words, it allows you to navigate through the data source.

• AsynchronousChannel – a channel that supports asynchronous I/O operations.

• AsynchronousByteChannel – a channel that supports asynchronous byte read/write operations.

• NetworkChannel – a channel that uses a network socket.

• MulticastChannel – a channel that supports multicasting over the IP protocol.

• InterruptibleChannel – a channel that can be closed asynchronously and interrupt the operation.

void nio_blocking_readFile() throws IOException, URISyntaxException {
    URL fileUrl = NioTest.class.getResource(testFilePath);
    var filepath = Path.of(fileUrl.toURI());
  
    try (ReadableByteChannel inputChannel = FileChannel.open(filepath)) {
        var buffer = ByteBuffer.allocate(300_000);
        int readByteCount = inputChannel.read(buffer);
        var fileString = new String(buffer.array(), StandardCharsets.UTF_8);
        System.out.println(fileString);
    }
}

void io_readFile() throws IOException {
    try (
        InputStream fileStream = NioTest.class.getResourceAsStream(testFilePath);
        var inputStream = new BufferedInputStream(fileStream)
    ) {
        byte[] fileBytes = inputStream.readAllBytes();
        var fileString = new String(fileBytes, StandardCharsets.UTF_8);
        System.out.println(fileString);
    }
}

void nio_async_readFile() throws URISyntaxException, IOException {
    URL fileUrl = NioTest.class.getResource(testFilePath);
    var path = Path.of(fileUrl.toURI());
  
    try (var inputChannel = AsynchronousFileChannel.open(path)) {
        var buffer = ByteBuffer.allocate(300_000);
        Future<Integer> futureResult = inputChannel.read(buffer, 0);
      
        while (!futureResult.isDone()) {
            System.out.println("Файл еще не загружен в буффер");
        }
      
        var fileString = new String(buffer.array(), StandardCharsets.UTF_8);
        System.out.println(fileString);
    }
}

void nio_async_readFile() throws URISyntaxException, IOException {
    URL fileUrl = NioTest.class.getResource(testFilePath);
    var path = Path.of(fileUrl.toURI());
  
    try (var inputChannel = AsynchronousFileChannel.open(path)) {
        var buffer = ByteBuffer.allocate(300_000);
        inputChannel.read(buffer, 0, buffer, new CompletionHandler<Integer, ByteBuffer>() {

            @Override
            public void completed(Integer result, ByteBuffer attachment) {
                var fileString = new String(buffer.array(), StandardCharsets.UTF_8);
                System.out.println(fileString);
            }

            @Override
            public void failed(Throwable exc, ByteBuffer attachment) {
                //do nothing
            }
        });
      
        try {
            Thread.sleep(3_000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

void nio_server_blockable() throws IOException {
    //Открытие канала. Под капотом вызывается SelectorProvider, реализация которого является платформозависимой
    var ssc = ServerSocketChannel.open();
    //Созданный канал является открытым, но не привязан к конкретному сокету. Что бы связать его с сокетом, необходимо вызвать код из следующей строки
    ssc.socket().bind(new InetSocketAddress(9999));
    //По дефолту канал является блокирующим. Что бы перевести его в неблокирующий режим, нужно в следующей строке передать false
    ssc.configureBlocking(true);
    var responseMessage = "Привет от сервера! : " + ssc.socket().getLocalSocketAddress();
    var sendBuffer = ByteBuffer.wrap(responseMessage.getBytes());
  
    while (true) {
        //Ловим соединения через вызов ssc.accept()
        //Поток блокируется до момента принятия соединения
        try (SocketChannel sc = ssc.accept()) {
            System.out.println("Принято соединение от  " + sc.socket().getRemoteSocketAddress());
            var receivedBuffer = ByteBuffer.allocate(100);
            sc.read(receivedBuffer);
            var requestMessage = new String(receivedBuffer.array());
            System.out.println(requestMessage);
          
            sendBuffer.rewind();
            sc.write(sendBuffer);
        }
    }
}

On line 14, the application stops waiting for a connection request.

void nio_server_non_blockable() throws IOException {
    var ssc = ServerSocketChannel.open();
    ssc.socket().bind(new InetSocketAddress(9999));
    //Включаем неблокирующий режим канала
    ssc.configureBlocking(false);
    var responseMessage = "Привет от сервера! : " + ssc.socket().getLocalSocketAddress();
    var sendBuffer = ByteBuffer.wrap(responseMessage.getBytes());
  
    while (true) {
        System.out.print(".");
        //Ловим соединения через вызов ssc.accept().
        //Т.к. стоит неблокирующий режим, метод accept немедленно вернет null, если нет ожидающих подключений
        try (SocketChannel sc = ssc.accept()) {
            if (sc != null) {
                System.out.println();
                System.out.println("Принято соединение от  " + sc.socket().getRemoteSocketAddress());
                var receivedBuffer = ByteBuffer.allocate(100);
                sc.read(receivedBuffer);
                var requestMessage = new String(receivedBuffer.array());
                System.out.println(requestMessage);
              
                sendBuffer.rewind();
                sc.write(sendBuffer);
            } else {
                Thread.sleep(100);
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

In case there are no pending connections, line 13 returns null.

void nio_client_blockable() throws IOException {
    try (SocketChannel sc = SocketChannel.open()) {
        sc.configureBlocking(true);
        sc.connect(new InetSocketAddress("localhost", 9999));
      
        var requestMessage = "Привет от клиента! " + LocalDateTime.now();
        ByteBuffer buffer = ByteBuffer.wrap(requestMessage.getBytes());
        sc.write(buffer);
      
        var receivedBuffer = ByteBuffer.allocate(100);
        //Приложение останавливается в ожидании ответа
        sc.read(receivedBuffer);
        var responseMessage = new String(receivedBuffer.array());
        System.out.println(responseMessage);
    }
}

On line 10, the application stops waiting for a response.

void nio_client_non_blockable() throws IOException {
    try (SocketChannel sc = SocketChannel.open()) {
        //Включаем неблокирующий режим канала
        sc.configureBlocking(false);
        sc.connect(new InetSocketAddress("localhost", 9999));
      
        while (!sc.finishConnect()) {
            System.out.println("waiting to finish connection");
        }
      
        var requestMessage = "Привет от клиента! " + LocalDateTime.now();
        ByteBuffer buffer = ByteBuffer.wrap(requestMessage.getBytes());
        sc.write(buffer);
      
        var receivedBuffer = ByteBuffer.allocate(100);
        //Ответа еще нет, канал ничего не прочтет, буффер останется пустым
        sc.read(receivedBuffer);
        Thread.sleep(1_000);
        var responseMessage = new String(receivedBuffer.array());
        //Консоль выведет пустую строку
        System.out.println(responseMessage);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}

On line 17, the client does not read anything from the socket because there is no response yet. The read operation is skipped. You may notice that nothing is being read in the background because an empty string is printed to the console despite waiting after calling the read operation.

void nio_client_non_blockable() throws IOException {
    try (SocketChannel sc = SocketChannel.open()) {
        sc.configureBlocking(false);
        sc.connect(new InetSocketAddress("localhost", 9999));

        while (!sc.finishConnect()) {
            System.out.println("waiting to finish connection");
        }

        ByteBuffer buffer = ByteBuffer.wrap(("Привет от клиента! " + LocalDateTime.now()).getBytes());
        sc.write(buffer);
        Thread.sleep(1_000);

        var receivedBuffer = ByteBuffer.allocate(100);
        sc.read(receivedBuffer);
        var responseMessage = new String(receivedBuffer.array());
        //Консоль выведет ответ
        System.out.println(responseMessage);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}

Since we set a delay of one second between writing the request and reading the response, by the time the read operation is called, the channel already has data to read, and the response will be printed to the console.

void nio_non_blockable_selector_server() throws IOException {
    try (ServerSocketChannel channel = ServerSocketChannel.open();
         //Открытие селектора. Под капотом вызывается SelectorProvider, реализация которого является платформозависимой
         Selector selector = Selector.open()) {
        channel.socket().bind(new InetSocketAddress(9999));
        channel.configureBlocking(false);
        //Регистрируем серверный канал в селекторе с интересующим типом операции - принятие подключения
        SelectionKey registeredKey = channel.register(selector, SelectionKey.OP_ACCEPT);

        while (true) {
            //Получаем количество готовых к обработке каналов.
            int numReadyChannels = selector.select();
            if (numReadyChannels == 0) {
                continue;
            }
            //Получаем готовые к обработке каналы
            Set<SelectionKey> selectedKeys = selector.selectedKeys();
            Iterator<SelectionKey> keyIterator = selectedKeys.iterator();

            //Обрабатываем каналы в соответствии с типом доступной каналу операции
            while (keyIterator.hasNext()) {
                SelectionKey key = keyIterator.next();

                if (key.isAcceptable()) {
                    //Принятие подключения серверным сокетом
                    ServerSocketChannel server = (ServerSocketChannel) key.channel();
                    SocketChannel client = server.accept();
                    if (client == null) {
                        continue;
                    }
                    client.configureBlocking(false);
                    //Регистрируем принятое подключение в селекторе с интересующим типом операции - чтение
                    client.register(selector, SelectionKey.OP_READ);
                }

                if (key.isReadable()) {
                    //Тут происходит обработка принятых подключений
                    SocketChannel client = (SocketChannel) key.channel();
                    ByteBuffer requestBuffer = ByteBuffer.allocate(100);
                    int r = client.read(requestBuffer);
                    if (r == -1) {
                        client.close();
                    } else {
                        //В этом блоке происходит обработка запроса
                        System.out.println(new String(requestBuffer.array()));
                        String responseMessage = "Привет от сервера! : " + client.socket().getLocalSocketAddress();
                        //Несмотря на то, что интересующая операция, переданная в селектор - чтение, мы все равно можем писать в сокет
                        client.write(ByteBuffer.wrap(responseMessage.getBytes()));
                    }
                }
                //Удаляем ключ после обработки. Если канал снова будет доступным, его ключ снова появится в selectedKeys
                keyIterator.remove();
            }
        }
    }
}

void nio_non_blockable_selector_server() throws IOException {
    try (ServerSocketChannel channel = ServerSocketChannel.open();
         Selector selector = Selector.open()) {
        channel.socket().bind(new InetSocketAddress(9999));
        channel.configureBlocking(false);
        SelectionKey registeredKey = channel.register(selector, SelectionKey.OP_ACCEPT);

        while (true) {
            //Обрабатываем доступные к ожиданию подключения с использованием каллбэка
            selector.select(key -> {

                if (key.isAcceptable()) {
                    try {
                        //Принятие подключения серверным сокетом
                        ServerSocketChannel server = (ServerSocketChannel) key.channel();
                        SocketChannel client = server.accept();
                        client.configureBlocking(false);
                        //Регистрируем принятое подключение в селекторе с интересующим типом операции - чтение
                        client.register(selector, SelectionKey.OP_READ);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                }

                if (key.isReadable()) {
                    try {
                        //Тут происходит обработка принятых подключений
                        SocketChannel client = (SocketChannel) key.channel();
                        ByteBuffer requestBuffer = ByteBuffer.allocate(100);
                        int r = client.read(requestBuffer);
                        if (r == -1) {
                            client.close();
                        } else {
                            //В этом блоке происходит обработка запроса
                            System.out.println(new String(requestBuffer.array()));
                            String responseMessage = "Привет от сервера! : " + client.socket().getLocalSocketAddress();
                            //Несмотря на то, что интересующая операция, переданная в селектор - чтение, мы все равно можем писать в сокет
                            client.write(ByteBuffer.wrap(responseMessage.getBytes()));
                        }
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                }
            });
        }
    }
}

void nio_clientSocket_non_blockable_selector_1() throws IOException {
    try (SocketChannel sc = SocketChannel.open();
         Selector selector = Selector.open()) {
        sc.configureBlocking(false);
        //Регистрируем канал в селекторе с интересующим типом операции - чтение
        SelectionKey registeredKey = sc.register(selector, SelectionKey.OP_READ);

        //Создаем поток, который будет опрашивать селектор и обрабатывать ответы на наши запросы
        var selectorThread = new Thread(() -> {
            while (true) {
                try {
                    int numReadyChannels = selector.select();
                    if (numReadyChannels == 0) {
                        continue;
                    }

                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> keyIterator = selectionKeys.iterator();
                    
                    while (keyIterator.hasNext()) {
                        SelectionKey key = keyIterator.next();
                        if (key.isReadable()) {
                            //Этот тот канал, который мы открыли в начале функции
                            //Мы отловили его дя чтения ответа
                            SocketChannel client = (SocketChannel) key.channel();
                            var received = ByteBuffer.allocate(100);
                            client.read(received);
                            System.out.println(new String(received.array()));
                        }
                        keyIterator.remove();
                    }
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        selectorThread.setDaemon(true);
        selectorThread.start();

        sc.connect(new InetSocketAddress("localhost", 9999));
        while (!sc.finishConnect()) {
            System.out.println("waiting to finish connection");
        }
        
        String requestMessage = "Привет от клиента! " + LocalDateTime.now();
        ByteBuffer requestBuffer = ByteBuffer.wrap(requestMessage.getBytes());
        sc.write(requestBuffer);

        Thread.sleep(2000);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}