Using an Executor

Using an Executor

The concurrent API supplies a feature called an executor that initiates and controls the execution of threads. As such, an executor offers an alternative to managing threads through the Thread class.

At the core of an executor is the Executor interface. It defines the following method: void execute(Runnable thread)

The thread specified by thread is executed. Thus, execute( ) starts the specified thread. The ExecutorService interface extends Executor by adding methods that help manage

and control the execution of threads. For example, ExecutorService defines shutdown( ), shown here, which stops the invoking ExecutorService.

void shutdown( )

ExecutorService also defines methods that execute threads that return results, that execute a set of threads, and that determine the shutdown status. We will look at several of these methods a little later.

Also defined is the interface ScheduledExecutorService, which extends ExecutorService to support the scheduling of threads.

The concurrent API defines three predefined executor classes: ThreadPoolExecutor and ScheduledThreadPoolExecutor, and ForkJoinPool. ThreadPoolExecutor implements the Executor and ExecutorService interfaces and provides support for a managed pool of threads. ScheduledThreadPoolExecutor also implements the ScheduledExecutorService interface to allow a pool of threads to be scheduled. ForkJoinPool implements the Executor and ExecutorService interfaces and is used by the Fork/Join Framework. It is described later in this chapter.

A thread pool provides a set of threads that is used to execute various tasks. Instead of each task using its own thread, the threads in the pool are used. This reduces the overhead associated with creating many separate threads. Although you can use ThreadPoolExecutor and ScheduledThreadPoolExecutor directly, most often you will want to obtain an executor by calling one of the following static factory methods defined by the Executors utility class. Here are some examples:

static ExecutorService newCachedThreadPool( )

static ExecutorService newFixedThreadPool(int numThreads)

static ScheduledExecutorService newScheduledThreadPool(int numThreads)

newCachedThreadPool( ) creates a thread pool that adds threads as needed but reuses threads if possible. newFixedThreadPool( ) creates a thread pool that consists of a specified number of threads. newScheduledThreadPool( ) creates a thread pool that supports thread scheduling. Each returns a reference to an ExecutorService that can be used to manage the pool.

A Simple Executor Example

Before going any further, a simple example that uses an executor will be of value. The following program creates a fixed thread pool that contains two threads. It then uses that pool to execute four tasks. Thus, four tasks share the two threads that are in the pool. After the tasks finish, the pool is shut down and the program ends.

// A simple example that uses an Executor.

import java.util.concurrent.*;

class SimpExec {

public static void main(String args[]) {

CountDownLatch cdl = new CountDownLatch(5);

CountDownLatch cdl2 = new CountDownLatch(5);

CountDownLatch cdl3 = new CountDownLatch(5);

CountDownLatch cdl4 = new CountDownLatch(5);

ExecutorService es = Executors.newFixedThreadPool(2);

System.out.println("Starting");

// Start the threads.

es.execute(new MyThread(cdl, "A")); es.execute(new MyThread(cdl2, "B")); es.execute(new MyThread(cdl3, "C")); es.execute(new MyThread(cdl4, "D"));

try { cdl.await(); cdl2.await(); cdl3.await(); cdl4.await();

} catch (InterruptedException exc) { System.out.println(exc);

}

es.shutdown();

System.out.println("Done");

}

}

class MyThread implements Runnable { String name;

CountDownLatch latch;

MyThread(CountDownLatch c, String n) { latch = c;

name = n;

new Thread(this);

}

public void run() {

for(int i = 0; i < 5; i++) { System.out.println(name + ": " + i); latch.countDown();

}

}

}

The output from the program is shown here. (The precise order in which the threads execute may vary.)

Starting

     0

     1

     2

     3

     4

     0

     1

     2

     3

     4

     0

     1

     2

     3

     4

     0

     1

     2

     3

     4 Done

As the output shows, even though the thread pool contains only two threads, all four tasks are still executed. However, only two can run at the same time. The others must wait until one of the pooled threads is available for use.

The call to shutdown( ) is important. If it were not present in the program, then the program would not terminate because the executor would remain active. To try this for yourself, simply comment out the call to shutdown( ) and observe the result.

Using Callable and Future

One of the most interesting features of the concurrent API is the Callable interface. This interface represents a thread that returns a value. An application can use Callable objects to compute results that are then returned to the invoking thread. This is a powerful mechanism because it facilitates the coding of many types of numerical computations in which partial results are computed simultaneously. It can also be used to run a thread that returns a status code that indicates the successful completion of the thread.

Callable is a generic interface that is defined like this: interface Callable<V>

Here, V indicates the type of data returned by the task. Callable defines only one method, call( ), which is shown here:

V call( ) throws Exception

Inside call( ), you define the task that you want performed. After that task completes, you return the result. If the result cannot be computed, call( ) must throw an exception.

A Callable task is executed by an ExecutorService, by calling its submit( ) method. There are three forms of submit( ), but only one is used to execute a Callable. It is shown here:

<T> Future<T> submit(Callable<T> task)

Here, task is the Callable object that will be executed in its own thread. The result is returned through an object of type Future.

Future is a generic interface that represents the value that will be returned by a Callable object. Because this value is obtained at some future time, the name Future is appropriate. Future is defined like this:

interface Future<V>

Here, V specifies the type of the result.

To obtain the returned value, you will call Future’s get( ) method, which has these two forms:

V get( )

throws InterruptedException, ExecutionException

V get(long wait, TimeUnit tu)

throws InterruptedException, ExecutionException, TimeoutException

The first form waits for the result indefinitely. The second form allows you to specify a timeout period in wait. The units of wait are passed in tu, which is an object of the TimeUnit enumeration, described later in this chapter.

The following program illustrates Callable and Future by creating three tasks that perform three different computations. The first returns the summation of a value, the second computes the length of the hypotenuse of a right triangle given the length of its sides, and the third computes the factorial of a value. All three computations occur simultaneously.

// An example that uses a Callable.

import java.util.concurrent.*;

class CallableDemo {

public static void main(String args[]) { ExecutorService es = Executors.newFixedThreadPool(3); Future<Integer> f;

Future<Double> f2; Future<Integer> f3;

System.out.println("Starting");

f = es.submit(new Sum(10));

f2 = es.submit(new Hypot(3, 4));

f3 = es.submit(new Factorial(5));

try { System.out.println(f.get()); System.out.println(f2.get()); System.out.println(f3.get());

} catch (InterruptedException exc) { System.out.println(exc);

}

catch (ExecutionException exc) { System.out.println(exc);

}

es.shutdown();

System.out.println("Done");

}

}

// Following are three computational threads.

class Sum implements Callable<Integer> { int stop;

Sum(int v) { stop = v; }

public Integer call() { int sum = 0;

for(int i = 1; i <= stop; i++) { sum += i;

}

return sum;

}

}

class Hypot implements Callable<Double> { double side1, side2;

Hypot(double s1, double s2) { side1 = s1;

side2 = s2;

}

public Double call() {

return Math.sqrt((side1*side1) + (side2*side2));

}

}

class Factorial implements Callable<Integer> { int stop;

Factorial(int v) { stop = v; }

public Integer call() { int fact = 1;

for(int i = 2; i <= stop; i++) { fact *= i;

}

return fact;

}

}

The output is shown here:

Starting 55 5.0 120

Done