A Lazy Developers Introduction to Java Concurrency Executors

I would make a fool out of myself if I tell you that util.concurrent APIs kicks cheetah’s ass when the classes are available since 2004. However, there are some cool features which I would like to revisit. Concurrency experts, now is the time for you to close this window. All others, stay tight for the fun ride.

Thou shall not forget your roots

Executor is the root interface with a single execute method. Anything that implements a Runnable interface can passed as a parameter. Silly Executor, however, has no support for Callable though.

Good news : ExecutorService interface, which extends Executor, adds support for Callable. Its implementation class is ThreadPoolExecutor.

I am going to pretend that the ScheduledExecutorService interface and its implementation class ScheduledThreadPoolExecutor does not exist as they just add scheduling capabilities on top of the ExecutorService and ThreadPoolExecutor. But remember this class when the powerful but boring java.util.Timer is not enough and a full blown external scheduler is just too much.

If you are new to concurrency or forgot the difference between Callable and Runnable, you might want to read up a little before reading further. A dummy guide is here

The ExecutorService.submit Facts :

The three submit variants :

Future submit(Callable task)
Future submit(Runnable task)
Future submit(Runnable task, T result)

1. The submit method of the ExecutorService is overloaded and accepts either a Callable or Runnable.
2. Since the run method of the Runnable returns void, it is no surprise that Future.get always returns a null when the task is complete.

   Future<?>   submit(Runnable task)

    
3. The other overloaded submit method that accepts a Runnable and a generic returns whatever you passed in as the second parameter as the result.

    <T> Future<T>   submit(Runnable task, T result)

    

In fact, opening up the code (FutureTask), you’ll notice that the RunnableAdapter top level nested class of Executors simply holds the result and returns the same result after the run method is complete.

static final class RunnableAdapter<T> implements Callable<T> {

     final Runnable task;
final T result;

RunnableAdapter(Runnable task, T result) {
         this.task = task;
         this.result = result;
        }

        public T [More ...] call() {
           task.run();
           return result;
        }

}

RunnableAdapter source

In both the cases, if you would like to (you should !) terminate the program instead of your executor thread blocking the program and entering a busy loop, you should call the shutdown method as in

executorService.shutdown()

shutDown facts

You could imagine shutdown as a half-closed door of a mall. No new customers will be let in but the existing customers can leave the mall once they are done.

To reiterate,

1. shutdown is a polite method. It does not actually shut down the tasks in the pool immediately. It just says that no new tasks will be accepted.
2. Unless you are executing your tasks using invokeAll, you would need to wait for all tasks in progress to complete. This is achieved by calling the awaitTermination method. (invokeAll and submit examples at the bottom of the post)
3. Once all the current tasks are done, the executor service shuts down.

If you are in need of an impolite, intruding method which doesn’t care whether the current threads are done with their tasks, then shutdownNow is your guy. However, there’s no guarantee that the method will shutdown the service on the dot but it is the closest you have to immediate shutdown.

On awaitTermination, you could specify the timeout period until which the main thread should wait for the pool threads to complete its tasks.

ExecutorService executorService=Executors.newFixedThreadPool(10);
    …
    future = executorService.submit(getInstanceOfCallable(count,sum));
    …
    executorService.shutdown();

    if (executorService.awaitTermination(10, TimeUnit.SECONDS)){
        System.out.println("All threads done with their jobs");
    }

Executors - the factory guy

The classes above are all awesome. But, say, you wanted to create a single thread executor, you would write something like

new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());

Compare that to

Executors.newSingleThreadExecutor()

So, here you go. Executors is a class with just factory methods for creating various forms of executor service with some commonly used defaults. Note that, other than the awesome factory methods, it doesn’t bring any new features to the table.

It is recommended that you have a quick look at the implementation of the factory methods and check if it suits your needs.

The invokeAll and the submit

The All part of invokeAll method of the ExecutorService gives no surprise. It just says that you need to pass in a Collection of Callables. Again, as expected, the method does not return until all the Threads are done with their tasks. So, for cases when you are interested in the result only after all the jobs are complete, invokeAll is your guy.

On the other hand, the submit method returns immediately after the callable is submitted to the executor service. Unless you are doing nothing at all in your call method of your Callable, your worker threads should ideally be running when the submit method returns.

The following samples might be useful for your reference. The programs just tries to find the sum of all the natural numbers till 100 (Brute force, of course)

package me.rerun.incubator;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

public class ExecutorInvokeAll {

public void runApp() throws InterruptedException, ExecutionException{

                //variable to store the sum
AtomicInteger sum=new AtomicInteger();

                //Use our friendly neighbourhood factory method of the Executors.
ExecutorService executorService=Executors.newFixedThreadPool(10);

List<Callable<AtomicInteger>> callableList=new ArrayList<Callable<AtomicInteger>>();

for (int count = 0; count <= 100;count++) {
callableList.add(getInstanceOfCallable(count,sum));

}

                //returns only after all tasks are complete
List<Future<AtomicInteger>> resultFuture = executorService.invokeAll(callableList);

//Prints 5050 all through
for (Future<AtomicInteger> future : resultFuture) {
//Didn't deliberately put a timeout here for the get method. Remember, the invoke All does not return until the task is done.
System.out.println("Status of future : " + future.isDone() +". Result of future : "+future.get().get());
}


executorService.shutdown();

// You might as well call a resultFuture.get(0).get().get() and that would give you the same
//result since all your worker threads hold reference to the same atomicinteger sum.
System.out.println("Final Sum : "+sum);

}
//Adds count to the sum and returns the reference of the sum as the result
private Callable<AtomicInteger> getInstanceOfCallable(final int count, final AtomicInteger sum) {

Callable<AtomicInteger> clientPlanCall=new Callable<AtomicInteger>(){
public AtomicInteger call() {
sum.addAndGet(count);
System.out.println("Intermediate sum :"+sum);
return sum;
}
};

return clientPlanCall;
}

public static void main(String[] args) throws ExecutionException {

try {
new ExecutorInvokeAll().runApp();
} catch (InterruptedException e) {
e.printStackTrace();
}
}

}

package me.rerun.incubator;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

public class ExecutorSubmit {

public void runApp() throws InterruptedException, ExecutionException{

                //holder for the total sum
AtomicInteger sum=new AtomicInteger();

                //Use the factory method of Executors
ExecutorService executorService=Executors.newFixedThreadPool(10);

Future<AtomicInteger> future = null;

for (int count = 0; count <= 100; count++) {
future = executorService.submit(getInstanceOfCallable(count,sum));
//prints intermediate sum
try {
System.out.println("Status of future : " + future.isDone() +". Result of future : "+future.get(1000, TimeUnit.MILLISECONDS).get());
} catch (TimeoutException e) {
System.out.println("<IGNORE> Timeout exception for count : "+count);
//e.printStackTrace();
}
//System.out.println("Result of future : "+future.get().get() +".Status of future : " + future.isDone());
}


executorService.shutdown();

if (executorService.awaitTermination(10, TimeUnit.SECONDS)){
System.out.println("All threads done with their jobs");
}
//exec
System.out.println("Final Sum : "+sum);

}

        //Adds count to the sum and returns the reference of the sum as the result
private Callable<AtomicInteger> getInstanceOfCallable(final int count, final AtomicInteger sum) {



Callable<AtomicInteger> clientPlanCall=new Callable<AtomicInteger>(){
public AtomicInteger call() {
sum.addAndGet(count);
//System.out.println("Intermediate sum :"+sum);
return sum;
}
};

return clientPlanCall;
}

public static void main(String[] args) throws ExecutionException {

try {
new ExecutorSubmit().runApp();
} catch (InterruptedException e) {
e.printStackTrace();
}
}

}

Further reading :

Alex Miller’s amazing blog

Alex Miller’s concurrency gotchas

Vogella’s article on comparison with original API

Good introduction to concurrency in general

Highly recommended book on Java concurrency