Java Concurrent Programming
Executor Concurrent Task Execution (with Return Value)
Using Executor and CompletableFuture to concurrent execute the Many Queries. Below is commit changes.
Java Concurrent Programming Case: Task Failure Fail-Fast Cancels Other Tasks
package com.whalefall541.cases.concurrentqry.jobversion;
import lombok.extern.slf4j.Slf4j;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import java.util.stream.Collectors;
@SuppressWarnings("all")
@Slf4j
public class JobFailFastAsyncExecutor implements AutoCloseable {
private final ExecutorService executor;
public JobFailFastAsyncExecutor(int threadCount, String jobName) {
this.executor = Executors.newFixedThreadPool(threadCount, r -> {
Thread thread = new Thread(r);
thread.setName(String.format("%s-%s", jobName, thread.getName()));
return thread;
});
}
/**
* Execute a group of asynchronous tasks, once any task fails, immediately cancel all tasks and propagate exception <br/>
* Truly strict "fail-fast + minimal log"
*
* @param inputs Input parameter list
* @param taskFunction Task processing function, input P return R
* @return An asynchronous CompletableFuture, returns result list on success, throws first exception on failure
*/
public <P, R> CompletableFuture<List<R>> executeFailFast(List<P> inputs, Function<P, R> taskFunction) {
List<CompletableFuture<R>> futures = inputs.stream()
.map(input -> CompletableFuture.supplyAsync(
// Asynchronous thread (from thread pool) executes logic below
() -> taskFunction.apply(input), executor))
.collect(Collectors.toList());
CompletableFuture<List<R>> resultFuture = new CompletableFuture<>();
CommonTaskSupport.registerFailFastHandlers(futures, resultFuture);
CommonTaskSupport.collectAllResults(futures, resultFuture);
return resultFuture;
}
static class CommonTaskSupport {
private CommonTaskSupport() {
}
public static <R> void registerFailFastHandlers(List<CompletableFuture<R>> futures,
CompletableFuture<List<R>> resultFuture) {
AtomicBoolean failFastTriggered = new AtomicBoolean(false);
futures.forEach(future -> future.whenComplete(
// Triggered after asynchronous threads below complete
(r, ex) -> {
if (ex != null && failFastTriggered.compareAndSet(false, true)) {
Throwable actual = unwrap(ex);
logIfNeeded(actual);
resultFuture.completeExceptionally(actual);
futures.forEach(f -> {
boolean cancelled = f.cancel(true);
log.debug("Try to cancel task {}: {}", f, cancelled ? "Success" : "Failed");
});
}
}));
}
private static Throwable unwrap(Throwable ex) {
if (ex instanceof CompletionException || ex instanceof ExecutionException) {
return ex.getCause();
}
return ex;
}
/**
* Gracefully shutdown thread pool
*
* @param executor Thread pool
*/
public static void shutdownGracefully(ExecutorService executor) {
executor.shutdown();
try {
if (!executor.awaitTermination(1, TimeUnit.MINUTES)) {
executor.shutdownNow();
}
} catch (InterruptedException e) {
executor.shutdownNow();
Thread.currentThread().interrupt();
}
}
private static void logIfNeeded(Throwable actual) {
if (!(actual instanceof CancellationException)) {
log.warn("Task failed, start fail-fast to cancel other tasks: {} - [{}]",
actual != null ? actual.getMessage() : "null",
actual != null ? actual.getClass().getSimpleName() : "null");
}
}
public static <R> void collectAllResults(List<CompletableFuture<R>> futures,
CompletableFuture<List<R>> resultFuture) {
CompletableFuture
.allOf(futures.toArray(new CompletableFuture[0]))
.whenComplete((v, ex) -> {
if (!resultFuture.isDone()) {
try {
List<R> results = futures.stream()
.map(CompletableFuture::join)
.collect(Collectors.toList());
resultFuture.complete(results);
} catch (CompletionException e) {
resultFuture.completeExceptionally(e.getCause());
}
}
});
}
}
@Override
public void close() {
CommonTaskSupport.shutdownGracefully(executor);
}
}
Parallel Task Executor (without Return Value)
Java Parallel Task Executor (without Return Value)
package com.whalefall541.cases.concurrenttskvoid;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Parallel Task Executor
*
* <p>Executes two Runnable tasks simultaneously with timeout control.
* Implements AutoCloseable for try-with-resources automatic thread pool release.
*
* <p>Usage example:
* <pre>{@code
* try (ParallelTaskExecutor executor = new ParallelTaskExecutor()) {
* executor.execute(task1, task2, 28, TimeUnit.SECONDS);
* }
* }</pre>
*
*/
public class ParallelTaskExecutor implements AutoCloseable {
private final ExecutorService executor;
private final boolean owned;
/**
* Create executor with default thread pool (2 threads)
*
* <p>Thread pool will be managed by this class and automatically closed on close()
*/
public ParallelTaskExecutor() {
this.executor = Executors.newFixedThreadPool(2, new NamedThreadFactory("parallel-task"));
this.owned = true;
}
/**
* Create executor with external thread pool
*
* @param executor External ExecutorService, this class will not close it
*/
public ParallelTaskExecutor(ExecutorService executor) {
this.executor = executor;
this.owned = false;
}
/**
* Execute two tasks in parallel with default 28 second timeout
*
* @param step1 Task 1
* @param step2 Task 2
* @throws ParallelExecutionException If any task times out, fails, or is interrupted
*/
public void execute(Runnable step1, Runnable step2) {
execute(step1, step2, 28, TimeUnit.SECONDS);
}
/**
* Execute two tasks in parallel with custom timeout
*
* @param step1 Task 1
* @param step2 Task 2
* @param timeout Timeout duration
* @param unit Time unit
* @throws ParallelExecutionException If any task times out, fails, or is interrupted
*/
public void execute(Runnable step1, Runnable step2, long timeout, TimeUnit unit) {
CompletableFuture<Void> f1 = CompletableFuture.runAsync(step1, executor);
CompletableFuture<Void> f2 = CompletableFuture.runAsync(step2, executor);
try {
CompletableFuture.allOf(f1, f2).get(timeout, unit);
} catch (TimeoutException e) {
throw new ParallelExecutionException("parallel task timeout", e);
} catch (ExecutionException e) {
throw new ParallelExecutionException("parallel task failed", unwrap(e));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new ParallelExecutionException("parallel task interrupted", e);
} finally {
f1.cancel(true);
f2.cancel(true);
}
}
private Throwable unwrap(Throwable ex) {
while ((ex instanceof ExecutionException || ex instanceof CompletionException)
&& ex.getCause() != null) {
ex = ex.getCause();
}
return ex;
}
@Override
public void close() {
if (owned) {
shutdown();
}
}
private void shutdown() {
executor.shutdown();
try {
if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
executor.shutdownNow();
}
} catch (InterruptedException e) {
executor.shutdownNow();
Thread.currentThread().interrupt();
}
}
/**
* Parallel task execution exception
*
* <p>Wraps TimeoutException, ExecutionException, InterruptedException
*/
public static class ParallelExecutionException extends RuntimeException {
public ParallelExecutionException(String message, Throwable cause) {
super(message, cause);
}
}
/**
* Named thread factory
*
* <p>Created thread format: {prefix}-{sequence}
*/
public static class NamedThreadFactory implements ThreadFactory {
private final String prefix;
private final AtomicInteger counter = new AtomicInteger(1);
public NamedThreadFactory(String prefix) {
this.prefix = prefix;
}
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r, prefix + "-" + counter.getAndIncrement());
t.setDaemon(false);
return t;
}
}
}
Agreement
The code part of this work is licensed under Apache License 2.0 . You may freely modify and redistribute the code, and use it for commercial purposes, provided that you comply with the license. However, you are required to:
- Attribution: Retain the original author's signature and code source information in the original and derivative code.
- Preserve License: Retain the Apache 2.0 license file in the original and derivative code.
- Attribution: Give appropriate credit, provide a link to the license, and indicate if changes were made.
- NonCommercial: You may not use the material for commercial purposes. For commercial use, please contact the author.
- ShareAlike: If you remix, transform, or build upon the material, you must distribute your contributions under the same license as the original.