Building a controlled java threading queue

Question

Here is my situation. I have to run X searches (between 10-200) getting the results of each and appending them. I want to add some concurrency to the searches but I want to be able to throttle it. What I mean is I dont want to kick of 200 threads and wait for all to complete. I want to kickoff N threads and as each one completes start a new thread until all have completed.

I know I could write this from scratch but I was wondering if a simple pattern or library already exists for this? I hate re-creating the wheel.

Thanks.

Answer 1

The executor framework in java.util.Concurrent is a standard way of creating thread pools and scheduling tasks on them. You'd create a java.util.concurrent.ThreadPoolExecutor with a number of threads and submit Runnable or Callable<V> instances on them (Callable is like runnable, but with the ability to return a result). Each time you submit a task, you get a Future<V> instance in return, which can be thought of as a pointer to a future result. You can query these future instances on whether they've completed or simply do a blocking wait using the get() method.

Answer 2

Design your program so each individual task is a Callable. You can then choose the Executor suiting you best from the Executors class.

Answer 3

Look at CompletionService and a TimingThreadPool from java.util.concurrent e.g.

private final Executor exec = new TimingThreadPool( NUMBER_MIN_THREADS, 2, 10, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>(NUMBER_IN_QUEUE ),
            new ThreadPoolExecutor.CallerRunsPolicy()   );
private final CompletionService<DownloadStatus> service = new ExecutorCompletionService<DownloadStatus>( exec );

Then for each search

service.submit( <a callable doing the search>);

say there are N of them

When all sent the wait for the results on the service queue

for (int i = 0; i < N); i++)
{
    final Future<DownloadStatus> fut = service.take();
    final Object d = fut.get(); // This is what the callable returns 

Answer 4

I think I would use Executors.newfixedThreadPool(int numberOfThreads) which does what you want, except for the fact that it re-uses threads in stead of creating new ones.

You give it a Runnable (or a Callable if you want the result and/or exceptions back) and it will execute that runnable in one of the available threads as soon as a thread comes free.

You can, as said, obtain the result by using a Callable, but you could also use a Runnable and dump the result in e.g. a BlockingQueue like a LinkedBlockingQueue. That way you can chomp away at the results as soon as they get produced by your thread.