I want to execute N tasks in parallel such that no individual task should run for more than 2 seconds (we can mark such tasks as failed). As an output i want to return output of successful tasks and status of failed tasks as failed. Also timeout of 1 task should not lead to circuit break, i.e other tasks execution should not stop.
NOTE: I am restricted to use JAVA 8.
I referenced this article for parallel processing. I am doing similar kind of parallel processing as given in example in this article:
public void parallelProcessing() {
try {
ExecutorService executorService = Executors.newWorkStealingPool(10);
List<CompletableFuture<Integer>> futuresList = new ArrayList<CompletableFuture<Integer>>();
futuresList.add(CompletableFuture.supplyAsync(()->(addFun1(10,5)), executorService));
futuresList.add(CompletableFuture.supplyAsync(()->(subFun1(10,5)), executorService));
futuresList.add(CompletableFuture.supplyAsync(()->(mulFun1(10,5)), executorService));
CompletableFuture<Void> allFutures = CompletableFuture.allOf(futuresList.toArray(new CompletableFuture[futuresList.size()]));
CompletableFuture<List<Integer>> allCompletableFuture = allFutures.thenApply(future -> futuresList.stream().map(completableFuture -> completableFuture.join())
.collect(Collectors.toList()));
CompletableFuture<List<Integer>> completableFuture = allCompletableFuture.toCompletableFuture();
List<Integer> finalList = (List<Integer>) completableFuture.get();
} catch (Exception ex) {
}
}
public static Integer addFun1(int a, int b) {
System.out.println(Thread.currentThread().getName());
for (int i = 0; i < 10; i++) {
System.out.print(Thread.currentThread().getName() + i);
}
return a + b;
}
public static Integer subFun1(int a, int b) {
System.out.println(Thread.currentThread().getName());
for (int i = 0; i < 10; i++) {
System.out.print(Thread.currentThread().getName() + i);
}
return a - b;
}
public static Integer mulFun1(int a, int b) {
System.out.println(Thread.currentThread().getName());
for (int i = 0; i < 10; i++) {
System.out.print(Thread.currentThread().getName() + i);
}
return a * b;
}
This works fine. But I want to set timeout for individual thread. I know I can use overloaded get function in last line. But that would set the timeout for combined futures , right? E.g. if I want no individual thread should be blocked for more than 2 sec, and if I set 2 sec timeout in the last line, it will be combined timeout, right ?
get(long timeout, TimeUnit unit)
Here's what I want to achieve as a final outcome:
Suppose there are 5 threads and 4 complete on time, 1 timeouts (due to running more than 2 secs). In this case I want to send output of 4 threads and send error for 5th thread in result. My input output format is in following way:
Sample input : List<Input> each item is run in separate thread, where each Input has a uniqueIdentifier.
Sample output : List<Output> such that->
Output :{
uniqueIdentifier: //same as input to map for which input this output was generated
result: success/fail // this Field I want to add. currently it's not there
data: {
// from output e.g. addFun1 , subFun1
}
}
Supposed you want a 10 threads running and want a returned value, you can use Callable<Boolean> interface and submit it to ExecutorService and then get the result using Future#get returned Boolean.
Here is an example usage.
final int NUM_THREADS=10;
List<Boolean> results=new ArrayList<Boolean>();
List<Callable<Boolean>> callables=new ArrayList<Callable<Boolean>>();
for(int i=0;i<NUM_THREADS;++i)
{
callables.add(new Callable<Boolean>()
{
public Boolean call()
{
//add your task here
return isTaskCompleted;
}
});
}
ExecutorService executorService=ExecutorService.newFixedThreadPool(NUM_THREADS); //run 10 threads
for(Callable<Boolean> callable:callables)
{
Future<Boolean> future=executor.submit(callable);
try
{
results.add(future.get(2, TimeUnit.SECONDS)); // timeout 2 seconds and add the result
}
catch(Exception ex)
{
results.add(false); //set result to false if task throw TimeOutExeption
}
}
If you want more info about these classes you can read this book: O'Reilly - Learning Java Chapter 9:Threads.
The following is a single-file mre (paste the entire code into RunParallelTasks.java and run). It is a prototype of the structure I suggested in my comment aimed to achieve the required functionality by using simple means:
import java.util.Optional;
public class RunParallelTasks {
public static void main(String[] args) {
new Thread(()->{
long duration = 3000;
Callback<Long> cb = new LongTask(duration);
Output<Long> output = new TaskExecuter<Long>().work(cb);
System.out.println( output);
}).start();
new Thread(()->{
long duration = 300;
Callback<Long> cb = new LongTask(duration);
Output<Long> output = new TaskExecuter<Long>().work(cb);
System.out.println( output);
}).start();
new Thread(()->{
long duration = 4000;
Callback<Long> cb = new LongTask(duration);
Output<Long> output = new TaskExecuter<Long>().work(cb);
System.out.println( output);
}).start();
new Thread(()->{
long duration = 1000;
Callback<Long> cb = new LongTask(duration);
Output<Long> output = new TaskExecuter<Long>().work(cb);
System.out.println( output);
}).start();
}
}
class TaskExecuter<T>{
private static final long TIMEOUT = 2000;//millis
private T value = null;
public Output<T> work(Callback<T> call){
Thread t = new Thread(()->{
value = call.work();
});
t.start();
try {
t.join(TIMEOUT);
} catch (InterruptedException ex) {
ex.printStackTrace();
}
return new Output<>(t.getId(), value == null ? Optional.empty() : Optional.of(value)) ;
}
}
interface Callback<T> {
T work();
}
class LongTask implements Callback<Long>{
private final long durationInMillis;
public LongTask(long durationInMillis) {
this.durationInMillis = durationInMillis;
}
@Override
public Long work() {
try {
Thread.sleep(durationInMillis);
} catch (InterruptedException ex) {
ex.printStackTrace();
}
return durationInMillis;
}
}
class Output<T> {
private final long id;
private boolean success = false;
private T data;
public Output(long id, Optional<T> op) {
this.id = id;
if(!op.isEmpty()) {
data = op.get();
success = true;
}
}
//todo add getters
@Override
public String toString() {
return "task "+ id+ (success ? " Completed, returned "+data : " Failed" );
}
}
We had similar requirement where we need capture timeout of each thread and ignore the results. Java 8 doesn't have this in built. One of the ways we achieved it,
List<CompletableFuture<?>> futures = new ArrayList<>();
List<?> results = new ArrayList<>(); // It can be anything you collect
futures.add(asyncService.fetchMethod()
.acceptEither(
timeoutAfter(timeout, TimeUnit.SECONDS),
results:add)
.handle(
(result, ex) -> {
//Handle the timeout exception
results.add(...);
return result
});
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
private <T> CompletableFuture<T> timeoutAfter(long timeout, TimeUnit unit) {
CompletableFuture<T> result = new CompletableFuture<>();
// We need a separate executor here
scheduledExecutor.schedule(
() -> result.completeExceptionally(new TimeoutException()), timeout, unit);
);
return result;
}