CODEWITHSUNDEEP
javaapache-zookeeperjava.util.concurrent
Greg
Greg

Reputation: 1719

Short lived ExecutorService for async processing

I have an method that can execute asynchronous request in fire and forget fashion.

Method is implemented as following : 

private void publishWorkItem(final Object payload, final ZkWorkCompleteCallback callback)
{
    if (payload == null)
        throw new NullPointerException();

    final ExecutorService executor = Executors.newSingleThreadExecutor(PUBLISH_WORK_THREAD_FACTORY);

    try
    {
        executor.execute(() -> {

            try
            {
                if (callback != null)
                {
                    final ZkWorkItem retval = publishWorkItem(payload);
                    callback.onCompleted(retval);
                }
            }
            catch (final InterruptedException e)
            {
                // suppressed
            }
            catch (final Exception e)
            {
                LOGGER.error("Unhandled exception", e);

                if (callback != null)
                    callback.onError(e);
            }
        });
    }
    finally
    {
        executor.shutdown();
    }
}

Issue is that I am creating new ExecutorService Executors.newSingleThreadExecutor for each async request instead of using fixed thread pool. Reason for that is that publishWorkItem(payload) method uses a CountDownLatch#await() which in turn will block the executing thread because is waits for Watcher to finish. This could quickly exhaust fixed size pool.

Simplified code of publishWorkItem(payload)

  final CountDownLatch latch = new CountDownLatch(1);

        zkClient.exists(pathToWatch, new Watcher()
        {
            @Override
            public void process(final WatchedEvent event)
            {
                try
                {
                    extractAndDelete(baos, event.getPath());
                }
                catch (final Exception e)
                {
                    LOGGER.error("Unable to perform cleanup", e);
                }
                finally
                {
                    latch.countDown();
                }
            }
        }, true);

       ------ THIS IS THE PROBLEM (Blocks current thread) ------ 
       latch.await();

So my question is: Are there better approaches to this type of problem.

I did profile the application and I don't see any performance issues, my concern was that it was creating large number of threads.

Upvotes: 0

Views: 665

Answers (1)

noscreenname
noscreenname

Reputation: 3370

Why don't you use a ExecutorService.newCachedThreadPool()?

According to the javadoc, it suits your use-case

These pools will typically improve the performance of programs that execute many short-lived asynchronous tasks ... will reuse previously constructed threads if available

Instead of creating a new single thread pool on each call of publishWorkItem(), you create a cached thread pool once and use for all your queries. The number of threads is capped by Integer.MAX_VALUE, so you will not be limited like with fixed thread pool, but it should be creating less threads overall.

Upvotes: 1

Related Questions