1. How can a Java application access the current thread?
public class MainThread {
public static void main(String[] args) {
long id = Thread.currentThread().getId();
String name = Thread.currentThread().getName();
...
}
}
2. What properties does each Java thread have?
Each Java thread has the following properties:
an identifier of type long that is unique within the JVM
a name of type String
a priority of type int
a state of type java.lang.Thread.State
a thread group the thread belongs to
3. What states can a thread have and what is the meaning of each state?
4. Is it possible to start a thread twice?
No, after having started a thread by invoking its start() method, a second invocation of start() will throw an IllegalThreadStateException.
5. What is a daemon thread?
A daemon thread is a thread whose execution state is not evaluated when the JVM decides if it should stop or not. The JVM stops when all user threads (in contrast to the daemon threads) are terminated. Hence daemon threads can be used to implement for example monitoring functionality as the thread is stopped by the JVM as soon as all user threads have stopped:
public class Example {
private static class MyDaemonThread extends Thread {
public MyDaemonThread() {
setDaemon(true);
}
@Override
public void run() {
while (true) {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public static void main(String[] args) throws InterruptedException {
Thread thread = new MyDaemonThread();
thread.start();
}
}
The example application above terminates even though the daemon thread is still running in its endless while loop.
6.How to take Thread Dump
You can generate a thread dump under Unix/Linux by running kill -QUIT <pid>, and under Windows by hitting Ctl + Break
7. What are differences between wait() and sleep() method in Java?
wait():
wait() method releases the lock.
wait() is the method of Object class.
wait() is the non-static method – public final void wait() throws InterruptedException { //…}
wait() should be notified by notify() or notifyAll() methods.
wait() method needs to be called from a loop in order to deal with false alarm.
wait() method must be called from synchronized context (i.e. synchronized method or block), otherwise it will throw IllegalMonitorStateException
sleep():
sleep() method doesn’t release the lock.
sleep() is the method of java.lang.Thread class.
sleep() is the static method – public static void sleep(long millis, int nanos) throws InterruptedException { //… }
after the specified amount of time, sleep() is completed.
sleep() better not to call from loop(i.e. see code below).
sleep() may be called from anywhere. there is no specific requirement.
8. What happens when an uncaught exception leaves the run() method?
I can happen that an unchecked exception escapes from the run() method. In this case the thread is stopped by the Java Virtual Machine. It is possible to catch this exception by registering an instance that implements the interface UncaughtExceptionHandler as an exception handler.
This is either done by invoking the static method Thread.setDefaultUncaughtExceptionHandler(Thread.UncaughtExceptionHandler), which tells the JVM to use the provided handler in case there was no specific handler registerd on the thread itself, or by invoking setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler) on the thread instance itself.
9. What is the difference between the two interfaces Runnable and
Callable?
The interface Runnable defines the method run() without any return value whereas the interface Callable allows the method call() to return a value and to throw an exception.
10. What is a shutdown hook?
A shutdown hook is a thread that gets executed when the JVM shuts down. It can be registered by invoking addShutdownHook(Runnable) on the Runtime instance:
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
}
});
11. Need for Callable
There are two ways of creating threads – one by extending the Thread class and other by creating a thread with a Runnable. However, one feature lacking in Runnable is that we cannot make a thread return result when it terminates, i.e. when run() completes. For supporting this feature, the Callable interface is present in Java.
12. Callable vs Runnable
For implementing Runnable, the run() method needs to be implemented which does not return anything, while for a Callable, the call() method needs to be implemented which returns a result on completion. Note that a thread can’t be created with a Callable, it can only be created with a Runnable.
Another difference is that the call() method can throw an exception whereas run() cannot.
A Runnable, however, does not return a result and cannot throw a checked exception.
Method signature that has to overridden for implementing Callable.
public Object call() throws Exception;
13. Future
A Future represents the result of an asynchronous computation. Methods are provided to check if the computation is complete, to wait for its completion, and to retrieve the result of the computation. The result can only be retrieved using method get when the computation has completed, blocking if necessary until it is ready. Cancellation is performed by the cancel method. Additional methods are provided to determine if the task completed normally or was cancelled. Once a computation has completed, the computation cannot be cancelled.
Observe that Callable and Future do two different things – Callable is similar to Runnable, in that it encapsulates a task that is meant to run on another thread, whereas a Future is used to store a result obtained from a different thread. In fact, the Future can be made to work with Runnable as well, which is something that will become clear when Executors come into the picture.
14. Shutting down exector
Finally, When you are done using the ExecutorService you should shut it down, so the threads do not keep running.
1
executor.shutdown();
For instance, if your application is started via a main() method and your main thread exits your application, the application will keep running if you have an active ExecutorService in your application. The active threads inside this ExecutorService prevents the JVM from shutting down.
To terminate the threads inside the ExecutorService you call its shutdown() method. The ExecutorService will not shut down immediately, but it will no longer accept new tasks, and once all threads have finished current tasks, the ExecutorService shuts down. All tasks submitted to the ExecutorService before shutdown() is called, are executed.
15. Future Task Example :
CallableCalculater.java:
package com.jcg;
import java.util.concurrent.Callable;
/**
* @author ashraf
*
*/
public class CallableCalculater implements Callable {
private long first;
private long last;
private long divisor;
/**
* Instantiates a new callable calculater.
*
* @param first the first
* @param last the last
* @param divisor the divisor
*/
public CallableCalculater(long first, long last, long divisor) {
this.first = first;
this.last = last;
this.divisor = divisor;
}
@Override
public Long call() throws Exception {
return Calculater.calculateNumberOfDivisible(first, last, divisor);
}
}
CallableCalculater.java is wrapping the calculateNumberOfDivisible() of Calculater.java in a Callable task to be given to our FutureTask later on.
FutureTaskDemo.java:
package com.jcg;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
/**
* @author ashraf
*
*/
public class FutureTaskDemo {
// Maximum number to check
public static final long MAX_NUMBER = 3000000000l;
// DIVISOR to be used in calculation
private static final long DIVISOR = 3;
/**
* @param args
* @throws ExecutionException
* @throws InterruptedException
*/
public static void main(String[] args) {
// Sequential execution
System.out.println("Starting sequential execution ....");
long timeStart = System.currentTimeMillis();
long result = Calculater.calculateNumberOfDivisible(0, MAX_NUMBER, DIVISOR);
long timeEnd = System.currentTimeMillis();
long timeNeeded = timeEnd - timeStart;
System.out.println("Result : " + result + " calculated in " + timeNeeded + " ms");
// Parallel execution
System.out.println("Starting parallel execution ....");
long timeStartFuture = System.currentTimeMillis();
long resultFuture = 0;
// Create a new ExecutorService with 2 thread to execute and store the Futures
ExecutorService executor = Executors.newFixedThreadPool(2);
List<FutureTask> taskList = new ArrayList<FutureTask>();
// Start thread for the first half of the numbers
FutureTask futureTask_1 = new FutureTask(new CallableCalculater(0, MAX_NUMBER / 2, DIVISOR));
taskList.add(futureTask_1);
executor.execute(futureTask_1);
// Start thread for the second half of the numbers
FutureTask futureTask_2 = new FutureTask(new CallableCalculater(MAX_NUMBER / 2 + 1, MAX_NUMBER, 3));
taskList.add(futureTask_2);
executor.execute(futureTask_2);
// Wait until all results are available and combine them at the same time
for (FutureTask futureTask : taskList) {
try {
resultFuture += futureTask.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
// Shutdown the ExecutorService
executor.shutdown();
long timeEndFuture = System.currentTimeMillis();
long timeNeededFuture = timeEndFuture - timeStartFuture;
System.out.println("Result (Future): " + resultFuture + " calculated in " + timeNeededFuture + " ms");
}
}
16. Fork/Join Framework
The effective use of parallel cores in a Java program has always been a challenge. There were few home-grown frameworks that would distribute the work across multiple cores and then join them to return the result set. Java 7 has incorporated this feature as a Fork and Join framework.
Basically the Fork-Join breaks the task at hand into mini-tasks until the mini-task is simple enough that it can be solved without further breakups. It’s like a divide-and-conquer algorithm. One important concept to note in this framework is that ideally no worker thread is idle. They implement a work-stealing algorithm in that idle workers steal the work from those workers who are busy.
ForkJoinPool
The ForkJoinPool is basically a specialized implementation of ExecutorService implementing the work-stealing algorithm we talked about above. We create an instance of ForkJoinPool by providing the target parallelism level i.e. the number of processors as shown below:
ForkJoinPool pool = new ForkJoinPool(numberOfProcessors);
Where numberOfProcessors = Runtime.getRunTime().availableProcessors();
If you use a no-argument constructor, by default, it creates a pool of size that equals the number of available processors obtained using above technique.
Although you specify any initial pool size, the pool adjusts its size dynamically in an attempt to maintain enough active threads at any given point in time. Another important difference compared to other ExecutorService's is that this pool need not be explicitly shutdown upon program exit because all its threads are in daemon mode.
ForkJoinTask
This is an abstract class for creating tasks that run within a ForkJoinPool. The Recursiveaction and RecursiveTask are the only two direct, known subclasses of ForkJoinTask. The only difference between these two classes is that the RecursiveAction does not return a value while RecursiveTask does have a return value and returns an object of specified type.
In both cases, you would need to implement the compute method in your subclass that performs the main computation desired by the task.
The ForkJoinTask class provides several methods for checking the execution status of a task. The isDone() method returns true if a task completes in any way. The isCompletedNormally() method returns true if a task completes without cancellation or encountering an exception, and isCancelled() returns true if the task was cancelled. Lastly, isCompletedabnormally() returns true if the task was either cancelled or encountered an exception.
18 Example Implementations of Fork/Join Pool Framework
In this example, you will learn how to use the asynchronous methods provided by the ForkJoinPool and ForkJoinTask classes for the management of tasks. You are going to implement a program that will search for files with a determined extension inside a folder and its subfolders. The ForkJoinTask class you’re going to implement will process the content of a folder. For each subfolder inside that folder, it will send a new task to the ForkJoinPool class in an asynchronous way. For each file inside that folder, the task will check the extension of the file and add it to the result list if it proceeds.
The solution to above problem is implemented in FolderProcessor class, which is given below:
Implementation Sourcecode
FolderProcessor.java
package forkJoinDemoAsyncExample;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.RecursiveTask;
public class FolderProcessor extends RecursiveTask<List<String>>
{
private static final long serialVersionUID = 1L;
//This attribute will store the full path of the folder this task is going to process.
private final String path;
//This attribute will store the name of the extension of the files this task is going to look for.
private final String extension;
//Implement the constructor of the class to initialize its attributes
public FolderProcessor(String path, String extension)
{
this.path = path;
this.extension = extension;
}
//Implement the compute() method. As you parameterized the RecursiveTask class with the List<String> type,
//this method has to return an object of that type.
@Override
protected List<String> compute()
{
//List to store the names of the files stored in the folder.
List<String> list = new ArrayList<String>();
//FolderProcessor tasks to store the subtasks that are going to process the subfolders stored in the folder
List<FolderProcessor> tasks = new ArrayList<FolderProcessor>();
//Get the content of the folder.
File file = new File(path);
File content[] = file.listFiles();
//For each element in the folder, if there is a subfolder, create a new FolderProcessor object
//and execute it asynchronously using the fork() method.
if (content != null)
{
for (int i = 0; i < content.length; i++)
{
if (content[i].isDirectory())
{
FolderProcessor task = new FolderProcessor(content[i].getAbsolutePath(), extension);
task.fork();
tasks.add(task);
}
//Otherwise, compare the extension of the file with the extension you are looking for using the checkFile() method
//and, if they are equal, store the full path of the file in the list of strings declared earlier.
else
{
if (checkFile(content[i].getName()))
{
list.add(content[i].getAbsolutePath());
}
}
}
}
//If the list of the FolderProcessor subtasks has more than 50 elements,
//write a message to the console to indicate this circumstance.
if (tasks.size() > 50)
{
System.out.printf("%s: %d tasks ran.\n", file.getAbsolutePath(), tasks.size());
}
//add to the list of files the results returned by the subtasks launched by this task.
addResultsFromTasks(list, tasks);
//Return the list of strings
return list;
}
//For each task stored in the list of tasks, call the join() method that will wait for its finalization and then will return the result of the task.
//Add that result to the list of strings using the addAll() method.
private void addResultsFromTasks(List<String> list, List<FolderProcessor> tasks)
{
for (FolderProcessor item : tasks)
{
list.addAll(item.join());
}
}
//This method compares if the name of a file passed as a parameter ends with the extension you are looking for.
private boolean checkFile(String name)
{
return name.endsWith(extension);
}
}
And to use above FolderProcessor, follow below code:
Main.java
package forkJoinDemoAsyncExample;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
public class Main
{
public static void main(String[] args)
{
//Create ForkJoinPool using the default constructor.
ForkJoinPool pool = new ForkJoinPool();
//Create three FolderProcessor tasks. Initialize each one with a different folder path.
FolderProcessor system = new FolderProcessor("C:\\Windows", "log");
FolderProcessor apps = new FolderProcessor("C:\\Program Files", "log");
FolderProcessor documents = new FolderProcessor("C:\\Documents And Settings", "log");
//Execute the three tasks in the pool using the execute() method.
pool.execute(system);
pool.execute(apps);
pool.execute(documents);
//Write to the console information about the status of the pool every second
//until the three tasks have finished their execution.
do
{
System.out.printf("******************************************\n");
System.out.printf("Main: Parallelism: %d\n", pool.getParallelism());
System.out.printf("Main: Active Threads: %d\n", pool.getActiveThreadCount());
System.out.printf("Main: Task Count: %d\n", pool.getQueuedTaskCount());
System.out.printf("Main: Steal Count: %d\n", pool.getStealCount());
System.out.printf("******************************************\n");
try
{
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e)
{
e.printStackTrace();
}
} while ((!system.isDone()) || (!apps.isDone()) || (!documents.isDone()));
//Shut down ForkJoinPool using the shutdown() method.
pool.shutdown();
//Write the number of results generated by each task to the console.
List<String> results;
results = system.join();
System.out.printf("System: %d files found.\n", results.size());
results = apps.join();
System.out.printf("Apps: %d files found.\n", results.size());
results = documents.join();
System.out.printf("Documents: %d files found.\n", results.size());
}
}
Output of above program will look like this:
Main: Parallelism: 2
Main: Active Threads: 3
Main: Task Count: 1403
Main: Steal Count: 5551
******************************************
******************************************
Main: Parallelism: 2
Main: Active Threads: 3
Main: Task Count: 586
Main: Steal Count: 5551
******************************************
System: 337 files found.
Apps: 10 files found.
Documents: 0 files found.
How it works?
In the FolderProcessor class, Each task processes the content of a folder. As you know, this content has the following two kinds of elements:
Files
Other folders
If the task finds a folder, it creates another Task object to process that folder and sends it to the pool using the fork() method. This method sends the task to the pool that will execute it if it has a free worker-thread or it can create a new one. The method returns immediately, so the task can continue processing the content of the folder. For every file, a task compares its extension with the one it’s looking for and, if they are equal, adds the name of the file to the list of results.
Once the task has processed all the content of the assigned folder, it waits for the finalization of all the tasks it sent to the pool using the join() method. This method called in a task waits for the finalization of its execution and returns the value returned by the compute() method. The task groups the results of all the tasks it sent with its own results and returns that list as a return value of the compute() method.
19. Difference between Fork/Join Framework And ExecutorService
The main difference between the Fork/Join and the Executor frameworks is the work-stealing algorithm. Unlike the Executor framework, when a task is waiting for the finalization of the sub-tasks it has created using the join operation, the thread that is executing that task (called worker thread ) looks for other tasks that have not been executed yet and begins its execution. By this way, the threads take full advantage of their running time, thereby improving the performance of the application.
Existing Implementations in JDK
There are some generally useful features in Java SE which are already implemented using the fork/join framework.
1) One such implementation, introduced in Java SE 8, is used by the java.util.Arrays class for its parallelSort() methods. These methods are similar to sort(), but leverage concurrency via the fork/join framework. Parallel sorting of large arrays is faster than sequential sorting when run on multiprocessor systems.
2) Parallelism used in Stream.parallel(). Read more about this parallel stream operation in java 8.
20. JAVA 8 Enhancements
Java JDK8 included the big fat interface called CompletionStage in the java.util.concurrent package. The same package also contains CompletableFuture which is a library implementation of CompletionStage. In this post we would see how CompletionStage and CompletableFuture provide piped asynchronous API thus enhancing reactive programming support in Java at the platform level.
21. CompletableFuture
CompletableFuture is a concrete implementation of a CompletionStage and it also implements the java.util.concurrent.Future interface as well. This is the class which models a task (which may or may not be asynchronous) and exposes various methods to interact with the task; for instance, we have methods to check if the task has completed; whether it has completed exceptionally; we even have APIs to chain dependencies between multiple tasks; cancelling uncompleted tasks, so on and so forth. We would be looking into some of these APIs soon.
A CompletableFuture can be instantiated and related methods can be called upon it, and we will see this in action in the subsequent section. However, there are convenient, static overloaded factory methods which provides further flexibility so that rather than worrying about harnessing CompletableFuture for a task, we can just concentrate on the task itself. I will explain this in a bit, but lets quickly have a look at the overloaded factory methods that I am talking about:
CompletableFuture supplyAsync() API
public static CompletableFuture supplyAsync(Supplier supplier)
public static CompletableFuture supplyAsync(Supplier supplier, Executor executor)
java.util.function.Supplier is a functional interface which accepts nothing and “supplies” an output. The supplyAsync() API expects that a result-producing task be wrapped in a Supplier instance and handed over to the supplyAsync() method, which would then return a CompletableFuture representing this task. This task would, by default, be executed with one of the threads from the standard java.util.concurrent.ForkJoinPool (public static ForkJoinPool commonPool()).
However, we can also provide custom thread pool by passing a java.util.concurrent.Executor instance and as such the Supplier tasks would be scheduled on threads from this Executor instance.
So to sum up the, the easiest way of using CompletableFuture API is to wrap the task you want to execute in a Supplier – you may additionally supply an Executor service as needed – and hand it over to the supplyAsync() method which would return you the CompletableFuture!
There is yet another variant API available for retrieving a CompletableFuture. Notice that while discussing supplyAsync() I wrote that this is to be used when task would be result-bearing, in other words, when we expect the task to return back some output. However, in all cases where the task might not return any output, we may use the runAsyn() API, instead:
CompletableFuture runAsync() API
public static CompletableFuture runAsync(Runnable runnable)
public static CompletableFuture runAsync(Runnable runnable, Executor executor)
Notice that runAsync() expects a java.lang.Runnable instance, and we know that Runnable.run() does not return any result! This is the reason that the returned CompletableFuture type erases itself to Void type.
22. Completing the CompletableFuturer manually
A CompletableFuture can be instantiated through its no-arg constructor. And then we can manually provide a Runnable instance to a custom thread; and then CompletableFuture API provides complete method using which the CompletableFuture can be manually completed:
How to manually complete a CompletableFuture
//1. Why named CompletableFuture?
CompletableFuture completableFuture1 = new CompletableFuture();
new Thread (()-> {
try {
Thread.sleep(4000L);
} catch (Exception e) {
completableFuture1.complete(-100.0);
}
/*
* we can manually "complete" a CompletableFuture!!
* this feature is not found with the classical Future interface
*/
completableFuture1.complete(100.0);
},"CompFut1-Thread").start();
System.out.println("ok...waiting at: "+new Date());
System.out.format("compFut value and received at: %f, %s \n", completableFuture1.join(), new Date());
23. More cases of Completable future
https://www.callicoder.com/java-8-completablefuture-tutorial/
https://examples.javacodegeeks.com/core-java/util/concurrent/completablefuture/java-completionstage-completablefuture-example/
Very Good Article for Completable future :
https://netjs.blogspot.com/2018/11/completablefuture-in-java-with-examples.html
24.
public class MainThread {
public static void main(String[] args) {
long id = Thread.currentThread().getId();
String name = Thread.currentThread().getName();
...
}
}
2. What properties does each Java thread have?
Each Java thread has the following properties:
an identifier of type long that is unique within the JVM
a name of type String
a priority of type int
a state of type java.lang.Thread.State
a thread group the thread belongs to
3. What states can a thread have and what is the meaning of each state?
4. Is it possible to start a thread twice?
No, after having started a thread by invoking its start() method, a second invocation of start() will throw an IllegalThreadStateException.
5. What is a daemon thread?
A daemon thread is a thread whose execution state is not evaluated when the JVM decides if it should stop or not. The JVM stops when all user threads (in contrast to the daemon threads) are terminated. Hence daemon threads can be used to implement for example monitoring functionality as the thread is stopped by the JVM as soon as all user threads have stopped:
public class Example {
private static class MyDaemonThread extends Thread {
public MyDaemonThread() {
setDaemon(true);
}
@Override
public void run() {
while (true) {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public static void main(String[] args) throws InterruptedException {
Thread thread = new MyDaemonThread();
thread.start();
}
}
The example application above terminates even though the daemon thread is still running in its endless while loop.
6.How to take Thread Dump
You can generate a thread dump under Unix/Linux by running kill -QUIT <pid>, and under Windows by hitting Ctl + Break
7. What are differences between wait() and sleep() method in Java?
wait():
wait() method releases the lock.
wait() is the method of Object class.
wait() is the non-static method – public final void wait() throws InterruptedException { //…}
wait() should be notified by notify() or notifyAll() methods.
wait() method needs to be called from a loop in order to deal with false alarm.
wait() method must be called from synchronized context (i.e. synchronized method or block), otherwise it will throw IllegalMonitorStateException
sleep():
sleep() method doesn’t release the lock.
sleep() is the method of java.lang.Thread class.
sleep() is the static method – public static void sleep(long millis, int nanos) throws InterruptedException { //… }
after the specified amount of time, sleep() is completed.
sleep() better not to call from loop(i.e. see code below).
sleep() may be called from anywhere. there is no specific requirement.
8. What happens when an uncaught exception leaves the run() method?
I can happen that an unchecked exception escapes from the run() method. In this case the thread is stopped by the Java Virtual Machine. It is possible to catch this exception by registering an instance that implements the interface UncaughtExceptionHandler as an exception handler.
This is either done by invoking the static method Thread.setDefaultUncaughtExceptionHandler(Thread.UncaughtExceptionHandler), which tells the JVM to use the provided handler in case there was no specific handler registerd on the thread itself, or by invoking setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler) on the thread instance itself.
9. What is the difference between the two interfaces Runnable and
Callable?
The interface Runnable defines the method run() without any return value whereas the interface Callable allows the method call() to return a value and to throw an exception.
10. What is a shutdown hook?
A shutdown hook is a thread that gets executed when the JVM shuts down. It can be registered by invoking addShutdownHook(Runnable) on the Runtime instance:
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
}
});
11. Need for Callable
There are two ways of creating threads – one by extending the Thread class and other by creating a thread with a Runnable. However, one feature lacking in Runnable is that we cannot make a thread return result when it terminates, i.e. when run() completes. For supporting this feature, the Callable interface is present in Java.
12. Callable vs Runnable
For implementing Runnable, the run() method needs to be implemented which does not return anything, while for a Callable, the call() method needs to be implemented which returns a result on completion. Note that a thread can’t be created with a Callable, it can only be created with a Runnable.
Another difference is that the call() method can throw an exception whereas run() cannot.
A Runnable, however, does not return a result and cannot throw a checked exception.
Method signature that has to overridden for implementing Callable.
public Object call() throws Exception;
13. Future
A Future represents the result of an asynchronous computation. Methods are provided to check if the computation is complete, to wait for its completion, and to retrieve the result of the computation. The result can only be retrieved using method get when the computation has completed, blocking if necessary until it is ready. Cancellation is performed by the cancel method. Additional methods are provided to determine if the task completed normally or was cancelled. Once a computation has completed, the computation cannot be cancelled.
Observe that Callable and Future do two different things – Callable is similar to Runnable, in that it encapsulates a task that is meant to run on another thread, whereas a Future is used to store a result obtained from a different thread. In fact, the Future can be made to work with Runnable as well, which is something that will become clear when Executors come into the picture.
14. Shutting down exector
Finally, When you are done using the ExecutorService you should shut it down, so the threads do not keep running.
1
executor.shutdown();
For instance, if your application is started via a main() method and your main thread exits your application, the application will keep running if you have an active ExecutorService in your application. The active threads inside this ExecutorService prevents the JVM from shutting down.
To terminate the threads inside the ExecutorService you call its shutdown() method. The ExecutorService will not shut down immediately, but it will no longer accept new tasks, and once all threads have finished current tasks, the ExecutorService shuts down. All tasks submitted to the ExecutorService before shutdown() is called, are executed.
15. Future Task Example :
CallableCalculater.java:
package com.jcg;
import java.util.concurrent.Callable;
/**
* @author ashraf
*
*/
public class CallableCalculater implements Callable {
private long first;
private long last;
private long divisor;
/**
* Instantiates a new callable calculater.
*
* @param first the first
* @param last the last
* @param divisor the divisor
*/
public CallableCalculater(long first, long last, long divisor) {
this.first = first;
this.last = last;
this.divisor = divisor;
}
@Override
public Long call() throws Exception {
return Calculater.calculateNumberOfDivisible(first, last, divisor);
}
}
CallableCalculater.java is wrapping the calculateNumberOfDivisible() of Calculater.java in a Callable task to be given to our FutureTask later on.
FutureTaskDemo.java:
package com.jcg;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
/**
* @author ashraf
*
*/
public class FutureTaskDemo {
// Maximum number to check
public static final long MAX_NUMBER = 3000000000l;
// DIVISOR to be used in calculation
private static final long DIVISOR = 3;
/**
* @param args
* @throws ExecutionException
* @throws InterruptedException
*/
public static void main(String[] args) {
// Sequential execution
System.out.println("Starting sequential execution ....");
long timeStart = System.currentTimeMillis();
long result = Calculater.calculateNumberOfDivisible(0, MAX_NUMBER, DIVISOR);
long timeEnd = System.currentTimeMillis();
long timeNeeded = timeEnd - timeStart;
System.out.println("Result : " + result + " calculated in " + timeNeeded + " ms");
// Parallel execution
System.out.println("Starting parallel execution ....");
long timeStartFuture = System.currentTimeMillis();
long resultFuture = 0;
// Create a new ExecutorService with 2 thread to execute and store the Futures
ExecutorService executor = Executors.newFixedThreadPool(2);
List<FutureTask> taskList = new ArrayList<FutureTask>();
// Start thread for the first half of the numbers
FutureTask futureTask_1 = new FutureTask(new CallableCalculater(0, MAX_NUMBER / 2, DIVISOR));
taskList.add(futureTask_1);
executor.execute(futureTask_1);
// Start thread for the second half of the numbers
FutureTask futureTask_2 = new FutureTask(new CallableCalculater(MAX_NUMBER / 2 + 1, MAX_NUMBER, 3));
taskList.add(futureTask_2);
executor.execute(futureTask_2);
// Wait until all results are available and combine them at the same time
for (FutureTask futureTask : taskList) {
try {
resultFuture += futureTask.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
// Shutdown the ExecutorService
executor.shutdown();
long timeEndFuture = System.currentTimeMillis();
long timeNeededFuture = timeEndFuture - timeStartFuture;
System.out.println("Result (Future): " + resultFuture + " calculated in " + timeNeededFuture + " ms");
}
}
16. Fork/Join Framework
The effective use of parallel cores in a Java program has always been a challenge. There were few home-grown frameworks that would distribute the work across multiple cores and then join them to return the result set. Java 7 has incorporated this feature as a Fork and Join framework.
Basically the Fork-Join breaks the task at hand into mini-tasks until the mini-task is simple enough that it can be solved without further breakups. It’s like a divide-and-conquer algorithm. One important concept to note in this framework is that ideally no worker thread is idle. They implement a work-stealing algorithm in that idle workers steal the work from those workers who are busy.
ForkJoinPool
The ForkJoinPool is basically a specialized implementation of ExecutorService implementing the work-stealing algorithm we talked about above. We create an instance of ForkJoinPool by providing the target parallelism level i.e. the number of processors as shown below:
ForkJoinPool pool = new ForkJoinPool(numberOfProcessors);
Where numberOfProcessors = Runtime.getRunTime().availableProcessors();
If you use a no-argument constructor, by default, it creates a pool of size that equals the number of available processors obtained using above technique.
Although you specify any initial pool size, the pool adjusts its size dynamically in an attempt to maintain enough active threads at any given point in time. Another important difference compared to other ExecutorService's is that this pool need not be explicitly shutdown upon program exit because all its threads are in daemon mode.
ForkJoinTask
This is an abstract class for creating tasks that run within a ForkJoinPool. The Recursiveaction and RecursiveTask are the only two direct, known subclasses of ForkJoinTask. The only difference between these two classes is that the RecursiveAction does not return a value while RecursiveTask does have a return value and returns an object of specified type.
In both cases, you would need to implement the compute method in your subclass that performs the main computation desired by the task.
The ForkJoinTask class provides several methods for checking the execution status of a task. The isDone() method returns true if a task completes in any way. The isCompletedNormally() method returns true if a task completes without cancellation or encountering an exception, and isCancelled() returns true if the task was cancelled. Lastly, isCompletedabnormally() returns true if the task was either cancelled or encountered an exception.
18 Example Implementations of Fork/Join Pool Framework
In this example, you will learn how to use the asynchronous methods provided by the ForkJoinPool and ForkJoinTask classes for the management of tasks. You are going to implement a program that will search for files with a determined extension inside a folder and its subfolders. The ForkJoinTask class you’re going to implement will process the content of a folder. For each subfolder inside that folder, it will send a new task to the ForkJoinPool class in an asynchronous way. For each file inside that folder, the task will check the extension of the file and add it to the result list if it proceeds.
The solution to above problem is implemented in FolderProcessor class, which is given below:
Implementation Sourcecode
FolderProcessor.java
package forkJoinDemoAsyncExample;
import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.RecursiveTask;
public class FolderProcessor extends RecursiveTask<List<String>>
{
private static final long serialVersionUID = 1L;
//This attribute will store the full path of the folder this task is going to process.
private final String path;
//This attribute will store the name of the extension of the files this task is going to look for.
private final String extension;
//Implement the constructor of the class to initialize its attributes
public FolderProcessor(String path, String extension)
{
this.path = path;
this.extension = extension;
}
//Implement the compute() method. As you parameterized the RecursiveTask class with the List<String> type,
//this method has to return an object of that type.
@Override
protected List<String> compute()
{
//List to store the names of the files stored in the folder.
List<String> list = new ArrayList<String>();
//FolderProcessor tasks to store the subtasks that are going to process the subfolders stored in the folder
List<FolderProcessor> tasks = new ArrayList<FolderProcessor>();
//Get the content of the folder.
File file = new File(path);
File content[] = file.listFiles();
//For each element in the folder, if there is a subfolder, create a new FolderProcessor object
//and execute it asynchronously using the fork() method.
if (content != null)
{
for (int i = 0; i < content.length; i++)
{
if (content[i].isDirectory())
{
FolderProcessor task = new FolderProcessor(content[i].getAbsolutePath(), extension);
task.fork();
tasks.add(task);
}
//Otherwise, compare the extension of the file with the extension you are looking for using the checkFile() method
//and, if they are equal, store the full path of the file in the list of strings declared earlier.
else
{
if (checkFile(content[i].getName()))
{
list.add(content[i].getAbsolutePath());
}
}
}
}
//If the list of the FolderProcessor subtasks has more than 50 elements,
//write a message to the console to indicate this circumstance.
if (tasks.size() > 50)
{
System.out.printf("%s: %d tasks ran.\n", file.getAbsolutePath(), tasks.size());
}
//add to the list of files the results returned by the subtasks launched by this task.
addResultsFromTasks(list, tasks);
//Return the list of strings
return list;
}
//For each task stored in the list of tasks, call the join() method that will wait for its finalization and then will return the result of the task.
//Add that result to the list of strings using the addAll() method.
private void addResultsFromTasks(List<String> list, List<FolderProcessor> tasks)
{
for (FolderProcessor item : tasks)
{
list.addAll(item.join());
}
}
//This method compares if the name of a file passed as a parameter ends with the extension you are looking for.
private boolean checkFile(String name)
{
return name.endsWith(extension);
}
}
And to use above FolderProcessor, follow below code:
Main.java
package forkJoinDemoAsyncExample;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
public class Main
{
public static void main(String[] args)
{
//Create ForkJoinPool using the default constructor.
ForkJoinPool pool = new ForkJoinPool();
//Create three FolderProcessor tasks. Initialize each one with a different folder path.
FolderProcessor system = new FolderProcessor("C:\\Windows", "log");
FolderProcessor apps = new FolderProcessor("C:\\Program Files", "log");
FolderProcessor documents = new FolderProcessor("C:\\Documents And Settings", "log");
//Execute the three tasks in the pool using the execute() method.
pool.execute(system);
pool.execute(apps);
pool.execute(documents);
//Write to the console information about the status of the pool every second
//until the three tasks have finished their execution.
do
{
System.out.printf("******************************************\n");
System.out.printf("Main: Parallelism: %d\n", pool.getParallelism());
System.out.printf("Main: Active Threads: %d\n", pool.getActiveThreadCount());
System.out.printf("Main: Task Count: %d\n", pool.getQueuedTaskCount());
System.out.printf("Main: Steal Count: %d\n", pool.getStealCount());
System.out.printf("******************************************\n");
try
{
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e)
{
e.printStackTrace();
}
} while ((!system.isDone()) || (!apps.isDone()) || (!documents.isDone()));
//Shut down ForkJoinPool using the shutdown() method.
pool.shutdown();
//Write the number of results generated by each task to the console.
List<String> results;
results = system.join();
System.out.printf("System: %d files found.\n", results.size());
results = apps.join();
System.out.printf("Apps: %d files found.\n", results.size());
results = documents.join();
System.out.printf("Documents: %d files found.\n", results.size());
}
}
Output of above program will look like this:
Main: Parallelism: 2
Main: Active Threads: 3
Main: Task Count: 1403
Main: Steal Count: 5551
******************************************
******************************************
Main: Parallelism: 2
Main: Active Threads: 3
Main: Task Count: 586
Main: Steal Count: 5551
******************************************
System: 337 files found.
Apps: 10 files found.
Documents: 0 files found.
How it works?
In the FolderProcessor class, Each task processes the content of a folder. As you know, this content has the following two kinds of elements:
Files
Other folders
If the task finds a folder, it creates another Task object to process that folder and sends it to the pool using the fork() method. This method sends the task to the pool that will execute it if it has a free worker-thread or it can create a new one. The method returns immediately, so the task can continue processing the content of the folder. For every file, a task compares its extension with the one it’s looking for and, if they are equal, adds the name of the file to the list of results.
Once the task has processed all the content of the assigned folder, it waits for the finalization of all the tasks it sent to the pool using the join() method. This method called in a task waits for the finalization of its execution and returns the value returned by the compute() method. The task groups the results of all the tasks it sent with its own results and returns that list as a return value of the compute() method.
19. Difference between Fork/Join Framework And ExecutorService
The main difference between the Fork/Join and the Executor frameworks is the work-stealing algorithm. Unlike the Executor framework, when a task is waiting for the finalization of the sub-tasks it has created using the join operation, the thread that is executing that task (called worker thread ) looks for other tasks that have not been executed yet and begins its execution. By this way, the threads take full advantage of their running time, thereby improving the performance of the application.
Existing Implementations in JDK
There are some generally useful features in Java SE which are already implemented using the fork/join framework.
1) One such implementation, introduced in Java SE 8, is used by the java.util.Arrays class for its parallelSort() methods. These methods are similar to sort(), but leverage concurrency via the fork/join framework. Parallel sorting of large arrays is faster than sequential sorting when run on multiprocessor systems.
2) Parallelism used in Stream.parallel(). Read more about this parallel stream operation in java 8.
20. JAVA 8 Enhancements
Java JDK8 included the big fat interface called CompletionStage in the java.util.concurrent package. The same package also contains CompletableFuture which is a library implementation of CompletionStage. In this post we would see how CompletionStage and CompletableFuture provide piped asynchronous API thus enhancing reactive programming support in Java at the platform level.
21. CompletableFuture
CompletableFuture is a concrete implementation of a CompletionStage and it also implements the java.util.concurrent.Future interface as well. This is the class which models a task (which may or may not be asynchronous) and exposes various methods to interact with the task; for instance, we have methods to check if the task has completed; whether it has completed exceptionally; we even have APIs to chain dependencies between multiple tasks; cancelling uncompleted tasks, so on and so forth. We would be looking into some of these APIs soon.
A CompletableFuture can be instantiated and related methods can be called upon it, and we will see this in action in the subsequent section. However, there are convenient, static overloaded factory methods which provides further flexibility so that rather than worrying about harnessing CompletableFuture for a task, we can just concentrate on the task itself. I will explain this in a bit, but lets quickly have a look at the overloaded factory methods that I am talking about:
CompletableFuture supplyAsync() API
public static CompletableFuture supplyAsync(Supplier supplier)
public static CompletableFuture supplyAsync(Supplier supplier, Executor executor)
java.util.function.Supplier is a functional interface which accepts nothing and “supplies” an output. The supplyAsync() API expects that a result-producing task be wrapped in a Supplier instance and handed over to the supplyAsync() method, which would then return a CompletableFuture representing this task. This task would, by default, be executed with one of the threads from the standard java.util.concurrent.ForkJoinPool (public static ForkJoinPool commonPool()).
However, we can also provide custom thread pool by passing a java.util.concurrent.Executor instance and as such the Supplier tasks would be scheduled on threads from this Executor instance.
So to sum up the, the easiest way of using CompletableFuture API is to wrap the task you want to execute in a Supplier – you may additionally supply an Executor service as needed – and hand it over to the supplyAsync() method which would return you the CompletableFuture!
There is yet another variant API available for retrieving a CompletableFuture. Notice that while discussing supplyAsync() I wrote that this is to be used when task would be result-bearing, in other words, when we expect the task to return back some output. However, in all cases where the task might not return any output, we may use the runAsyn() API, instead:
CompletableFuture runAsync() API
public static CompletableFuture runAsync(Runnable runnable)
public static CompletableFuture runAsync(Runnable runnable, Executor executor)
Notice that runAsync() expects a java.lang.Runnable instance, and we know that Runnable.run() does not return any result! This is the reason that the returned CompletableFuture type erases itself to Void type.
22. Completing the CompletableFuturer manually
A CompletableFuture can be instantiated through its no-arg constructor. And then we can manually provide a Runnable instance to a custom thread; and then CompletableFuture API provides complete method using which the CompletableFuture can be manually completed:
How to manually complete a CompletableFuture
//1. Why named CompletableFuture?
CompletableFuture completableFuture1 = new CompletableFuture();
new Thread (()-> {
try {
Thread.sleep(4000L);
} catch (Exception e) {
completableFuture1.complete(-100.0);
}
/*
* we can manually "complete" a CompletableFuture!!
* this feature is not found with the classical Future interface
*/
completableFuture1.complete(100.0);
},"CompFut1-Thread").start();
System.out.println("ok...waiting at: "+new Date());
System.out.format("compFut value and received at: %f, %s \n", completableFuture1.join(), new Date());
23. More cases of Completable future
https://www.callicoder.com/java-8-completablefuture-tutorial/
https://examples.javacodegeeks.com/core-java/util/concurrent/completablefuture/java-completionstage-completablefuture-example/
Very Good Article for Completable future :
https://netjs.blogspot.com/2018/11/completablefuture-in-java-with-examples.html
24.
No comments:
Post a Comment