Is there a clean/recommended way to create an executorService that maintains a pool of threads that poll from a shared BlockingQueue?

[Seulgi Kim]

Hi,

I want to make an ExecutorService that maintains a pool of threads that poll from a shared BlockingQueue.

My original approach was to implement a class that implements Managed class (let's call the class MyClass).

MyClass has a reference to ExecutorService myExecutorService, and a reference to AtomicBoolean isStopRequested, which all Threads that have been submitted to myExecutorService has a reference to.

run() method of each Thread looks like below

public void run() {

 while(!isStopRequested.get()) {

    /// run some code

 }

}

stop and stop method of MyClass (that implements Managed) looks as below

@Override

   public void start() throws Exception {

       isStopRequested.set(false);

       for (int i = 0; i < threadPoolSize; i++) {

           final MyThread myThread = MyThread(isStopRequested, sharedBlockingQueue... );

           myExecutorService.execute(myThread);

       }

   }

@Override

   public void stop() throws Exception {

       isStopRequested.set(true);

        waitForSharedQueueToDrain(); // a method that waits until a shared BlockingQueue gets drained

        executorService.shutdown();

       executorService.awaitTermination(shutdownTimeout, TimeUnit.SECONDS);

   }

I realize that DropWizard provides an ExecutorService that is already managed by Dropwizard. ( environment.lifecycle().executorService()....).

It would be nice to use the executorService provided by Dropwizard, but I started to have difficulty to gracefully shutdown my pool of threads if I used the provided ExecutorService. 

The biggest problem was how to tell each Thread to continue polling. 

Has anybody been able to use an ExecutorService provided by Dropwizard that maintains a pool of threads that poll from a shared BlockingQueue?

Thank you! :-)

 

[arte...@gmail.com]

Hi,

I believe you do it wrong. If you explicetely poll from the shared queue, why do nee *ExecutorService* at all? Just start a pool of threads and send interrupts to them when you need to stop.
In your situation graceful shutdown would be hard to implement, because you need somehow to block shared queue for writing while consumers finish the work.
But if you can - block the queue, wait some time and then send interrupts.

Artem

[Seulgi Kim]

Thanks for your reply, Artem.

I thought Java provided ExecutorService so that users do not have to implement thread-pool on their own (http://docs.oracle.com/javase/tutorial/essential/concurrency/pools.html)

Are you suggesting putting Threads in some kind of Collection (e.g. ImmutableList<Thread> or Set<Thread>) instead?

[arte...@gmail.com]

ExecutorService is an abstraction over BlockingQueue and thread pool. It works best in assumption that you pass tasks to it as Runnable or Callable by execute, submit or invoke methods. In this case you get benefits like graceful shutdown, rejection policy and etc. If you don't pass tasks to it, there little profit to use it, but only additional overhead.
Yes, I think a list of pre-created threads is a quite simple and an easily managed solution.
 
Artem

[Seulgi Kim]

I've decided to have MyClass implement Runnable and Managed, and pass that instances of the class to my ExecutorService. I use a volatile boolean shouldRun to indicate if an instance of MyClass, and set sholdRun = false in MyClass#stop. This gave a result I wanted.

Thanks for your reply, Artem! :-)