| name | java-async-concurrent |
| description | 5 async/concurrent patterns with full Java implementations for high-performance systems |
| version | 1.0.0 |
| category | toolchain |
| author | Claude MPM Team |
| license | MIT |
| progressive_disclosure | {"entry_point":{"summary":"Java async/concurrent patterns: Virtual Threads, CompletableFuture, Reactive Streams, Thread Pool, Resilience4j","when_to_use":"When implementing concurrent, async, or resilient systems in Java 21+","quick_start":"Each pattern includes full implementation, key principles, and when to use vs alternatives"}} |
| context_limit | 700 |
| tags | ["java","async","concurrent","virtual-threads","completable-future","reactive","project-reactor","thread-pool","resilience4j","retry"] |
| requires_tools | [] |
Java Async/Concurrent Patterns
Async/Concurrent Patterns
1. Virtual Threads (Java 21)
import java.time.*;
import java.util.concurrent.*;
import java.util.*;
public class VirtualThreadPatterns {
public static <T> List<T> processConcurrentTasks(
List<Callable<T>> tasks,
Duration timeout
) throws InterruptedException, ExecutionException, TimeoutException {
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
List<Future<T>> futures = executor.invokeAll(
tasks,
timeout.toMillis(),
TimeUnit.MILLISECONDS
);
List<T> results = new ArrayList<>();
for (Future<T> future : futures) {
if (!future.isCancelled()) {
results.add(future.get());
}
}
return results;
}
}
public static void runAsyncTask(Runnable task) {
Thread.startVirtualThread(task);
}
}
2. CompletableFuture Pattern
import java.util.concurrent.*;
import java.time.*;
import java.util.*;
import java.util.stream.*;
public class CompletableFuturePatterns {
public static <T> CompletableFuture<T> withTimeout(
Supplier<T> operation,
Duration timeout
) {
return CompletableFuture.supplyAsync(operation)
.orTimeout(timeout.toMillis(), TimeUnit.MILLISECONDS)
.exceptionally(ex -> {
if (ex instanceof TimeoutException) {
throw new RuntimeException("Operation timed out", ex);
}
throw new RuntimeException("Operation failed", ex);
});
}
public static <T> CompletableFuture<List<T>> allOf(
List<CompletableFuture<T>> futures
) {
return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
.thenApply(v -> futures.stream()
.map(CompletableFuture::join)
.collect(Collectors.toList())
);
}
public static CompletableFuture<String> chainedOperations() {
return CompletableFuture.supplyAsync(() -> "initial")
.thenApply(String::toUpperCase)
.thenCompose(s -> CompletableFuture.supplyAsync(() -> s + "_PROCESSED"))
.exceptionally(ex -> "FALLBACK_VALUE");
}
}
3. Reactive Streams (Project Reactor)
import reactor.core.publisher.*;
import reactor.core.scheduler.*;
import java.time.Duration;
import java.util.*;
public class ReactivePatterns {
public static Flux<String> processStream(
Flux<String> input,
int concurrency
) {
return input
.flatMap(
item -> Mono.fromCallable(() -> processItem(item))
.subscribeOn(Schedulers.boundedElastic()),
concurrency
)
.onErrorContinue((error, item) -> {
System.err.println("Failed to process: " + item + ", error: " + error);
})
.timeout(Duration.ofSeconds(10));
}
public static <T> Mono<T> retryWithBackoff(
Mono<T> operation,
int maxRetries
) {
return operation.retryWhen(
Retry.backoff(maxRetries, Duration.ofMillis(100))
.maxBackoff(Duration.ofSeconds(5))
.filter(throwable -> throwable instanceof RuntimeException)
);
}
private static String processItem(String item) {
return "processed_" + item;
}
}
4. Thread Pool Pattern (Traditional)
import java.util.concurrent.*;
import java.time.Duration;
import java.util.*;
public class ThreadPoolPatterns {
public static ExecutorService createCpuBoundPool() {
int cores = Runtime.getRuntime().availableProcessors();
return new ThreadPoolExecutor(
cores,
cores,
60L, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(100),
new ThreadPoolExecutor.CallerRunsPolicy()
);
}
public static ExecutorService createIoBoundPool() {
int cores = Runtime.getRuntime().availableProcessors();
int maxThreads = cores * 2;
return Executors.newFixedThreadPool(maxThreads);
}
public static void shutdownGracefully(ExecutorService executor, Duration timeout) {
executor.shutdown();
try {
if (!executor.awaitTermination(timeout.toMillis(), TimeUnit.MILLISECONDS)) {
executor.shutdownNow();
if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
System.err.println("Executor did not terminate");
}
}
} catch (InterruptedException e) {
executor.shutdownNow();
Thread.currentThread().interrupt();
}
}
}
5. Resilience4j Retry Pattern
import io.github.resilience4j.retry.*;
import io.github.resilience4j.retry.RetryConfig.*;
import java.time.Duration;
import java.util.function.Supplier;
public class ResiliencePatterns {
public static <T> T executeWithRetry(
Supplier<T> operation,
int maxRetries
) {
RetryConfig config = RetryConfig.custom()
.maxAttempts(maxRetries)
.waitDuration(Duration.ofMillis(100))
.intervalFunction(IntervalFunction.ofExponentialBackoff(
Duration.ofMillis(100),
2.0
))
.retryExceptions(RuntimeException.class)
.ignoreExceptions(IllegalArgumentException.class)
.build();
Retry retry = Retry.of("operationRetry", config);
retry.getEventPublisher()
.onRetry(event -> System.out.println("Retry attempt: " + event.getNumberOfRetryAttempts()))
.onError(event -> System.err.println("All retries failed: " + event.getLastThrowable()));
Supplier<T> decoratedSupplier = Retry.decorateSupplier(retry, operation);
return decoratedSupplier.get();
}
}