What is the way to get the first finish future from a list of future in java?

Viewed 1214

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?

4 Answers

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 CompletionService that uses a supplied Executor to execute tasks. This class arranges that submitted tasks are, upon completion, placed on a queue accessible using take. 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.

Related