Внутренний мир: Java NIO21.03.2024 00:15

• Channel — родительский класс для всего семейства

• ReadableByteChannel — канал, читающий байты из источника данных.

• ScatteringByteChannel — канал, читающий байты из источника данных в массив буфферов.

• WritableByteChannel — канал, записывающий байты в приемник данных.

• GatheringByteChannel — канал, записывающий байты в приемник данных из массива буфферов.

• ByteChannel — канал, который может как считывать, так и записывать данные. Интерфейс, объединяющий ReadableByteChannel и WritableByteChannel.

• SeekableByteChannel — канал, запоминающий текущую позицию чтения и имеющий возможность ее изменять. Иными словами, позволяет перемещаться по источнику данных.

• AsynchronousChannel — канал, поддерживающий асинхронные операции ввода/вывода.

• AsynchronousByteChannel — канал, поддерживающий асинхронные операции чтения/записи байт.

• NetworkChannel — канал, использующий сетевой сокет.

• MulticastChannel — канал, поддерживающий многоадресную рассылку по IP протоколу.

• InterruptibleChannel — канал, который возможно асинхронно закрыть и прервать операцию.

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 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() {

            @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);
        }
    }
}

На строке 14 приложение встает в ожидании запроса на подключение.

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);
        }
    }
}

В случае, если нет подключений в состоянии ожидания, на строке 13 возвращается 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);
    }
}

На 10 строке приложение останавливается в ожидании ответа.

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);
    }
}

На строке 17 клиент ничего не читает из сокета, поскольку ответа еще нет. Операция чтения пропускается. Можно заметить, что в фоновом режиме ничего не читается, поскольку в консоль выводится пустая строка, несмотря на ожидание после вызова операции чтения.

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);
    }
}

Поскольку мы установили задержку в секунду между записью запроса и чтением ответа, к моменту вызова операции чтения канал уже имеет данные для чтения, и в консоль выведется ответ.

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 selectedKeys = selector.selectedKeys();
            Iterator 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 selectionKeys = selector.selectedKeys();
                    Iterator 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);
    }
}