Asynchronous File Processing with CompletableFuture

Welcome to the course on Mastering Java Concurrency with Practice! In this course, we’ll explore how to apply concurrency utilities such as Semaphores, Phasers, CompletableFuture, and more to real-world problems. Each unit will introduce a problem scenario, walk through its solution using concurrency tools, and follow up with similar practice problems to reinforce your understanding. In this unit, we focus on Asynchronous File Processing using CompletableFuture . This lesson will build on the concurrency concepts you’ve already learned, showing you how to handle multiple file operations asynchronously and keep your programs efficient and responsive.

What You’ll Learn

In this unit, you’ll learn how to: Apply CompletableFuture to handle asynchronous tasks. Manage exceptions in an asynchronous environment. Combine multiple asynchronous operations. Aggregate results from concurrent file tasks efficiently. By now, you should be comfortable with basic threading concepts, so we’ll dive deeper into how asynchronous file processing works in practice using CompletableFuture .

Recap: CompletableFuture and Its Role in Asynchronous Programming

In a previous lesson, you learned about CompletableFuture, a versatile class in Java’s java.util.concurrent package that simplifies asynchronous programming. It allows you to handle tasks without blocking the main thread, combine multiple asynchronous computations, handle exceptions gracefully, and chain operations together. As a quick recap: supplyAsync() runs a task asynchronously and returns a CompletableFuture representing the future result of that task. thenApply() and thenAccept() allow you to process the result of a completed task. exceptionally() provides a way to handle exceptions in an asynchronous pipeline. In this lesson, we’ll use CompletableFuture to process a series of text files asynchronously. Each file will be processed on a separate thread, and once all files have been processed, we’ll aggregate the results to get a total word count. This approach will allow us to process multiple files simultaneously, improving efficiency and responsiveness, particularly when dealing with large files. Now, let’s see how to set up the AsynchronousFileProcessing class and its methods.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Java

public class AsynchronousFileProcessing {

    private final ExecutorService executor;

    public AsynchronousFileProcessing(ExecutorService executor) {
        this.executor = executor;
    }
}

public class AsynchronousFileProcessing {

    private final ExecutorService executor;

    public AsynchronousFileProcessing(ExecutorService executor) {
        this.executor = executor;
    }
}

Java

private long countWordsInFile(Path file) {
    try (Stream<String> lines = Files.lines(file)) {
        return lines
                .flatMap(line -> Stream.of(line.split("\\s+")))
                .filter(word -> !word.isEmpty())
                .count();
    } catch (Exception e) {
        throw new RuntimeException("Failed to read file: " + file.getFileName(), e);
    }
}

private long countWordsInFile(Path file) {
    try (Stream<String> lines = Files.lines(file)) {
        return lines
                .flatMap(line -> Stream.of(line.split("\\s+")))
                .filter(word -> !word.isEmpty())
                .count();
    } catch (Exception e) {
        throw new RuntimeException("Failed to read file: " + file.getFileName(), e);
    }
}

Java

private CompletableFuture<Long> processFileAsync(Path file) {
    return CompletableFuture.supplyAsync(() -> countWordsInFile(file), executor)
            .exceptionally(ex -> {
                System.err.println("Error processing " + file.getFileName() + ": " + ex.getMessage());
                return 0L;
            });
}

private CompletableFuture<Long> processFileAsync(Path file) {
    return CompletableFuture.supplyAsync(() -> countWordsInFile(file), executor)
            .exceptionally(ex -> {
                System.err.println("Error processing " + file.getFileName() + ": " + ex.getMessage());
                return 0L;
            });
}

Java

public CompletableFuture<Long> processFiles(List<Path> files) {
    List<CompletableFuture<Long>> futures = files.stream()
            .map(this::processFileAsync)
            .collect(Collectors.toList());

    return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
            .thenApply(v -> futures.stream()
                    .map(CompletableFuture::join)
                    .reduce(0L, Long::sum));
}

public CompletableFuture<Long> processFiles(List<Path> files) {
    List<CompletableFuture<Long>> futures = files.stream()
            .map(this::processFileAsync)
            .collect(Collectors.toList());

    return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
            .thenApply(v -> futures.stream()
                    .map(CompletableFuture::join)
                    .reduce(0L, Long::sum));
}

Java

ExecutorService executor = Executors.newFixedThreadPool(4);
AsynchronousFileProcessing asyncProcessor = new AsynchronousFileProcessing(executor);

ExecutorService executor = Executors.newFixedThreadPool(4);
AsynchronousFileProcessing asyncProcessor = new AsynchronousFileProcessing(executor);

Java

CompletableFuture<Long> totalWordCountFuture = asyncProcessor.processFiles(files);
totalWordCountFuture.thenAccept(total -> System.out.println("Total word count: " + total))
                    .exceptionally(ex -> {
                        System.err.println("Error: " + ex.getMessage());
                        return null;
                    })
                    .get(); // Wait for all tasks to complete

CompletableFuture<Long> totalWordCountFuture = asyncProcessor.processFiles(files);
totalWordCountFuture.thenAccept(total -> System.out.println("Total word count: " + total))
                    .exceptionally(ex -> {
                        System.err.println("Error: " + ex.getMessage());
                        return null;
                    })
                    .get(); // Wait for all tasks to complete

Java

try {
    totalWordCountFuture.get(); // Block until all tasks are complete
} catch (InterruptedException | ExecutionException e) {
    System.err.println("Error during file processing: " + e.getMessage());
    Thread.currentThread().interrupt();
} finally {
    shutdownExecutor(executor);
}

try {
    totalWordCountFuture.get(); // Block until all tasks are complete
} catch (InterruptedException | ExecutionException e) {
    System.err.println("Error during file processing: " + e.getMessage());
    Thread.currentThread().interrupt();
} finally {
    shutdownExecutor(executor);
}

Java

private void shutdownExecutor(ExecutorService executor) {
    executor.shutdown();
    try {
        if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
            executor.shutdownNow();
            if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
                System.err.println("Executor did not terminate.");
            }
        }
    } catch (InterruptedException e) {
        executor.shutdownNow();
        Thread.currentThread().interrupt();
    }
}

private void shutdownExecutor(ExecutorService executor) {
    executor.shutdown();
    try {
        if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
            executor.shutdownNow();
            if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
                System.err.println("Executor did not terminate.");
            }
        }
    } catch (InterruptedException e) {
        executor.shutdownNow();
        Thread.currentThread().interrupt();
    }
}

Java

public class AsynchronousFileProcessing {

    private final ExecutorService executor;

    public AsynchronousFileProcessing(ExecutorService executor) {
        this.executor = executor;
    }
}

Java

private long countWordsInFile(Path file) {
    try (Stream<String> lines = Files.lines(file)) {
        return lines
                .flatMap(line -> Stream.of(line.split("\\s+")))
                .filter(word -> !word.isEmpty())
                .count();
    } catch (Exception e) {
        throw new RuntimeException("Failed to read file: " + file.getFileName(), e);
    }
}

Java

private CompletableFuture<Long> processFileAsync(Path file) {
    return CompletableFuture.supplyAsync(() -> countWordsInFile(file), executor)
            .exceptionally(ex -> {
                System.err.println("Error processing " + file.getFileName() + ": " + ex.getMessage());
                return 0L;
            });
}

Java

public CompletableFuture<Long> processFiles(List<Path> files) {
    List<CompletableFuture<Long>> futures = files.stream()
            .map(this::processFileAsync)
            .collect(Collectors.toList());

    return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
            .thenApply(v -> futures.stream()
                    .map(CompletableFuture::join)
                    .reduce(0L, Long::sum));
}

Java

ExecutorService executor = Executors.newFixedThreadPool(4);
AsynchronousFileProcessing asyncProcessor = new AsynchronousFileProcessing(executor);

Java

CompletableFuture<Long> totalWordCountFuture = asyncProcessor.processFiles(files);
totalWordCountFuture.thenAccept(total -> System.out.println("Total word count: " + total))
                    .exceptionally(ex -> {
                        System.err.println("Error: " + ex.getMessage());
                        return null;
                    })
                    .get(); // Wait for all tasks to complete

Java

try {
    totalWordCountFuture.get(); // Block until all tasks are complete
} catch (InterruptedException | ExecutionException e) {
    System.err.println("Error during file processing: " + e.getMessage());
    Thread.currentThread().interrupt();
} finally {
    shutdownExecutor(executor);
}

Java

private void shutdownExecutor(ExecutorService executor) {
    executor.shutdown();
    try {
        if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
            executor.shutdownNow();
            if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
                System.err.println("Executor did not terminate.");
            }
        }
    } catch (InterruptedException e) {
        executor.shutdownNow();
        Thread.currentThread().interrupt();
    }
}

Asynchronous File Processing with CompletableFuture

What You’ll Learn

Recap: CompletableFuture and Its Role in Asynchronous Programming

Join the 1M+ learners on CodeSignal

Creating the AsynchronousFileProcessing Class

Using Stream API

Counting Words in Each File

Processing Files Asynchronously

Aggregating Results from Multiple Files

Configuring the ExecutorService

Processing Files in the Main Class

Handling Exceptions and Shutting Down the Executor

Shutting Down the Executor Service

Why It Matters