In this post, we are going to discuss about the class java.util.concurrent.RunnableFuture and give you and idea of how you can use it on your own code when building robust multi-threaded applications.
1. RunnableFuture Class
Usually, when you develop a simple, concurrent-programming application in Java, you create some Runnable objects and then create the corresponding Thread objects to execute them. If you have to develop a program that runs a lot of concurrent tasks, this approach has the following disadvantages:
- You have to implement all the code-related information to the management of the Thread objects (creation, ending, obtaining results).
- You create a Thread object per task. If you have to execute a big number of tasks, this can affect the throughput of the application.
- You have to control and manage efficiently the resources of the computer. If you create too many threads, you can saturate the system.
Since Java 5, the Java concurrency API provides a mechanism that aims at resolving problems these kind of problems. This mechanism is called the Executor Framework and is around the Executor interface, its subinterface ExecutorService, and ThreadPoolExecutor class that implements both interfaces.
The ExecutorService interface extends Executor by adding methods that help manage and control the execution of threads. For example, ExecutorService defines void shutdown(), which stops the invoking ExecutorService. ExecutorService also defines methods that execute threads that return results, that execute a set of threads, and that determine the shutdown status.
ThreadPoolExecutor implements the Executor and ExecutorService interfaces and provides support for a managed pool of threads.
The Executor Framework, separates the task creation and its execution. With an executor, you only have to implement the Runnable objects and send them to the executor. The executor is responsible for their execution, instantiation, and running with necessary threads. But it goes beyond that and improves performance using a pool of threads.
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 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:
public static ExecutorService newCachedThreadPool()public static ExecutorService newFixedThreadPool(int nThreads)
newCachedThreadPool() creates a thread pool that adds threads as needed but reuses threads if possible. newFixedThreadPool(int nThreads) creates a thread pool that consists of a specified number of threads.
1.1 Using Callable and Future Classes
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:
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:
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, ExecutionExceptionV get(long timeout, TimeUnit unit) 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 unit, which is an object of the TimeUnit enumeration.
1.2 FutureTask Class
If you look at the JDK 7 API documentation, it barely says something about the RunnableFuture Class (Interface):
We can see, that this nterface is implemented by the FutureTask class, which is very useful when we want to control a task finishing in an executor.
The FutureTask class provides a method called done() that allows you to execute some code after the finalization of a task executed in an executor. It can be used to make some post-process operations, generating a report, sending results by e-mail, or releasing some resources. This method is called internally by the FutureTask class when the execution of the task that this FutureTask object is controlling finishes. The method is called after the result of the task is set and its status is changed to the isDone status, regardless of whether the task has been canceled or finished normally. By default, this method is empty. You can override the FutureTask class and implement this method to change this behavior.
2. Executing some code
In this example, we will learn how to override the done() method to execute code after the finalization of the tasks.
ExecutableTask.java
package com.javacodegeeks.examples.runnablefutureexample.runnables;
//~--- JDK imports ------------------------------------------------------------
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
// Implements the Callable interface parameterized with the String class.
public class ExecutableTask implements Callable<String> {
private static final Logger logger = Logger.getLogger("ExecutableTask");
// It will store the name of the task.
private String name;
// Implement the constructor of the class to initialize the name of the task.
public ExecutableTask(String name) {
this.name = name;
}
public String getName() {
return this.name;
}
// Put the task to sleep for a random period of time and return a message
// with the name of the task.
@Override
public String call() throws Exception {
try {
long duration = (long) (Math.random() * 10);
logger.info(this.name + ": Waiting " + duration + " seconds for results.");
TimeUnit.SECONDS.sleep(duration);
} catch (InterruptedException ie) {
logger.log(Level.SEVERE, ie.getLocalizedMessage());
ie.printStackTrace(System.err);
}
return "Hello, world. I'm " + this.name;
}
}
ResultTask.java
package com.javacodegeeks.examples.runnablefutureexample.futuretasks;
//~--- non-JDK imports --------------------------------------------------------
import com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask;
//~--- JDK imports ------------------------------------------------------------
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.logging.Logger;
//Extends the FutureTask class parameterized with the String class.
public class ResultTask extends FutureTask<String> {
private static final Logger logger = Logger.getLogger("ResultTask");
// It will store the name of the task.
private String name;
// Implement the constructor of the class.
// It has to receive a Callable object as a parameter.
public ResultTask(Callable<String> callable) {
super(callable);
this.name = ((ExecutableTask) callable).getName();
}
@Override
protected void done() {
if (this.isCancelled()) {
logger.info(this.name + ": has been canceled");
} else if (this.isDone()) {
logger.info(this.name + ": has finished");
}
}
}
App.java
package com.javacodegeeks.examples.runnablefutureexample;
//~--- non-JDK imports --------------------------------------------------------
import com.javacodegeeks.examples.runnablefutureexample.futuretasks.ResultTask;
import com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask;
//~--- JDK imports ------------------------------------------------------------
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
public class App {
private static final Logger logger = Logger.getLogger("App");
public static void main(String[] args) {
// Create ExecutorService using the newCachedThreadPool() method
// of the Executors class.
ExecutorService executorService = (ExecutorService) Executors.newCachedThreadPool();
// Create an array to store five ResultTask objects.
ResultTask[] resultTasks = new ResultTask[5];
// Send the each ResultTask to the executor ResultTask
// using the submit() method.
for (int i = 0; i < 5; i++) {
ExecutableTask executableTask = new ExecutableTask("Task" + i);
resultTasks[i] = new ResultTask(executableTask);
executorService.submit(resultTasks[i]);
}
// Put the main thread to sleep for 5 seconds.
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException ie) {
logger.log(Level.SEVERE, ie.getLocalizedMessage());
ie.printStackTrace(System.err);
}
// Cancel all the tasks you have sent to the executor.
for (int i = 0; i < resultTasks.length; i++) {
resultTasks[i].cancel(true);
}
// Write to the console the result of those tasks that haven't been
// canceled using the get() method of the ResultTask objects.
for (int i = 0; i < resultTasks.length; i++) {
try {
if (resultTasks[i].isCancelled()) {
logger.info("Task" + i + " was cancelled.");
} else if (resultTasks[i].isDone()) {
logger.info(resultTasks[i].get());
}
} catch (InterruptedException | ExecutionException e) {
logger.log(Level.SEVERE, e.getLocalizedMessage());
e.printStackTrace(System.err);
}
}
// Finish the executor using the shutdown() method.
executorService.shutdown();
}
}
Let’s explain the methods used in the previous code
protected void done()– Protected method invoked when this task transitions to state isDone (whether normally or via cancellation). The default implementation does nothing. Subclasses may override this method to invoke completion callbacks or perform bookkeeping. Note that you can query status inside the implementation of this method to determine whether this task has been cancelled.public boolean isCancelled()– Returns true if this task was cancelled before it completed normally.public boolean isDone()– Returns true if this task completed. Completion may be due to normal termination, an exception, or cancellation — in all of these cases, this method will return true.public boolean cancel(boolean mayInterruptIfRunning)– Attempts to cancel execution of this task. This attempt will fail if the task has already completed, has already been cancelled, or could not be cancelled for some other reason. If successful, and this task has not started when cancel is called, this task should never run. If the task has already started, then the mayInterruptIfRunning parameter determines whether the thread executing this task should be interrupted in an attempt to stop the task. After this method returns, subsequent calls to Future.isDone() will always return true. Subsequent calls to Future.isCancelled() will always return true if this method returned true.
The output of the command
com.javacodegeeks.examples.runnablefutureexample.App
should be similar to:
Oct 06, 2014 10:40:42 AM com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask call INFO: Task0: Waiting 7 seconds for results. Oct 06, 2014 10:40:42 AM com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask call INFO: Task2: Waiting 2 seconds for results. Oct 06, 2014 10:40:42 AM com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask call INFO: Task1: Waiting 9 seconds for results. Oct 06, 2014 10:40:42 AM com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask call INFO: Task3: Waiting 9 seconds for results. Oct 06, 2014 10:40:42 AM com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask call INFO: Task4: Waiting 5 seconds for results. Oct 06, 2014 10:40:44 AM com.javacodegeeks.examples.runnablefutureexample.futuretasks.ResultTask done INFO: Task2: has finished Oct 06, 2014 10:40:47 AM com.javacodegeeks.examples.runnablefutureexample.futuretasks.ResultTask done INFO: Task0: has been canceled Oct 06, 2014 10:40:47 AM com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask call SEVERE: sleep interrupted java.lang.InterruptedException: sleep interrupted at java.lang.Thread.sleep(Native Method) at java.lang.Thread.sleep(Thread.java:340) at java.util.concurrent.TimeUnit.sleep(TimeUnit.java:360) at com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask.call(ExecutableTask.java:34) at com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask.call(ExecutableTask.java:1) at java.util.concurrent.FutureTask.run(FutureTask.java:262) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471) at java.util.concurrent.FutureTask.run(FutureTask.java:262) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:745) Oct 06, 2014 10:40:47 AM com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask call SEVERE: sleep interrupted java.lang.InterruptedException: sleep interrupted at java.lang.Thread.sleep(Native Method) at java.lang.Thread.sleep(Thread.java:340) at java.util.concurrent.TimeUnit.sleep(TimeUnit.java:360) at com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask.call(ExecutableTask.java:34) at com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask.call(ExecutableTask.java:1) at java.util.concurrent.FutureTask.run(FutureTask.java:262) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471) at java.util.concurrent.FutureTask.run(FutureTask.java:262) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:745) Oct 06, 2014 10:40:47 AM com.javacodegeeks.examples.runnablefutureexample.futuretasks.ResultTask done INFO: Task1: has been canceled Oct 06, 2014 10:40:47 AM com.javacodegeeks.examples.runnablefutureexample.futuretasks.ResultTask done INFO: Task3: has been canceled Oct 06, 2014 10:40:47 AM com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask call SEVERE: sleep interrupted java.lang.InterruptedException: sleep interrupted at java.lang.Thread.sleep(Native Method) at java.lang.Thread.sleep(Thread.java:340) at java.util.concurrent.TimeUnit.sleep(TimeUnit.java:360) at com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask.call(ExecutableTask.java:34) at com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask.call(ExecutableTask.java:1) at java.util.concurrent.FutureTask.run(FutureTask.java:262) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471) at java.util.concurrent.FutureTask.run(FutureTask.java:262) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:745) Oct 06, 2014 10:40:47 AM com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask call SEVERE: sleep interrupted java.lang.InterruptedException: sleep interrupted at java.lang.Thread.sleep(Native Method) at java.lang.Thread.sleep(Thread.java:340) at java.util.concurrent.TimeUnit.sleep(TimeUnit.java:360) at com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask.call(ExecutableTask.java:34) at com.javacodegeeks.examples.runnablefutureexample.runnables.ExecutableTask.call(ExecutableTask.java:1) at java.util.concurrent.FutureTask.run(FutureTask.java:262) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471) at java.util.concurrent.FutureTask.run(FutureTask.java:262) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:745) Oct 06, 2014 10:40:47 AM com.javacodegeeks.examples.runnablefutureexample.futuretasks.ResultTask done INFO: Task4: has been canceled Oct 06, 2014 10:40:47 AM com.javacodegeeks.examples.runnablefutureexample.App main INFO: Task0 was cancelled. Oct 06, 2014 10:40:47 AM com.javacodegeeks.examples.runnablefutureexample.App main INFO: Task1 was cancelled. Oct 06, 2014 10:40:47 AM com.javacodegeeks.examples.runnablefutureexample.App main INFO: Hello, world. I'm Task2 Oct 06, 2014 10:40:47 AM com.javacodegeeks.examples.runnablefutureexample.App main INFO: Task3 was cancelled. Oct 06, 2014 10:40:47 AM com.javacodegeeks.examples.runnablefutureexample.App main INFO: Task4 was cancelled.
3. Download the Eclipse project of this tutorial:
This was an example of how to set use the RunnableFuture Class.
You can download the full source code of this example here : runnablefutureexample.zip
