Observer pattern with threads

Question

I want to start several threads and join them at the end of my main method, so I can find out when they finished and processed some information.

I don’t want to put my threads into an array and do join () one by one, since joining is a locking method, and I remain waiting in the main thread for some threads still working, while other threads may have already finished not being able to know.

I thought about the possibility of implementing an observer pattern for my threads: an interface with the update () method, an abstract class extending from the stream (or an executable executable) using the set and get methods for listeners and starting the class all my topics and waiting for them to complete.

If my understanding is correct, the observer will not block the specific connection () for the stream. Instead, it will wait one way or another until the update () method is called by the thread to complete the action. In this case, update () should be called immediately after the stream ends.

I do not know how to implement this. I tried with similar models, but I don't know how to use the observer / listener to wake / block my main thread. I used this old post as a template: How do I know if other threads have ended? , but I can’t find a way to wake up my main method after the thread calls the update () method. Only one observer object will be created for all threads.

Could you please think about how to use the observer pattern to wait for the completion of all threads without blocking the main one call join ()? Any other suggestion to solve this problem would be very helpful. Thanks in advance.

Answer 1

I think you do not need an observer pattern. A thread waiting for any results will have to block, otherwise it will end or loop in infinity. You can use some kind of BlockingQueue - manufacturers will add the result of the calculation to the lock queue (then finish), and the main thread will just get these result locks when there is no result.

Good news for you, this has already been implemented :) Excellent CompletionService mechanism and Executors framework . Try the following:

 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 for this: CompletionService .

A service that separates the production of new asynchronous tasks from the consumption of the results of completed tasks. Producers send assignments for execution. Consumers fulfill their tasks and process their results in the order they are filled.

Answer 3

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