Observer pattern with threads

Question

I want to run several threads and join them at the end of my main method, so I can know when they have finished and process some info.

I don't want to put my threads in an array and do a join() one by one as join is a blocking method and I stay would waiting in the main thread for some threads still running, while other threads may have already finished, without having a possibility of knowing.

I have thought on the possibility of implementing an observer pattern for my threads: An interface with a update() method, an abstract class extending from thread (or implementing runnable) with set and get methods for the listeners and a class starting all my threads and waiting them to finish.

If my understanding is right, an observer would not block in a specific join() for a thread. Instead it will wait somehow until an update() method is called by a thread to perform an action. In this case, the update() should be called right after the thread finishes.

I'm clueless on how to implement this. I've tried with similar models, but I don't know how to use the observer/listener to wake/block my main thread. I've used this old post as a template: How to know if other threads have finished? but I can't find a way to wake my main method once a thread calls the update() method. There will be only one observer object instantiated for all threads.

Could you think of a way to use an observer pattern to wait for all threads to finish without blocking main with one by one join() calls? Any other suggestion to solve this problem would be greatly appreciated. Thanks in advance.

Answer 1

I think you don't need an observer pattern. Thread waiting for any results will have to block, otherwise it will finish or loop in infinity. You can use some kind of BlockingQueue - producers will add result of computation to the blocking queue (then finish) and main thread will just receive these results blocking when there's not any result yet..

Good news for you, it's already implemented :) Great mechanism of CompletionService and Executors framework. Try this:

private static final int NTHREADS = 5;
private static final int NTASKS = 100;
private static final ExecutorService exec = Executors.newFixedThreadPool(NTHREADS);

public static void main(String[] args) throws InterruptedException {
    final CompletionService<Long> ecs = new ExecutorCompletionService<Long>(exec);
    for (final int i = 0; i < NTASKS ; ++i) {
        Callable<Long> task = new Callable<Long>() {
            @Override
            public Long call() throws Exception {
                return i;
            }
        };
        ecs.submit(task);
    }
    for (int i = 0; i < NTASKS; ++i) {
        try {
            long l = ecs.take().get();
            System.out.print(l);
        } catch (ExecutionException e) {
            e.getCause().printStackTrace();
        }
    }
    exec.shutdownNow();
    exec.awaitTermination(50, TimeUnit.MILLISECONDS);
}

Answer 2

Java already has an API to do that: a CompletionService.

A service that decouples the production of new asynchronous tasks from the consumption of the results of completed tasks. Producers submit tasks for execution. Consumers take completed tasks and process their results in the order they complete.

Answer 3

Sounds to me like you are looking for something like the Counting Completion Service recently discussed by Dr. Heinz M. Kabutz.