Внутренний мир: Java NIO

• 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 resultBytes = new byte[readByteCount];         //Записываем считанные данные в resultBytes         //Если просто вызвать buffer.array(), то если массив больше считываемого файла,         //в конце он будет заполнен нулями         buffer.get(0, resultBytes);         var fileString = new String(resultBytes, 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);     } }

Из вышесказанного может показаться, что чтение файлов в Java NIO реализовано довольно неудобно. Однако в нем появился утилитный класс Files, который делает взаимодействие с файлами очень удобным. Вот пример, как с помощью него можно прочитать файл:

void nio_readFile_files() throws IOException, URISyntaxException {     URL fileUrl = NioTest.class.getResource(testFilePath);     var filePath = Path.of(fileUrl.toURI());     var fileString = Files.readString(path);     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);         }     } }

На строке 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);         //Ответа еще нет, канал ничего не прочтет, буффер останется пустым         int receiveReadCount = sc.read(receivedBuffer);         Thread.sleep(1_000);          var resultBytes = new byte[receiveReadCount];         buffer.get(0, resultBytes);         var responseMessage = new String(resultBytes, StandardCharsets.UTF_8);                //Консоль выведет пустую строку         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);         int receiveReadCount = sc.read(receivedBuffer);                var resultBytes = new byte[receiveReadCount];         buffer.get(0, resultBytes);         var responseMessage = new String(resultBytes, StandardCharsets.UTF_8);                //Консоль выведет ответ         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<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);     } }