The following way of iterating a list of future always wait for the first job to be done:
for (Future<MyFutureResult> future : list) {
List<MyFutureResult> result = future.get();
}
Is there a way to iterate all the finish job first?
The following way of iterating a list of future always wait for the first job to be done:
for (Future<MyFutureResult> future : list) {
List<MyFutureResult> result = future.get();
}
Is there a way to iterate all the finish job first?
Getting the first completed Future from the list of futures is not possible directly since those are processed in parallel and you would have to block anwyay to find the result.
However you could have control over task completion by using ExecutorsCompletionService for your parallel processing. This class has take and poll methods that return Future of next completed task :
A
CompletionServicethat uses a suppliedExecutorto execute tasks. This class arranges that submitted tasks are, upon completion, placed on a queue accessible usingtake. The class is lightweight enough to be suitable for transient use when processing groups of tasks.
ExecutorService threadPool = Executors.newCachedThreadPool();
CompletionService<Integer> ecs = new ExecutorCompletionService<>(threadPool);
int tasks = 10;
IntStream.range(0, tasks)
.forEach(i -> ecs.submit(() -> i)); // submit tasks
for(int i = 0; i < tasks; i++) {
Future<Integer> take = ecs.take(); // this is blocking operation but futures are returned in completion order. Also you will have to handle InterruptedException
}
// remember to close the ExecutorService after you are done
Have a look into ExecutorService.invokeAny(..) that returns the first result, or ExecutorService.invokeAll(..) that returns all completed tasks (within a timeout).
class InvokeAnyAllTest {
ExecutorService es = Executors.newCachedThreadPool();
// create some Callable tasks
List<Callable<MyFutureResult>> tasks = IntStream.range(0, 10) //
.mapToObj(this::createTask)
.collect(toList());
private Callable<MyFutureResult> createTask(int i) {
return () -> new MyFutureResult(i);
}
@Test
void testFirstCallable() throws Exception {
MyFutureResult result = es.invokeAny(tasks);
assertTrue(result.i >= 0 && result.i < 10);
}
@Test
void testAllCompleted() throws Exception {
List<Future<MyFutureResult>> results = es.invokeAll(tasks, 5, TimeUnit.SECONDS);
// all futures that are done within 5s, either normally or by throwing an exception
Set<Integer> values = results.stream().map(f -> {
try {
return f.get();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} catch (ExecutionException e) {
// are we interested in failed ones too?
}
return null;
}).filter(Objects::nonNull).map(result -> result.i).collect(toSet());
IntStream.range(0, 10).forEach(i -> assertTrue(values.contains(i)));
}
// in case we have only futures, not callables
@Test
void testFirstFuture() throws Exception {
// create futures
List<Future<MyFutureResult>> futures = IntStream.range(0, 10) //
.mapToObj(i -> es.submit(createTask(i)))
.collect(toList());
// turn futures into callables
List<Callable<MyFutureResult>> callables = futures.stream()
.map(f -> (Callable<MyFutureResult>) () -> f.get())
.collect(toList());
MyFutureResult result = es.invokeAny(callables);
assertTrue(result.i >= 0 && result.i < 10);
}
private static class MyFutureResult {
int i;
public MyFutureResult(int i) {
this.i = i;
}
}
}
If by any reason you are not supposed to use ExecutorsCompletionService (as mentioned by Michał Krzywański). Then you can replace ecs.take() or future.get() with below method.
getCompletedFuture(futureSet, 1000).get();
.
.
.
private static Future<V> getCompletedFuture(Set<Future<V>> futureSet, long pollInterval)
throws ExecutionException, InterruptedException {
Iterator<Future<V>> iterator = futureSet.iterator();
while (!Thread.currentThread().isInterrupted()) {
if (!iterator.hasNext()) {
iterator = futureSet.iterator();
}
try {
V v = iterator.next().get(pollInterval, TimeUnit.MILLISECONDS);
if (v != null) {
iterator.remove();
return iterator.next();
}
} catch (TimeoutException e) {
continue;
}
}
throw new InterruptedException();
}
You can put the futures onto a BlockingQueue in order of their completion.
public static <T> BlockingQueue<CompletableFuture<T>> collect(Stream<CompletableFuture<T>> futures) {
var queue = new LinkedBlockingQueue<CompletableFuture<T>>();
futures.forEach(future ->
future.handle((success, failure) -> queue.add(future)));
return queue;
}
Each call to BlockingQueue.take on the queue returned by collect will block until the next future become available by completing and return that future.