google-site-verification: google61fe8ba583a51912.html
MACRO NEPAL
CompletableFuture in Java
awiskar acharya
October 31, 2025
9:39 am

CompletableFuture is a powerful class introduced in Java 8 that represents a future result of an asynchronous computation. It provides a rich API for composing, combining, and executing asynchronous tasks.

1. Basic CompletableFuture Operations

Creating CompletableFuture

import java.util.concurrent.*; import java.util.function.*; public class CompletableFutureBasics { public static void main(String[] args) throws Exception { // 1. Creating completed CompletableFuture CompletableFuture<String> completedFuture = CompletableFuture.completedFuture("Hello World"); System.out.println("Completed future: " + completedFuture.get()); // 2. Running async task without return value CompletableFuture<Void> runAsyncFuture = CompletableFuture.runAsync(() -> { System.out.println("Running in: " + Thread.currentThread().getName()); sleep(1000); System.out.println("Run async completed"); }); // 3. Running async task with return value CompletableFuture<String> supplyAsyncFuture = CompletableFuture.supplyAsync(() -> { System.out.println("Supplying in: " + Thread.currentThread().getName()); sleep(1000); return "Supply Async Result"; }); System.out.println("Supply async result: " + supplyAsyncFuture.get()); // 4. Using custom executor ExecutorService customExecutor = Executors.newFixedThreadPool(3); CompletableFuture<String> customFuture = CompletableFuture.supplyAsync(() -> { System.out.println("Using custom executor: " + Thread.currentThread().getName()); return "Custom executor result"; }, customExecutor); System.out.println("Custom executor result: " + customFuture.get()); customExecutor.shutdown(); } private static void sleep(long millis) { try { Thread.sleep(millis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } }

Basic Transformation and Chaining

public class TransformationExamples { public static void main(String[] args) throws Exception { // 1. thenApply - transform result CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello") .thenApply(s -> s + " World") .thenApply(String::toUpperCase); System.out.println("thenApply result: " + future.get()); // 2. thenAccept - consume result without returning CompletableFuture<Void> acceptFuture = CompletableFuture.supplyAsync(() -> "Hello") .thenAccept(s -> System.out.println("Consuming: " + s)); acceptFuture.get(); // 3. thenRun - run after completion, no access to result CompletableFuture<Void> runFuture = CompletableFuture.supplyAsync(() -> "Hello") .thenRun(() -> System.out.println("Task completed")); runFuture.get(); // 4. thenCompose - chain dependent futures (flatMap equivalent) CompletableFuture<String> composeFuture = getUser(1L) .thenCompose(TransformationExamples::getUserProfile); System.out.println("Composed result: " + composeFuture.get()); // 5. Async variants - execute on different threads CompletableFuture<String> asyncFuture = CompletableFuture.supplyAsync(() -> "Hello") .thenApplyAsync(s -> { System.out.println("Async transformation in: " + Thread.currentThread().getName()); return s + " Async World"; }); System.out.println("Async result: " + asyncFuture.get()); } private static CompletableFuture<User> getUser(Long userId) { return CompletableFuture.supplyAsync(() -> { sleep(500); return new User(userId, "user" + userId); }); } private static CompletableFuture<String> getUserProfile(User user) { return CompletableFuture.supplyAsync(() -> { sleep(500); return "Profile for " + user.getName(); }); } private static void sleep(long millis) { try { Thread.sleep(millis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } class User { private Long id; private String name; public User(Long id, String name) { this.id = id; this.name = name; } public Long getId() { return id; } public String getName() { return name; } }

2. Combining Multiple Futures

Combining Independent Futures

public class CombiningFutures { public static void main(String[] args) throws Exception { // 1. thenCombine - combine two independent futures CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> { sleep(1000); return "Hello"; }); CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> { sleep(500); return "World"; }); CompletableFuture<String> combined = future1.thenCombine(future2, (s1, s2) -> s1 + " " + s2); System.out.println("Combined: " + combined.get()); // 2. thenAcceptBoth - consume both results CompletableFuture<Void> acceptBoth = future1.thenAcceptBoth(future2, (s1, s2) -> System.out.println("Both: " + s1 + " and " + s2)); acceptBoth.get(); // 3. runAfterBoth - run after both complete CompletableFuture<Void> runAfterBoth = future1.runAfterBoth(future2, () -> System.out.println("Both futures completed")); runAfterBoth.get(); // 4. allOf - wait for all futures to complete CompletableFuture<String>[] futures = new CompletableFuture[] { CompletableFuture.supplyAsync(() -> { sleep(1000); return "Task1"; }), CompletableFuture.supplyAsync(() -> { sleep(1500); return "Task2"; }), CompletableFuture.supplyAsync(() -> { sleep(500); return "Task3"; }) }; CompletableFuture<Void> allOf = CompletableFuture.allOf(futures); allOf.thenRun(() -> { System.out.println("All tasks completed"); for (CompletableFuture<String> f : futures) { System.out.println("Result: " + f.join()); // join() doesn't throw checked exception } }).get(); // 5. anyOf - wait for any future to complete CompletableFuture<Object> anyOf = CompletableFuture.anyOf(futures); System.out.println("First completed: " + anyOf.get()); } private static void sleep(long millis) { try { Thread.sleep(millis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } }

Complex Combination Examples

public class AdvancedCombination { public static void main(String[] args) throws Exception { // Combine multiple API calls CompletableFuture<User> userFuture = getUser(1L); CompletableFuture<List<Order>> ordersFuture = getOrders(1L); CompletableFuture<List<Address>> addressesFuture = getAddresses(1L); // Combine all results CompletableFuture<UserProfile> profileFuture = userFuture .thenCombine(ordersFuture, (user, orders) -> new UserProfile(user, orders, null)) .thenCombine(addressesFuture, (profile, addresses) -> { profile.setAddresses(addresses); return profile; }); UserProfile profile = profileFuture.get(); System.out.println("User profile: " + profile); // Process results as they complete processResultsAsTheyComplete(); } private static void processResultsAsTheyComplete() throws Exception { CompletableFuture<String> fastTask = CompletableFuture.supplyAsync(() -> { sleep(500); return "Fast task completed"; }); CompletableFuture<String> mediumTask = CompletableFuture.supplyAsync(() -> { sleep(1000); return "Medium task completed"; }); CompletableFuture<String> slowTask = CompletableFuture.supplyAsync(() -> { sleep(2000); return "Slow task completed"; }); // Apply to either - process whichever completes first fastTask.acceptEither(mediumTask, result -> System.out.println("First completed: " + result)); // Chain with the faster one CompletableFuture<String> result = fastTask.applyToEither(slowTask, firstResult -> "Processing: " + firstResult); System.out.println("Apply to either result: " + result.get()); } private static CompletableFuture<User> getUser(Long userId) { return CompletableFuture.supplyAsync(() -> { sleep(1000); return new User(userId, "John Doe"); }); } private static CompletableFuture<List<Order>> getOrders(Long userId) { return CompletableFuture.supplyAsync(() -> { sleep(1500); return List.of(new Order(1L, "Order1"), new Order(2L, "Order2")); }); } private static CompletableFuture<List<Address>> getAddresses(Long userId) { return CompletableFuture.supplyAsync(() -> { sleep(1200); return List.of(new Address("123 Main St"), new Address("456 Oak Ave")); }); } private static void sleep(long millis) { try { Thread.sleep(millis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } class UserProfile { private User user; private List<Order> orders; private List<Address> addresses; public UserProfile(User user, List<Order> orders, List<Address> addresses) { this.user = user; this.orders = orders; this.addresses = addresses; } // getters and setters public void setAddresses(List<Address> addresses) { this.addresses = addresses; } @Override public String toString() { return String.format("UserProfile{user=%s, orders=%d, addresses=%d}", user.getName(), orders.size(), addresses.size()); } } class Order { private Long id; private String name; public Order(Long id, String name) { this.id = id; this.name = name; } public String getName() { return name; } } class Address { private String street; public Address(String street) { this.street = street; } public String getStreet() { return street; } }

3. Error Handling and Recovery

public class ErrorHandlingExamples { public static void main(String[] args) throws Exception { // 1. exceptionally - handle exceptions CompletableFuture<String> futureWithException = CompletableFuture.supplyAsync(() -> { if (true) throw new RuntimeException("Something went wrong!"); return "Success"; }); CompletableFuture<String> recovered = futureWithException.exceptionally(ex -> { System.out.println("Exception caught: " + ex.getMessage()); return "Default Value"; }); System.out.println("Recovered result: " + recovered.get()); // 2. handle - handle both success and failure CompletableFuture<String> handled = CompletableFuture.supplyAsync(() -> { if (Math.random() > 0.5) { throw new RuntimeException("Random failure"); } return "Success"; }).handle((result, exception) -> { if (exception != null) { return "Recovered from: " + exception.getMessage(); } return result; }); System.out.println("Handled result: " + handled.get()); // 3. whenComplete - always execute, similar to finally CompletableFuture<String> withCompletion = CompletableFuture.supplyAsync(() -> { if (Math.random() > 0.7) { throw new RuntimeException("Operation failed"); } return "Operation succeeded"; }).whenComplete((result, exception) -> { if (exception != null) { System.out.println("Task failed with: " + exception.getMessage()); } else { System.out.println("Task succeeded with: " + result); } }); try { System.out.println("Final result: " + withCompletion.get()); } catch (Exception e) { System.out.println("Exception in get: " + e.getCause().getMessage()); } // 4. Complex recovery chain complexRecoveryExample(); } private static void complexRecoveryExample() throws Exception { CompletableFuture<String> complexFuture = fetchDataFromPrimary() .exceptionally(ex -> { System.out.println("Primary failed, trying secondary: " + ex.getMessage()); return fetchDataFromSecondary().join(); // join since we're in exceptionally }) .handle((result, ex) -> { if (ex != null) { System.out.println("Both primary and secondary failed, using cache"); return getCachedData(); } return result; }) .thenApply(data -> "Processed: " + data); System.out.println("Complex recovery result: " + complexFuture.get()); } private static CompletableFuture<String> fetchDataFromPrimary() { return CompletableFuture.supplyAsync(() -> { sleep(500); if (Math.random() > 0.3) { throw new RuntimeException("Primary service unavailable"); } return "Data from primary"; }); } private static CompletableFuture<String> fetchDataFromSecondary() { return CompletableFuture.supplyAsync(() -> { sleep(300); if (Math.random() > 0.2) { throw new RuntimeException("Secondary service unavailable"); } return "Data from secondary"; }); } private static String getCachedData() { return "Cached data"; } private static void sleep(long millis) { try { Thread.sleep(millis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } }

4. Timeout and Control Operations

public class TimeoutControlExamples { public static void main(String[] args) throws Exception { // 1. Complete manually CompletableFuture<String> manualFuture = new CompletableFuture<>(); new Thread(() -> { sleep(2000); manualFuture.complete("Manually completed"); }).start(); // Complete exceptionally CompletableFuture<String> exceptionalFuture = new CompletableFuture<>(); new Thread(() -> { sleep(1000); exceptionalFuture.completeExceptionally(new RuntimeException("Forced failure")); }).start(); // 2. Timeout with completeOnTimeout (Java 9+) CompletableFuture<String> timeoutFuture = CompletableFuture.supplyAsync(() -> { sleep(3000); return "Slow result"; }).completeOnTimeout("Timeout default", 1, TimeUnit.SECONDS); System.out.println("Timeout result: " + timeoutFuture.get()); // 3. Timeout with orTimeout (Java 9+) CompletableFuture<String> orTimeoutFuture = CompletableFuture.supplyAsync(() -> { sleep(3000); return "Slow result"; }).orTimeout(1, TimeUnit.SECONDS); try { System.out.println("OrTimeout result: " + orTimeoutFuture.get()); } catch (ExecutionException e) { System.out.println("Timeout occurred: " + e.getCause().getMessage()); } // 4. Cancel operation CompletableFuture<String> cancelableFuture = new CompletableFuture<>(); new Thread(() -> { sleep(1000); cancelableFuture.cancel(true); }).start(); try { cancelableFuture.get(); } catch (CancellationException e) { System.out.println("Future was cancelled"); } // 5. Check completion status CompletableFuture<String> statusFuture = CompletableFuture.supplyAsync(() -> { sleep(500); return "Done"; }); System.out.println("Is done: " + statusFuture.isDone()); System.out.println("Is cancelled: " + statusFuture.isCancelled()); System.out.println("Is completed exceptionally: " + statusFuture.isCompletedExceptionally()); statusFuture.get(); System.out.println("Is done after get: " + statusFuture.isDone()); } private static void sleep(long millis) { try { Thread.sleep(millis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } }

5. Real-World Use Cases

Parallel API Calls

public class ParallelApiCalls { public static void main(String[] args) throws Exception { long startTime = System.currentTimeMillis(); // Make multiple API calls in parallel CompletableFuture<UserInfo> userInfoFuture = getUserInfo(1L); CompletableFuture<List<Product>> productsFuture = getRecommendedProducts(1L); CompletableFuture<List<Review>> reviewsFuture = getUserReviews(1L); // Combine all results CompletableFuture<UserDashboard> dashboardFuture = userInfoFuture .thenCombine(productsFuture, (userInfo, products) -> new UserDashboard(userInfo, products, null)) .thenCombine(reviewsFuture, (dashboard, reviews) -> { dashboard.setReviews(reviews); return dashboard; }); UserDashboard dashboard = dashboardFuture.get(); long endTime = System.currentTimeMillis(); System.out.println("Dashboard: " + dashboard); System.out.println("Total time: " + (endTime - startTime) + "ms"); // Compare with sequential execution sequentialExecution(); } private static void sequentialExecution() throws Exception { long startTime = System.currentTimeMillis(); UserInfo userInfo = getUserInfo(1L).get(); List<Product> products = getRecommendedProducts(1L).get(); List<Review> reviews = getUserReviews(1L).get(); UserDashboard dashboard = new UserDashboard(userInfo, products, reviews); long endTime = System.currentTimeMillis(); System.out.println("Sequential time: " + (endTime - startTime) + "ms"); } private static CompletableFuture<UserInfo> getUserInfo(Long userId) { return CompletableFuture.supplyAsync(() -> { sleep(1000); // Simulate API call return new UserInfo(userId, "John Doe", "[email protected]"); }); } private static CompletableFuture<List<Product>> getRecommendedProducts(Long userId) { return CompletableFuture.supplyAsync(() -> { sleep(1200); // Simulate API call return List.of( new Product(1L, "Laptop"), new Product(2L, "Mouse"), new Product(3L, "Keyboard") ); }); } private static CompletableFuture<List<Review>> getUserReviews(Long userId) { return CompletableFuture.supplyAsync(() -> { sleep(800); // Simulate API call return List.of( new Review(1L, "Great product!"), new Review(2L, "Fast delivery") ); }); } private static void sleep(long millis) { try { Thread.sleep(millis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } class UserInfo { private Long id; private String name; private String email; public UserInfo(Long id, String name, String email) { this.id = id; this.name = name; this.email = email; } public String getName() { return name; } } class Product { private Long id; private String name; public Product(Long id, String name) { this.id = id; this.name = name; } public String getName() { return name; } } class Review { private Long id; private String comment; public Review(Long id, String comment) { this.id = id; this.comment = comment; } public String getComment() { return comment; } } class UserDashboard { private UserInfo userInfo; private List<Product> products; private List<Review> reviews; public UserDashboard(UserInfo userInfo, List<Product> products, List<Review> reviews) { this.userInfo = userInfo; this.products = products; this.reviews = reviews; } public void setReviews(List<Review> reviews) { this.reviews = reviews; } @Override public String toString() { return String.format("UserDashboard{user=%s, products=%d, reviews=%d}", userInfo.getName(), products.size(), reviews.size()); } }

Batch Processing

public class BatchProcessing { public static void main(String[] args) throws Exception { // Process items in batches asynchronously List<Long> itemIds = List.of(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, 9L, 10L); // Process all items in parallel List<CompletableFuture<ProcessedItem>> futures = itemIds.stream() .map(BatchProcessing::processItemAsync) .toList(); // Wait for all to complete CompletableFuture<Void> allFutures = CompletableFuture.allOf( futures.toArray(new CompletableFuture[0])); // Collect results CompletableFuture<List<ProcessedItem>> resultsFuture = allFutures.thenApply(v -> futures.stream() .map(CompletableFuture::join) .toList()); List<ProcessedItem> results = resultsFuture.get(); System.out.println("Processed " + results.size() + " items"); // Process with limited parallelism processWithLimitedParallelism(itemIds); } private static void processWithLimitedParallelism(List<Long> itemIds) throws Exception { ExecutorService limitedExecutor = Executors.newFixedThreadPool(3); List<CompletableFuture<ProcessedItem>> futures = itemIds.stream() .map(id -> CompletableFuture.supplyAsync(() -> processItem(id), limitedExecutor)) .toList(); CompletableFuture<List<ProcessedItem>> allResults = CompletableFuture.allOf( futures.toArray(new CompletableFuture[0])) .thenApply(v -> futures.stream() .map(CompletableFuture::join) .toList()); List<ProcessedItem> results = allResults.get(); System.out.println("Limited parallelism processed: " + results.size() + " items"); limitedExecutor.shutdown(); } private static CompletableFuture<ProcessedItem> processItemAsync(Long itemId) { return CompletableFuture.supplyAsync(() -> processItem(itemId)); } private static ProcessedItem processItem(Long itemId) { sleep(500); // Simulate processing time System.out.println("Processing item " + itemId + " in " + Thread.currentThread().getName()); return new ProcessedItem(itemId, "Processed_" + itemId); } private static void sleep(long millis) { try { Thread.sleep(millis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } class ProcessedItem { private Long id; private String data; public ProcessedItem(Long id, String data) { this.id = id; this.data = data; } public Long getId() { return id; } public String getData() { return data; } }

6. Best Practices and Common Patterns

public class BestPractices { public static void main(String[] args) { // 1. Always handle exceptions safeAsyncOperation(); // 2. Use custom executors for resource control useCustomExecutor(); // 3. Avoid blocking in async operations nonBlockingComposition(); // 4. Proper resource cleanup resourceCleanupExample(); } private static void safeAsyncOperation() { CompletableFuture<String> safeFuture = CompletableFuture.supplyAsync(() -> { // Some operation that might fail return riskyOperation(); }).exceptionally(ex -> { // Always handle exceptions System.err.println("Operation failed: " + ex.getMessage()); return "default-value"; }); // Don't forget to handle the result safeFuture.thenAccept(result -> { System.out.println("Safe result: " + result); }); } private static void useCustomExecutor() { ExecutorService ioBoundExecutor = Executors.newFixedThreadPool(10); ExecutorService cpuBoundExecutor = Executors.newWorkStealingPool(); CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { // I/O bound operation return fetchFromDatabase(); }, ioBoundExecutor).thenApplyAsync(data -> { // CPU intensive processing return processData(data); }, cpuBoundExecutor); future.thenRunAsync(() -> { System.out.println("Completed with custom executors"); }, ioBoundExecutor); // Always shutdown custom executors ioBoundExecutor.shutdown(); cpuBoundExecutor.shutdown(); } private static void nonBlockingComposition() { // Good: Non-blocking composition CompletableFuture<String> goodFuture = CompletableFuture.supplyAsync(() -> "data") .thenApplyAsync(data -> transform(data)) .thenComposeAsync(transformed -> furtherProcessing(transformed)); // Bad: Blocking in async chain CompletableFuture<String> badFuture = CompletableFuture.supplyAsync(() -> "data") .thenApply(data -> { try { // Blocking call in transformation! return blockingOperation(data); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException(e); } }); } private static void resourceCleanupExample() { CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { // Acquire resource Resource resource = acquireResource(); try { return useResource(resource); } finally { // Always release resource releaseResource(resource); } }); // Handle completion future.whenComplete((result, ex) -> { if (ex != null) { System.err.println("Operation failed: " + ex.getMessage()); } else { System.out.println("Operation succeeded: " + result); } }); } // Mock methods for demonstration private static String riskyOperation() { return "result"; } private static String fetchFromDatabase() { return "data"; } private static String processData(String data) { return "processed"; } private static String transform(String data) { return "transformed"; } private static CompletableFuture<String> furtherProcessing(String data) { return CompletableFuture.completedFuture("processed"); } private static String blockingOperation(String data) throws InterruptedException { Thread.sleep(1000); return "blocking-result"; } private static Resource acquireResource() { return new Resource(); } private static String useResource(Resource resource) { return "used"; } private static void releaseResource(Resource resource) { /* cleanup */ } } class Resource { // Mock resource class }

Key Features Summary:

  1. Async Execution: Run tasks asynchronously with supplyAsync() and runAsync()
  2. Transformation: Transform results with thenApply(), thenCompose()
  3. Combination: Combine multiple futures with thenCombine(), allOf(), anyOf()
  4. Error Handling: Recover from exceptions with exceptionally(), handle()
  5. Control: Manually complete, cancel, or timeout futures
  6. Chaining: Create complex async workflows with method chaining
  7. Executor Control: Use custom executors for resource management

CompletableFuture provides a powerful and flexible way to write asynchronous, non-blocking code in Java, making it essential for modern high-performance applications.

Category : JAVA

Leave a Reply

Your email address will not be published. Required fields are marked *


Macro Nepal Helper