How to know when a CompletionService is finished delivering results?

See http://www.javaspecialists.eu/archive/Issue214.html for a decent suggestion on how to extend the ExecutorCompletionService to do what you're looking for. I've pasted the relevant code below for your convenience. The author also suggests making the service implement Iterable, which I think would be a good idea.

FWIW, I agree with you that this really should be part of the standard implementation, but alas, it's not.

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

public class CountingCompletionService<V> extends ExecutorCompletionService<V> {
  private final AtomicLong submittedTasks = new AtomicLong();
  private final AtomicLong completedTasks = new AtomicLong();

  public CountingCompletionService(Executor executor) {
    super(executor);
  }

  public CountingCompletionService(
      Executor executor, BlockingQueue<Future<V>> queue) {
    super(executor, queue);
  }

  public Future<V> submit(Callable<V> task) {
    Future<V> future = super.submit(task);
    submittedTasks.incrementAndGet();
    return future;
  }

  public Future<V> submit(Runnable task, V result) {
    Future<V> future = super.submit(task, result);
    submittedTasks.incrementAndGet();
    return future;
  }

  public Future<V> take() throws InterruptedException {
    Future<V> future = super.take();
    completedTasks.incrementAndGet();
    return future;
  }

  public Future<V> poll() {
    Future<V> future = super.poll();
    if (future != null) completedTasks.incrementAndGet();
    return future;
  }

  public Future<V> poll(long timeout, TimeUnit unit)
      throws InterruptedException {
    Future<V> future = super.poll(timeout, unit);
    if (future != null) completedTasks.incrementAndGet();
    return future;
  }

  public long getNumberOfCompletedTasks() {
    return completedTasks.get();
  }

  public long getNumberOfSubmittedTasks() {
    return submittedTasks.get();
  }

  public boolean hasUncompletedTasks() {
    return completedTasks.get() < submittedTasks.get();
  }
}

Answering to these questions gives you the answer?

• Do your asynchronous tasks create other tasks submitted to CompletionService?
• Is service the only object that is supposed to handle the tasks created in your application?

Based on reference documentation, CompletionService acts upon a consumer/producer approach and takes advantage of an internal Executor. So, as long as, you produce the tasks in one place and consume them in another place, CompletionService.take() will denote if there are any more results to give out.

I believe this question also helps you.

The code below is inspired by @Mark's answer, but I find it more convenient to use:

package com.example;

import java.util.Iterator;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class CompletionIterator<T> implements Iterator<T>, AutoCloseable {

    private AtomicInteger count = new AtomicInteger(0);

    private CompletionService<T> completer;

    private ExecutorService executor = Executors.newWorkStealingPool(100);

    public CompletionIterator() {
        this.completer = new ExecutorCompletionService<>(executor);
    }

    public void submit(Callable<T> task) {
        completer.submit(task);
        count.incrementAndGet();
      }

    @Override
    public boolean hasNext() {
        return count.decrementAndGet() > 0;
    }

    @Override
    public T next() {
        try {
            return completer.take().get();
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void close() {
        try {
            executor.shutdown();
            executor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
            executor = null;
            completer = null;
            count = null;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

}

This is how it can be used :

try(CompletionIterator service = new CompletionIterator()) {
  service.submit(task1);
  service.submit(task2);
  // all tasks must be submitted before iterating, to avoid race condition
  for (Future<Integer> future : service) {
    System.out.printf("Job %d is done%n", future.get());
  }
}