Продвинутый ввод-вывод в Java: NIO, NIO.2 и асинхронный I/O

Selector selector = Selector.open();
ServerSocketChannel serverSocket = ServerSocketChannel.open();
serverSocket.configureBlocking(false);
serverSocket.register(selector, SelectionKey.OP_ACCEPT);
 
while (true) {
    selector.select();
    // Обработка готовых операций...
}

public class BlockingServer {
    public static void main(String[] args) throws IOException {
        ServerSocket serverSocket = new ServerSocket(8080);
        while (true) {
            Socket clientSocket = serverSocket.accept(); // Блокирующий вызов
            
            // Создаём новый поток для каждого клиента
            new Thread(() -> {
                try (InputStream in = clientSocket.getInputStream();
                     OutputStream out = clientSocket.getOutputStream()) {
                     
                    byte[] buffer = new byte[1024];
                    int bytesRead = in.read(buffer); // Ещё один блокирующий вызов
                    
                    // Обработка данных и ответ клиенту
                    // ...
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

try (FileInputStream fis = new FileInputStream("huge_file.dat");
     FileOutputStream fos = new FileOutputStream("copy.dat")) {
    byte[] buffer = new byte[8192];
    int len;
    while ((len = fis.read(buffer)) > 0) {
        fos.write(buffer, 0, len);
    }
}

// Типичный антипаттерн: создание потока для каждого клиента
ExecutorService executor = Executors.newCachedThreadPool(); // Неограниченный пул!
 
while (true) {
    final Socket client = serverSocket.accept();
    executor.submit(() -> {
        try {
            // Поток блокируется здесь, просто ожидая данные
            byte[] request = readRequest(client.getInputStream());
            // Обработка и ответ
            sendResponse(client.getOutputStream(), processRequest(request));
        } catch (IOException e) {
            // Обработка ошибок
        }
    });
}

// Создание буфера на 1024 байта
ByteBuffer buffer = ByteBuffer.allocate(1024);
 
// Запись данных в буфер
buffer.put((byte) 'H');
buffer.put((byte) 'e');
buffer.put((byte) 'l');
buffer.put((byte) 'l');
buffer.put((byte) 'o');
 
// Переключение режима буфера с записи на чтение
buffer.flip();
 
// Чтение данных из буфера
while (buffer.hasRemaining()) {
    System.out.print((char) buffer.get());
}
// Очистка буфера для повторного использования
buffer.clear();

// Открытие файлового канала для чтения
try (FileChannel channel = FileChannel.open(Paths.get("data.txt"), StandardOpenOption.READ)) {
    ByteBuffer buffer = ByteBuffer.allocate(1024);
    
    // Чтение данных из канала в буфер
    int bytesRead = channel.read(buffer);
    
    // Подготовка буфера к чтению данных
    buffer.flip();
    
    // Обработка прочитанных данных
    // ...
}

// Создание прямого буфера на 10 МБ
ByteBuffer directBuffer = ByteBuffer.allocateDirect(10 * 1024 * 1024);

// Создание селектора и регистрация канала
Selector selector = Selector.open();
ServerSocketChannel serverChannel = ServerSocketChannel.open();
serverChannel.configureBlocking(false); // Важно: канал должен быть неблокирующим
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
 
while (true) {
    // Блокирующий вызов, но ждёт любой активности на любом из зарегистрированных каналов
    selector.select();
    
    // Получаем набор ключей, представляющих каналы, готовые к операциям
    Set<SelectionKey> selectedKeys = selector.selectedKeys();
    Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
    
    while (keyIterator.hasNext()) {
        SelectionKey key = keyIterator.next();
        
        if (key.isAcceptable()) {
            // Новое соединение готово к принятию
            handleAccept(key);
        } else if (key.isReadable()) {
            // Данные готовы для чтения
            handleRead(key);
        }
        
        keyIterator.remove(); // Важно удалить ключ из набора обработанных
    }
}

buffer.compact(); // Перемещает непрочитанные данные в начало буфера

private ByteBuffer ensureCapacity(ByteBuffer buffer, int minCapacity) {
    if (buffer.capacity() >= minCapacity) {
        return buffer;
    }
    
    // Создаём новый буфер с увеличенной ёмкостью
    ByteBuffer newBuffer = ByteBuffer.allocate(Math.max(buffer.capacity() * 2, minCapacity));
    buffer.flip();
    newBuffer.put(buffer);
    return newBuffer;
}

ByteBuffer header = ByteBuffer.allocate(128);
ByteBuffer body = ByteBuffer.allocate(1024);
// Заполняем буферы...
 
// Gather-запись: записываем данные из нескольких буферов за одну операцию
WritableByteChannel channel = Channels.newChannel(outputStream);
ByteBuffer[] buffers = { header, body };
channel.write(buffers);

Selector selector = Selector.open();
ServerSocketChannel serverSocket = ServerSocketChannel.open();
serverSocket.bind(new InetSocketAddress("localhost", 5454));
serverSocket.configureBlocking(false);
serverSocket.register(selector, SelectionKey.OP_ACCEPT);
 
ByteBuffer buffer = ByteBuffer.allocate(256);
 
while (true) {
    selector.select();
    Set<SelectionKey> selectedKeys = selector.selectedKeys();
    Iterator<SelectionKey> iter = selectedKeys.iterator();
    
    while (iter.hasNext()) {
        SelectionKey key = iter.next();
        
        if (key.isAcceptable()) {
            ServerSocketChannel server = (ServerSocketChannel) key.channel();
            SocketChannel client = server.accept();
            client.configureBlocking(false);
            client.register(selector, SelectionKey.OP_READ);
        } else if (key.isReadable()) {
            SocketChannel client = (SocketChannel) key.channel();
            buffer.clear();
            int bytesRead = client.read(buffer);
            
            if (bytesRead == -1) {
                // Соединение закрыто клиентом
                client.close();
            } else {
                buffer.flip();
                client.write(buffer);
            }
        }
        iter.remove();
    }
}

// Копирование файла с максимальной эффективностью
try (FileChannel sourceChannel = FileChannel.open(Paths.get("source.dat"), StandardOpenOption.READ);
     FileChannel destChannel = FileChannel.open(Paths.get("dest.dat"), StandardOpenOption.CREATE, StandardOpenOption.WRITE)) {
    
    sourceChannel.transferTo(0, sourceChannel.size(), destChannel);
    // Альтернативно: destChannel.transferFrom(sourceChannel, 0, sourceChannel.size());
}

RandomAccessFile raf = new RandomAccessFile("bigdata.bin", "rw");
FileChannel fc = raf.getChannel();
 
// Отображение части файла в память (первые 100МБ)
MappedByteBuffer buffer = fc.map(FileChannel.MapMode.READ_WRITE, 0, 100 * 1024 * 1024);
 
// Работа с файлом как с массивом байтов
buffer.putInt(100); // Запись целого числа в начало файла
buffer.position(1000000); // Прыжок на позицию 1MB
buffer.putInt(999); // Запись числа на новой позиции
 
// Изменения автоматически синхронизируются с файлом
fc.close();
raf.close();

// Scatter-read
ByteBuffer headerBuffer = ByteBuffer.allocate(128);
ByteBuffer bodyBuffer = ByteBuffer.allocate(1024);
ByteBuffer[] buffers = { headerBuffer, bodyBuffer };
 
SocketChannel channel = SocketChannel.open(new InetSocketAddress("example.com", 80));
long bytesRead = channel.read(buffers);
 
// Gather-write
headerBuffer.flip();
bodyBuffer.flip();
long bytesWritten = channel.write(buffers);

Charset utf8 = Charset.forName("UTF-8");
Charset utf16 = Charset.forName("UTF-16");
 
// Преобразование UTF-8 в UTF-16
ByteBuffer utf8Buffer = utf8.encode("Привет, мир!");
CharBuffer charBuffer = utf8.decode(utf8Buffer);
ByteBuffer utf16Buffer = utf16.encode(charBuffer);

// Неправильно:
buffer.clear();
channel.read(buffer); // Игнорируем возвращаемое значение!
buffer.flip();
// Ошибка: буфер может быть пуст или заполнен частично
 
// Правильно:
buffer.clear();
int bytesRead = channel.read(buffer);
if (bytesRead > 0) {
    buffer.flip();
    // Теперь можно безопасно обрабатывать данные
}

// Старый подход
File oldFile = new File("/home/user/documents/report.txt");
 
// Новый подход с NIO.2
Path path = Paths.get("/home/user/documents/report.txt");
// или используя URI
Path pathFromUri = Paths.get(URI.create("file:///home/user/documents/report.txt"));

Path basePath = Paths.get("/home/user");
Path fullPath = basePath.resolve("documents").resolve("report.txt");
 
// Получение компонентов пути
Path fileName = fullPath.getFileName(); // report.txt
Path parent = fullPath.getParent();     // /home/user/documents

// Чтение всего содержимого файла одной командой
List<String> lines = Files.readAllLines(Paths.get("data.txt"), StandardCharsets.UTF_8);
 
// Запись строк в файл
List<String> outputLines = Arrays.asList("Первая строка", "Вторая строка");
Files.write(Paths.get("output.txt"), outputLines, StandardCharsets.UTF_8);
 
// Атомарное копирование файла с заменой существующего
Files.copy(source, target, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE);

Path file = Paths.get("document.docx");
BasicFileAttributes attrs = Files.readAttributes(file, BasicFileAttributes.class);
 
System.out.println("Создан: " + attrs.creationTime());
System.out.println("Изменён: " + attrs.lastModifiedTime());
System.out.println("Размер: " + attrs.size() + " байт");
System.out.println("Директория? " + attrs.isDirectory());
System.out.println("Обычный файл? " + attrs.isRegularFile());
System.out.println("Символическая ссылка? " + attrs.isSymbolicLink());

PosixFileAttributes posixAttrs = Files.readAttributes(file, PosixFileAttributes.class);
Set<PosixFilePermission> permissions = posixAttrs.permissions();
 
System.out.println("Владелец: " + posixAttrs.owner().getName());
System.out.println("Группа: " + posixAttrs.group().getName());
System.out.println("Права: " + PosixFilePermissions.toString(permissions)); // rwxr-x---

try (WatchService watchService = FileSystems.getDefault().newWatchService()) {
    Path directory = Paths.get("/path/to/watch");
    directory.register(watchService, 
                      StandardWatchEventKinds.ENTRY_CREATE, 
                      StandardWatchEventKinds.ENTRY_MODIFY, 
                      StandardWatchEventKinds.ENTRY_DELETE);
    
    while (true) {
        WatchKey key = watchService.take(); // Блокирующий вызов, ждёт событий
        for (WatchEvent<?> event : key.pollEvents()) {
            WatchEvent.Kind<?> kind = event.kind();
            Path name = (Path) event.context();
            Path child = directory.resolve(name);
            
            System.out.println(kind + ": " + child);
            
            // Обработка события...
        }
        
        // Важно: нужно сбросить ключ, иначе он станет недействительным
        boolean valid = key.reset();
        if (!valid) {
            break; // Директория стала недоступна
        }
    }
} catch (IOException | InterruptedException e) {
    e.printStackTrace();
}

// Создание символической ссылки
Path original = Paths.get("original.txt");
Path link = Paths.get("link.txt");
Files.createSymbolicLink(link, original);
 
// Проверка, является ли файл символической ссылкой
boolean isLink = Files.isSymbolicLink(link);
 
// Чтение целевого пути символической ссылки
Path target = Files.readSymbolicLink(link);

// Создание жёсткой ссылки
Files.createLink(Paths.get("hardlink.txt"), original);

Files.walkFileTree(Paths.get("/home/user/documents"), new SimpleFileVisitor<Path>() {
@Override
public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
    if (file.toString().endsWith(".txt")) {
        System.out.println("Найден текстовый файл: " + file);
    }
    return FileVisitResult.CONTINUE;
}
 
@Override
public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) {
    System.out.println("Входим в директорию: " + dir);
    return FileVisitResult.CONTINUE;
}
 
@Override
public FileVisitResult visitFileFailed(Path file, IOException exc) {
    System.err.println("Ошибка при посещении файла: " + file);
    return FileVisitResult.CONTINUE;
}
});

// Этот код корректно работает и в Windows, и в Linux/Mac
Path config = Paths.get("app", "config", "settings.xml");
 
// В Windows будет: app\config\settings.xml
// В Unix-системах: app/config/settings.xml

FileSystem fs = FileSystems.getDefault();
 
// Получение разделителя пути для текущей файловой системы
String separator = fs.getSeparator(); // \ в Windows, / в UNIX
 
// Получение корневых директорий (особенно важно для Windows с её буквами дисков)
Iterable<Path> rootDirectories = fs.getRootDirectories();
for (Path root : rootDirectories) {
System.out.println(root);
}

// Атомарное перемещение файла (или переименование)
Files.move(source, target, StandardCopyOption.ATOMIC_MOVE);
 
// Атомарное создание файла (с ошибкой, если файл уже существует)
Files.createFile(path);
 
// Атомарное создание временного файла
Path temp = Files.createTempFile("prefix", ".suffix");

Path file = Paths.get("document.txt");
 
// Установка времени последнего доступа и модификации
FileTime lastModified = FileTime.from(Instant.now().minus(1, ChronoUnit.DAYS));
Files.setAttribute(file, "lastModifiedTime", lastModified);
 
// В POSIX-системах можно изменить права доступа
if (FileSystems.getDefault().supportedFileAttributeViews().contains("posix")) {
    Set<PosixFilePermission> perms = EnumSet.of(
        PosixFilePermission.OWNER_READ, PosixFilePermission.OWNER_WRITE,
        PosixFilePermission.GROUP_READ
    );
    Files.setPosixFilePermissions(file, perms);
}

// Открытие асинхронного файлового канала
AsynchronousFileChannel fileChannel = AsynchronousFileChannel.open(
    Paths.get("bigdata.txt"), StandardOpenOption.READ);
 
// Создание буфера для чтения
ByteBuffer buffer = ByteBuffer.allocate(1024);
long position = 0;
 
// Вариант 1: использование Future
Future<Integer> future = fileChannel.read(buffer, position);
// В этот момент можно делать другую работу...
 
// Когда нужен результат:
try {
    Integer bytesRead = future.get(); // Блокирующий вызов, но только когда нам реально нужны данные
    buffer.flip();
    // Обработка данных...
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
}

CompletableFuture<Integer> readFuture = new CompletableFuture<>();
 
// Асинхронное чтение с использованием CompletionHandler
fileChannel.read(buffer, position, buffer, new CompletionHandler<Integer, ByteBuffer>() {
    @Override
    public void completed(Integer result, ByteBuffer attachment) {
        attachment.flip();
        readFuture.complete(result);
    }
    
    @Override
    public void failed(Throwable exc, ByteBuffer attachment) {
        readFuture.completeExceptionally(exc);
    }
});
 
// Цепочка обработки результата
readFuture
    .thenApply(bytesRead -> {
        // Преобразование данных буфера
        byte[] data = new byte[buffer.limit()];
        buffer.get(data);
        return new String(data);
    })
    .thenAccept(content -> System.out.println("Прочитано: " + content))
    .exceptionally(throwable -> {
        System.err.println("Ошибка при чтении: " + throwable.getMessage());
        return null;
    });

// Пример с использованием Project Reactor
Flux.create(sink -> {
    try {
        AsynchronousFileChannel channel = AsynchronousFileChannel.open(
            Paths.get("data.txt"), StandardOpenOption.READ);
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        
        // Рекурсивное асинхронное чтение
        AtomicLong position = new AtomicLong(0);
        readChunk(channel, buffer, position, sink);
    } catch (IOException e) {
        sink.error(e);
    }
})
.map(bytes -> new String(bytes, StandardCharsets.UTF_8))
.flatMap(content -> Flux.fromArray(content.split("\n")))
.filter(line -> line.contains("ERROR"))
.subscribe(
    errorLine -> System.out.println("Найдена ошибка: " + errorLine),
    error -> System.err.println("Произошла ошибка: " + error.getMessage()),
    () -> System.out.println("Обработка завершена")
);
 
// Вспомогательный метод для рекурсивного чтения
private static void readChunk(AsynchronousFileChannel channel, ByteBuffer buffer, 
                            AtomicLong position, FluxSink<byte[]> sink) {
    buffer.clear();
    channel.read(buffer, position.get(), null, new CompletionHandler<Integer, Void>() {
        @Override
        public void completed(Integer bytesRead, Void attachment) {
            if (bytesRead > 0) {
                buffer.flip();
                byte[] data = new byte[buffer.limit()];
                buffer.get(data);
                position.addAndGet(bytesRead);
                sink.next(data);
                readChunk(channel, buffer, position, sink); // Рекурсивное чтение следующего куска
            } else {
                try {
                    channel.close();
                    sink.complete();
                } catch (IOException e) {
                    sink.error(e);
                }
            }
        }
        
        @Override
        public void failed(Throwable exc, Void attachment) {
            sink.error(exc);
        }
    });
}

// Антипаттерн: игнорирование исключений
fileChannel.read(buffer, 0, null, new CompletionHandler<Integer, Void>() {
    @Override
    public void completed(Integer result, Void attachment) {
        // Обработка успешного результата
    }
    
    @Override
    public void failed(Throwable exc, Void attachment) {
        // Не делаем ничего с исключением - ПЛОХО!
    }
});
 
// Правильное решение: логирование и/или распространение исключения
fileChannel.read(buffer, 0, null, new CompletionHandler<Integer, Void>() {
    @Override
    public void completed(Integer result, Void attachment) {
        // Обработка успешного результата
    }
    
    @Override
    public void failed(Throwable exc, Void attachment) {
        logger.error("Ошибка при асинхронном чтении", exc);
        // Возможно, уведомление системы обработки ошибок
        errorHandler.handleError(exc);
    }
});

@RestController
public class AsyncFileController {
    @GetMapping("/download/{filename}")
    public Mono<ResponseEntity<ByteArrayResource>> downloadFile(@PathVariable String filename) {
        Path path = Paths.get("/files/" + filename);
        
        return Mono.fromCallable(() -> AsynchronousFileChannel.open(path, StandardOpenOption.READ))
            .flatMap(channel -> {
                FileAttribute<?> attr = Files.readAttributes(path, BasicFileAttributes.class);
                long fileSize = attr.size();
                ByteBuffer buffer = ByteBuffer.allocate((int) fileSize);
                
                return Mono.create(sink -> {
                    channel.read(buffer, 0, null, new CompletionHandler<Integer, Void>() {
                        @Override
                        public void completed(Integer result, Void attachment) {
                            buffer.flip();
                            byte[] data = new byte[buffer.limit()];
                            buffer.get(data);
                            sink.success(data);
                            try {
                                channel.close();
                            } catch (IOException e) {
                                sink.error(e);
                            }
                        }
                        
                        @Override
                        public void failed(Throwable exc, Void attachment) {
                            sink.error(exc);
                        }
                    });
                });
            })
            .map(data -> {
                ByteArrayResource resource = new ByteArrayResource(data);
                return ResponseEntity.ok()
                    .header("Content-Disposition", "attachment; filename=" + filename)
                    .contentLength(data.length)
                    .body(resource);
            })
            .onErrorResume(e -> {
                return Mono.just(ResponseEntity.notFound().build());
            });
    }
}

@Test
public void testAsyncFileRead() {
    Path testFile = Files.createTempFile("test", ".txt");
    Files.write(testFile, "Hello, Async World!".getBytes());
    
    Flux<String> fileContentFlux = asyncFileService.readFileAsLines(testFile);
    
    StepVerifier.create(fileContentFlux)
        .expectNext("Hello, Async World!")
        .verifyComplete();
    
    Files.delete(testFile);
}

// Правильное освобождение ресурсов в асинхронном контексте
public Mono<String> readFileAsync(Path path) {
    return Mono.using(
        // Открытие ресурса
        () -> AsynchronousFileChannel.open(path, StandardOpenOption.READ),
        
        // Использование ресурса
        channel -> {
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            return Mono.create(sink -> {
                channel.read(buffer, 0, null, new CompletionHandler<Integer, Void>() {
                    @Override
                    public void completed(Integer result, Void attachment) {
                        buffer.flip();
                        byte[] data = new byte[buffer.limit()];
                        buffer.get(data);
                        sink.success(new String(data, StandardCharsets.UTF_8));
                    }
                    
                    @Override
                    public void failed(Throwable exc, Void attachment) {
                        sink.error(exc);
                    }
                });
            });
        },
        
        // Закрытие ресурса (выполняется после завершения Mono)
        channel -> {
            try {
                channel.close();
            } catch (IOException e) {
                logger.error("Ошибка при закрытии канала", e);
            }
        }
    );
}

public class IOBenchmark {
    private static final int ITERATIONS = 100;
    private static final int WARM_UP = 20;
    private static final String FILE_PATH = "test_data.bin";
    private static final int FILE_SIZE = 100 * 1024 * 1024; // 100MB
    
    public static void main(String[] args) throws Exception {
        // Создаем тестовый файл
        createTestFile(FILE_PATH, FILE_SIZE);
        
        // Прогрев JVM
        for (int i = 0; i < WARM_UP; i++) {
            readFileClassicIO(FILE_PATH);
            readFileNIO(FILE_PATH);
            readFileNIODirect(FILE_PATH);
            readFileAsync(FILE_PATH);
        }
        
        // Бенчмарк классического I/O
        long classicIOTime = benchmarkOperation(IOBenchmark::readFileClassicIO);
        System.out.println("Classic I/O: " + classicIOTime + " ms");
        
        // Бенчмарк NIO с обычным буфером
        long nioTime = benchmarkOperation(IOBenchmark::readFileNIO);
        System.out.println("NIO: " + nioTime + " ms");
        
        // Бенчмарк NIO с прямым буфером
        long nioDirectTime = benchmarkOperation(IOBenchmark::readFileNIODirect);
        System.out.println("NIO Direct: " + nioDirectTime + " ms");
        
        // Бенчмарк асинхронного I/O
        long asyncTime = benchmarkOperation(IOBenchmark::readFileAsync);
        System.out.println("Async I/O: " + asyncTime + " ms");
    }
    
    // Методы тестирования и вспомогательные функции...
}

// АНТИПАТТЕРН: загрузка большого файла целиком в память
byte[] fileContent = Files.readAllBytes(Paths.get("huge.dat")); // Опасно!

// Правильный подход для больших файлов
try (InputStream in = Files.newInputStream(Paths.get("huge.dat"));
     BufferedInputStream bis = new BufferedInputStream(in, 8192)) {
    byte[] buffer = new byte[8192];
    int bytesRead;
    while ((bytesRead = bis.read(buffer)) != -1) {
        // Обработка порции данных
        processChunk(buffer, bytesRead);
    }
}

// Потенциальная утечка памяти
for (int i = 0; i < 1000000; i++) {
    ByteBuffer directBuffer = ByteBuffer.allocateDirect(1024 * 1024); // 1MB
    // Использование буфера...
    // Буфер остаётся в нативной памяти до сборки мусора, которая может задерживаться
}

// Паттерн "пул буферов" для эффективного использования DirectByteBuffer
public class DirectBufferPool {
    private final Queue<ByteBuffer> buffers = new ConcurrentLinkedQueue<>();
    private final int bufferSize;
    private final int maxBuffers;
    private AtomicInteger count = new AtomicInteger(0);
    
    public DirectBufferPool(int bufferSize, int maxBuffers) {
        this.bufferSize = bufferSize;
        this.maxBuffers = maxBuffers;
    }
    
    public ByteBuffer acquire() {
        ByteBuffer buffer = buffers.poll();
        if (buffer == null) {
            if (count.incrementAndGet() <= maxBuffers) {
                buffer = ByteBuffer.allocateDirect(bufferSize);
            } else {
                count.decrementAndGet();
                throw new IllegalStateException("Достигнут лимит буферов");
            }
        }
        return buffer;
    }
    
    public void release(ByteBuffer buffer) {
        buffer.clear();
        buffers.offer(buffer);
    }
}

public byte[] readFile(Path path) throws IOException {
    long size = Files.size(path);
    
    if (size < THRESHOLD) {
        // Для маленьких файлов классический I/O эффективнее
        return Files.readAllBytes(path);
    } else {
        // Для больших файлов используем NIO с прямым буфером
        try (FileChannel channel = FileChannel.open(path, StandardOpenOption.READ)) {
            ByteBuffer buffer = ByteBuffer.allocateDirect((int) size);
            channel.read(buffer);
            buffer.flip();
            
            byte[] result = new byte[buffer.remaining()];
            buffer.get(result);
            return result;
        }
    }
}

int processors = Runtime.getRuntime().availableProcessors();
ExecutorService executor = Executors.newFixedThreadPool(processors);
 
List<Path> files = getFilesToProcess();
List<Future<Result>> results = files.stream()
    .map(file -> executor.submit(() -> processFileAsync(file)))
    .collect(Collectors.toList());