Java Thread Queue

Question

We have an application which processes items and on each iteration, starts a thread to do an update on an other database - it is not hugely important what happens on that other thread, it is a very straightforward update.

Our original intention was (by using a thread) to make sure the main processing is not held up by initalizing a connection to this other db and running the update.

Yesterday, we had an issue where (for a yet unknown reason) the database slowed down and the number of parallel threads went to the sky, resulting in 1000+ connections in this DB. So we realized we need more control over the threads.

I need a lib or tool for our software which can:

1) Put threads / jobs / tasks (anything - we can rewrite the code if required, we have Thread objects at the mintue) into a queue like system 2) we can define how many threads are running at most at the same time 3) After the thread finished, the thread is removed from the queue so GC can remove all the entities involved.

I was doing a fair bit of reading and i found ExecutorService (Executors.newFixedThreadPool(5);) but may problem is that it fails with 3) because according to the javadocs:

The threads in the pool will exist until it is explicitly shutdown.

Which, i believe, means that if the app keeps adding threads, the threads will exists until the application is restarted (or if i shutdown and reinstantiate the ExecutorService, but that seems to be a hack to me).

Am i right in thinking Executors.newFixedThreadPool(5) is failing at my 3) requirement? Am i actually getting the problem from the good end? Do i need Threads or something different?

Answer 1

The sentence you are afraid of:

The threads in the pool will exist until it is explicitly shutdown

describes only calls to Executors.newFixedThreadPool(). To keep more control on the thread pool behavior, use ThreadPoolExecutor constructor expicitly, e.g

new ThreadPoolExecutor(1, //minimal Pool Size, 10, // maximal Pool Size, 30, TimeUnit.SECONDS // idle thread dies in 30 seconds new ArrayBlockingQueue<Runnable>())

Answer 2

Here you need to understand the difference between Thread and the Runnable/Callable Task. so the meaning of The threads in the pool will exist until it is explicitly shutdown. is that at any point of time there will be 5 threads in the thread pool if you use Executors.newFixedThreadPool(5); . And the work that you want these threads to do would be submitted as Tasks (Runnable/Callable). So essentially at any point of time at max there will be 5 threads executing via this thread pool which in your case would be 5 connections.

Answer 3

The threads will stay there (waiting for other tasks to run), but it won't hold on to all the data that you put there. When the thread in the threadpool has executed the task, it will take a next task and won't reference the existing task anymore.

So your fears are baseless, unless you explicitly keep references to the tasks.

Answer 4

Use the ScheduledExecutorService with a fixed pool of threads for however many connections you need.

Have a BlockingQueue that you put requests in, the worker threads wait on the queue and process the requests as they appear.

Answer 5

The ExecutorService is the way to go. It provides you an interface (Future) to the state of the underlying thread, the ability to detect exceptions and return a value from the completed thread.

Here is a simple example of how to use ExecutorSerivce and the Future interface.

public class Updater implements Future< boolean > {

    public Updater() { }

    public boolean call() throws Exception {
        System.out.println( "Hello, World!" );
        return true;
    }
}

public class Main {
    public Main() { }

    public static void main( String[] args ) {
        ExecutorService pool = Executors.newFixedThreadPool( 1 );


        boolean again = true;
        do {
            if ( again ) { 
                Future< ? > update = pool.submit( new Updater() );
            }

            /* Do other work while waiting for update to finish */

            if( update.isDone() ) { //may be because of completion or an exception
                try {
                    again = update.get(); // This would block if the Updater was still running
                } catch( ExecutionException ee ) { // This is thrown by get() if an exception occurred in Updater.call()
                    again = false;
                    ee.printStackTrace();
                }
            }
        } while ( true );
    }
}

The example above will start an update to your database if the last update succeeded without an exception. This way you are controlling how many threads are trying to connect, and catching any errors that are causing the update to fail.