Java programming language

Concurrent programming

Advanced thread synchronization and coordination

In the previous section of this unit we learnt some basic techniques for coordinating and synchronizing threads, such as joining threads or synchronizing methods and blocks of code. In this document we are going to see more advanced strategies provided by the java.util.concurrent API, together with some modern executor options available in current Java versions.

1. Using executors

When we are dealing with multiple threads in our application, we may face two problems:

These problems can be partially avoided by using executors. An executor separates the task that we want to perform from the details of the threads that execute it. Many executors use a pool of reusable worker threads, although other executors, such as virtual-thread-per-task executors, follow a different strategy. For instance, if we define a task like this:

public class MyThread implements Runnable
{
    ...

    @Override
    public void run()
    {
        ...
    }
}

Then we can create an executor that handles objects of type MyThread (and any other Runnable object), this way:

ExecutorService executor = Executors.newCachedThreadPool();

MyThread t1 = new MyThread();
MyThread t2 = new MyThread();
executor.execute(t1);
executor.execute(t2);

executor.shutdown();

We can typically define a loop to create tasks using a lambda expression, and submit them to the executor, this way:

ExecutorService executor = ...

for (int i = 0; i < N; i++)
{
    executor.execute(() -> { 
        // Thread code
    }); 
}

executor.shutdown();

In these examples we use the ExecutorService interface (from the java.util.concurrent package). The factory method Executors.newCachedThreadPool() returns an executor backed by a ThreadPoolExecutor. We submit tasks, not Thread objects: the executor decides which worker thread will run each task and when it will start. When we are no longer going to submit tasks, we should shut the executor down so that its resources can eventually be released. Calling shutdown() performs an orderly shutdown: previously submitted tasks are still executed, but new tasks are rejected. Since Java 19, ExecutorService also implements AutoCloseable, so in modern Java it can be used with try-with-resources when that style fits the lifetime of the executor.

1.1. Advantages of using executors

What advantages do executors offer if we compare them with traditional thread management?

ExecutorService executor = Executors.newFixedThreadPool(10);

If all worker threads are busy, additional tasks wait in the executor queue until a worker becomes available. For CPU-bound work, a starting point can be a pool size close to the number of available processors:

ExecutorService executor = Executors.newFixedThreadPool(
    Runtime.getRuntime().availableProcessors()
);

This is only a rule of thumb for CPU-intensive tasks. I/O-bound tasks spend part of their time waiting, so a different strategy may be more appropriate.

Java also provides a work-stealing executor:

ExecutorService executor = Executors.newWorkStealingPool();

newWorkStealingPool() returns an ExecutorService backed by a ForkJoinPool, not a ThreadPoolExecutor. It uses the number of available processors as its target parallelism level by default.

If we specifically have a ThreadPoolExecutor, it offers monitoring methods such as getPoolSize() (current number of worker threads in the pool) and getActiveCount() (an approximate number of threads actively executing tasks). shutdownNow() should also be interpreted carefully: it prevents waiting tasks from starting and makes a best-effort attempt to stop active tasks, usually by interrupting their threads. It does not guarantee that active tasks finish immediately.

1.2. Virtual-thread executors (Java 21+)

As we saw in the previous document, current Java versions also provide virtual threads. A convenient way of using them is through an executor that creates one virtual thread for every submitted task:

ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();

for (int i = 0; i < 1000; i++)
{
    executor.submit(() -> {
        // Task that may spend a lot of time waiting for I/O
    });
}

executor.shutdown();

Unlike a fixed thread pool, this executor does not reuse a small set of platform threads: each task gets its own virtual thread. This is especially useful when we have a very large number of mostly blocking tasks, for example network requests, database operations or file I/O.

Virtual threads are not intended to make CPU-intensive calculations run faster than the available processor cores. For CPU-bound parallel computations, bounded pools and the Fork/Join framework are usually more appropriate.

2. Using Callables and CompletableFutures

In this section we are going to see alternatives to plain Runnable tasks. Depending on whether we need a returned value, cancellation, completion tracking or dependent processing, we can use Callable, Future and CompletableFuture.

2.1. Using Callable

In addition to Runnable, executors support another kind of task named Callable. Callable is a functional interface similar to Runnable, but its call() method returns a value and may throw an exception. The Callable interface defines the type of returned data using generics.

In this example, we are going to create a Callable task and submit it to an ExecutorService. ThreadPoolExecutor and ForkJoinPool, among others, implement ExecutorService. When we submit a Callable, the executor returns a Future object that represents the pending result of the task.

Calling get() on the Future blocks the calling thread until the task completes (unless it has already completed). isDone() can be useful for monitoring, but polling it repeatedly is not required before calling get(). We use submit() instead of execute() when we need a Future to track a task, retrieve its result or cancel it.

public static void main(String[] args) 
{
    Callable<Integer> callInt = () -> {
        try 
        {
            TimeUnit.SECONDS.sleep(3);
            return 20;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("task interrupted", e);
        }
    };

    ExecutorService executor = Executors.newFixedThreadPool(1); 

    // Calling submit executes the thread and returns a Future
    Future<Integer> future = executor.submit(callInt);

    executor.shutdown();

    System.out.println("future done? " + future.isDone());
    Integer result;
    try 
    {
        result = future.get(); // It BLOCKS main thread until it returns!
        System.out.println("future done? " + future.isDone());
        System.out.println("Result: " + result); // Prints 20 
    } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
    } catch (ExecutionException ex) {
        System.err.println("Task failed: " + ex.getCause());
    }
}

Passing a timeout

When calling get() on the Future object to retrieve the result, we can pass a timeout, so when that time passes, if the thread hasn’t finished, it will throw a TimeoutException. It’s also a good idea to cancel the task when that happens:

try 
{
    result = future.get(1, TimeUnit.SECONDS); // Blocks 1 second maximum
    System.out.println("Result: " + result);
} catch (InterruptedException ex) {
    Thread.currentThread().interrupt();
} catch (ExecutionException ex) {
    System.err.println("Task failed: " + ex.getCause());
} catch (TimeoutException ex) { // When the timeout expires...
    System.err.println("The task took more than 1 second to complete!");
    future.cancel(true); // Request interruption of the task if it is running
}

Launching several Callable tasks at the same time

We can submit more than one Callable task at the same time using an executor. If we pass a collection of Callable objects to invokeAll(), the method waits until all tasks complete (or the calling thread is interrupted) and returns a list of Future objects containing their status and results.

public static Callable<Integer> getSumCallable(int num1, int num2, 
    int secondsSleep) 
{
    return () -> {
        try 
        {
            TimeUnit.SECONDS.sleep(secondsSleep);
            return num1 + num2;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("task interrupted", e);
        }
    };
}

public static void main(String[] args) 
{
    List<Callable<Integer>> callables = Arrays.asList(
        getSumCallable(3, 6, 2),
        getSumCallable(5, 8, 3),
        getSumCallable(12, 3, 1)
    );

    ExecutorService executor = Executors.newWorkStealingPool(); 
    List<Future<Integer>> futures;
    try 
    {
        futures = executor.invokeAll(callables);
        futures.forEach(future -> {
            try 
            { 
                System.out.println(future.get()); 
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } catch (ExecutionException e) {
                System.err.println("Task failed: " + e.getCause());
            }
        }); 
    } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
    } finally {
        executor.shutdown();
    }
}

In this case, we have created a static method that returns the Callable object. We call this method many times to add many callables to our list. Then we invoke all of them from the executor. You can download here the source code of this example.

If we don’t want to wait for every task result and instead want the result of one task that completes successfully, we can use invokeAny(). This method blocks until one task completes successfully and returns its result directly (not a Future). When the method returns, tasks that have not completed are cancelled. Notice that the first task to finish is not necessarily the one returned if that task finishes with an exception.

ExecutorService executor = Executors.newWorkStealingPool(); 
try 
{
    // Blocks and returns one successful result
    int firstResult = executor.invokeAny(callables); 
    System.out.println(firstResult); // 15 -> 12 + 3 finishes in 1 second
} catch (InterruptedException ex) {
    Thread.currentThread().interrupt();
} catch (ExecutionException ex) {
    System.err.println("No task completed successfully");
} finally {
    executor.shutdown();
}

2.2. Scheduled executors

If we wanted to run a task periodically, instead of doing it manually, we could use a ScheduledExecutorService. First of all, we’ll see an example of a task that doesn’t run periodically, but instead has a delay and waits for a number of seconds before starting. This kind of executor service returns an ScheduledFuture object.

ScheduledExecutorService executor = Executors.newScheduledThreadPool(1); 
try 
{
    // Usage: schedule(Callable/Runnable, delay, Time unit)
    ScheduledFuture<Integer> schedFuture = executor.schedule(
        getSumCallable(3, 6, 2), 3, TimeUnit.SECONDS); 

    executor.shutdown();
    TimeUnit.MILLISECONDS.sleep(1500); // Sleeps for about 1.5 seconds
    long remainingDelay = schedFuture.getDelay(TimeUnit.MILLISECONDS);
    System.out.printf("Remaining Delay: %dms\n", remainingDelay); 
    // 1498ms
    int result = schedFuture.get(); 
    // blocks 3.5 sec. (1.5 delay + 2 task)
    System.out.println("Result: " + result); 
} catch (InterruptedException ex) {
    Thread.currentThread().interrupt();
} catch (ExecutionException ex) {
    System.err.println("Task failed: " + ex.getCause());
}

To run a periodic task, we can call one of these two methods: scheduleAtFixedRate() or scheduleWithFixedDelay().

scheduleAtFixedRate() uses a fixed planned period between starting times. For example, with a period of 3 seconds, the planned starts are at approximately 0, 3, 6, 9… seconds after the initial delay. However, executions of the same periodic task do not overlap. If one execution takes longer than its period, the next execution starts late, after the previous one has finished.

scheduleWithFixedDelay() works differently: the specified delay is measured from the end of one execution to the start of the next one. This is useful when we always want to leave a certain pause between executions.

Periodic tasks continue until they are cancelled, the executor terminates, or one execution ends exceptionally. We should therefore keep the returned ScheduledFuture if we want to cancel one particular periodic task, and avoid shutting down the executor until we no longer need it.

public static void main(String[] args) 
{
    Runnable task = () -> {
        System.out.println("Time now: " + LocalTime.now().toString());
    };

    ScheduledExecutorService executor =
        Executors.newScheduledThreadPool(1);
    // Delay (1 second), runs every 3 seconds
    executor.scheduleWithFixedDelay(task, 1, 3, TimeUnit.SECONDS);
    BufferedReader in = new BufferedReader(
        new InputStreamReader(System.in));
    String command;
    try 
    {
        do 
        { 
            // When user presses "q" and "enter", program will end
            command = in.readLine();
        } while (!command.equals("q"));
    } catch (IOException ex) {
        System.err.println("Input error: " + ex.getMessage());
    } finally {
        executor.shutdown(); // Stop future periodic executions
    }
}

Exercise 1:

Create a project named CallableWordCounting. Launch 3 Callable tasks at the same time using the executor’s invokeAll() method.

Each thread will read a different text file (create 3 text files with a lot of text inside, or use these ones) and search how many times a text appears in that file. At the end return the number of times the text has appeared in the file (Integer).

Hint: Create a class that implements Callable<Integer> and pass to the constructor the file name and the text to search, or create a static method that receives these 2 parameters and returns a Callable lambda.

The main thread will get the results and add them, printing the total number of times the word or text has appeared in all files (notice that you don’t need any synchronized section or variable for this exercise)

2.3. Using CompletableFuture

CompletableFuture is a class that implements the Future and CompletionStage interfaces. A CompletableFuture lets us describe dependent stages that should run when a previous stage completes, so we do not need to block a thread with Future.get() just to continue processing a result. Its factory methods runAsync() and supplyAsync() accept Runnable-like and Supplier tasks respectively.

When we use asynchronous methods without explicitly passing an Executor (for example, runAsync() or supplyAsync()), Java normally uses ForkJoinPool.commonPool() as the default asynchronous execution facility. We can also pass our own executor when we need explicit control over where the task runs.

This example uses a runnable (this is, a CompletableFuture that does not return anything) to print a message in the screen after 3 seconds. We launch it with runAsync method.

public static void main(String[] args) 
{
    // runAsync receives a runnable that doesn't return anything
    CompletableFuture<Void> compRunnable =
            CompletableFuture.runAsync(() -> {
        try
        {
            TimeUnit.SECONDS.sleep(3);
            System.out.println("Task completed");
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
        }
    });

    // thenRun adds a dependent action that runs when the current stage finishes.
    // It is NOT guaranteed to run in the main thread.
    compRunnable.thenRun(() -> 
        System.out.println("CompletableFuture finish"));

    InputStreamReader in = new InputStreamReader(System.in);
    System.out.println("Press enter to exit (let the task finish first)");
    try 
    {
        in.read();
    } catch (IOException ex) { }
}

If we want the asynchronous computation to return a value, we can use supplyAsync() instead of runAsync(). supplyAsync() receives a Supplier and stores its eventual result in a CompletableFuture. We can then attach dependent stages that process that result. Notice that methods without the Async suffix, such as thenAccept() or thenApply(), are not guaranteed to execute in a new thread; their execution follows the rules of CompletionStage. If we specifically want asynchronous execution, we can use methods such as thenAcceptAsync() or thenApplyAsync(), optionally passing an executor.

public static void main(String[] args) 
{
    CompletableFuture<Integer> compRunnable =
                CompletableFuture.supplyAsync(
    () -> {
        try 
        {
            TimeUnit.SECONDS.sleep(3);
            // Return random 0 - 100 (inclusive)
            return (new Random()).nextInt(101); 
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
            return -1;
        }
    });

    // thenAccept receives the result of the previous task to process
    compRunnable.thenAccept((num) -> 
        System.out.println("Number generated: " + num));

    InputStreamReader in = new InputStreamReader(System.in);
    System.out.println(
            "Press enter to exit (let the task finish first)");
    try 
    {
        in.read();
    } catch (IOException ex) { }
}

CompletableFuture allows us to build a pipeline of dependent stages. This resembles method chaining in streams, but the purpose is different: a CompletableFuture represents the eventual completion of asynchronous or dependent computations. We can use thenAccept() when a stage consumes a result without returning another value, or thenApply() when it transforms the previous result and returns a new one.

This example uses a CompletableFuture to get a string formatted with a person name and an age (separated by a semicolon). Once the data is obtained, it adds a dependent stage to split the string and return a Person object with these attributes. Finally, it adds another dependent stage to print the person on the screen. These stages are not necessarily executed in separate threads because we are using the non-Async methods.

public static void main(String[] args)
{
    CompletableFuture<String> compRunnable =
                CompletableFuture.supplyAsync(
    () -> {
        try 
        {
            TimeUnit.SECONDS.sleep(3);
            return "Peter;28";
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
            return "Error;0";
        }
    });

    // thenApply gets the previous result and returns another (Person)
    CompletableFuture<Person> comPerson = compRunnable.thenApply((str) -> 
    {
        String[] parts = str.split(";");
        return new Person(parts[0], Integer.parseInt(parts[1]));
    });

    // thenAccept consumes the result of the previous stage
    comPerson.thenAccept((person) -> System.out.println(person));

    InputStreamReader in = new InputStreamReader(System.in);
    System.out.println(
        "Press enter to exit (let the task finish first)");
    try 
    {
        in.read();
    } catch (IOException ex) { }
}

What happens if the original task throws an exception and we want to recover from it? (return valid data that can be processed). We can use exceptionally method. This method will act like a catch statement for exceptions that are thrown in the previous task (it must return the same type of data as the previous task). In this example, we’ll see how we can chain all these tasks without using intermediate variables:

public static void main(String[] args)
{
    CompletableFuture.supplyAsync(() -> {
        try
        {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
            return "Error;0";
        }
        return "Peter;28";
    }).exceptionally((error) -> { 
        // Only if previous step throws an error
        System.err.println("Error: " + error.getMessage());
        return "Error;0";
    }).thenApply((str) -> { 
        // Process the string and return a Person
        String[] parts = str.split(";");
        return new Person(parts[0], Integer.parseInt(parts[1]));
    }).thenAccept((person) -> System.out.println(person)); 

    InputStreamReader in = new InputStreamReader(System.in);
    System.out.println(
        "Press enter to exit (let the task finish first)");
    try 
    {
        in.read();
    } catch (IOException ex) { }
}

Finally (although this CompletableFuture API has many more possibilities), we’ll see how to execute a task when a number (greater than 1) of CompletableFutures end their tasks, using CompletableFuture.allOf. In this example, we’ll create tasks that try to ping some servers and at the end, we’ll show the results and end the program.

public class ThreadsExamples 
{
    public static Deque<String> ipMessages = 
        new ConcurrentLinkedDeque<>();

    public static CompletableFuture<Void> pingIp(String address) 
    {
        return CompletableFuture.supplyAsync(() -> { 
            try 
            {
                InetAddress inet = InetAddress.getByName(address);
                if(inet.isReachable(4000)) // 4 seconds
                { 
                    return true;
                }
            } catch (UnknownHostException ex) { 
            } catch (IOException ex) { }
            return false;
        }).thenAccept(result -> {
            ipMessages.add(address + (result?" ping OK":" unreachable"));
        });
    }

    public static void main(String[] args) 
    {
        CompletableFuture<Void> allTasks = CompletableFuture.allOf(
            pingIp("google.es"),
            pingIp("iessanvicente.com"),
            pingIp("apache.org"),
            pingIp("facebook.com")
        );

        CompletableFuture<Void> finished = allTasks.thenRun(() -> {
            System.out.println("All tasks finished");
            System.out.println(ipMessages);
        });

        // Wait only because this simple console application must not exit yet
        finished.join();
    }
}

If instead of allOf(), we use CompletableFuture.anyOf(), the returned future completes when any of the supplied futures completes. Unlike allOf(), anyOf() returns a CompletableFuture<Object> because the supplied futures may have different result types. We can use thenAccept() to receive the first completed result. You can download here the source code from previous examples.

Exercise 2:

Create a project called FastestWordCounting. This exercise will be similar to previous Exercise 1, but with some differences.

The first thread has finished and found the text “cat” 24 times.

3. Synchronizing with Lock

Since Java 5, the java.util.concurrent.locks package provides explicit lock objects as an alternative to the synchronized keyword for some synchronization scenarios. The Lock interface defines operations for acquiring and releasing a lock, and ReentrantLock is its most commonly used implementation.

private final Lock myLock = new ReentrantLock();

public void myMethod()
{
    myLock.lock();
    try
    {
        ... // Critical section
    }
    finally
    {
        myLock.unlock();
    }
}

The finally block is important: it guarantees that the lock is released even if the code inside the critical section throws an exception. ReentrantLock also offers features that intrinsic synchronized blocks do not provide directly, such as tryLock(), interruptible lock acquisition and optional fairness policies.

Exercise 3:

Create a project call BankAccountLock, that is a copy of the project created in previous documents (BankAccountSynchronized). Replace the old synchronized methods with the Lock mechanisms that we have just seen, and check that everything goes OK.

3.1. Read / Write locks

Besides, there is an improvement brought by this Lock interface: the possibility of having read and write operations working separately, so that there can be multiple read operations running at the same time on a given file or resource, but only one write operation (when a thread is writing, no one else can be reading or writing). We can achieve this with the ReadWriteLock interface and its implementation in ReentrantReadWriteLock class. This class has two locks, one for reading operations and one for writing operations, so that we can use any of them depending on the operation we actually want to do.

ReadWriteLock lock = new ReentrantReadWriteLock();
...
public void readOperation()
{
    lock.readLock().lock();
    try
    {
        ... // Multiple readers may execute this area concurrently
    }
    finally
    {
        lock.readLock().unlock();
    }
}

public void writeOperation()
{
    lock.writeLock().lock();
    try
    {
        ... // A writer needs exclusive access
    }
    finally
    {
        lock.writeLock().unlock();
    }
}

As you can see in the code below, the read lock allows to lock an object for reading, so that any other reading operation can also get to the critical section. However, when a write lock wants to be set, no other lock can be currently applied. In other words, we can have multiple readers running the critical section at the same time, but whenever a writer is running the critical section, no other thread can be running it.

Let’s see how it works with the following example: we are going to create a class that stores an integer value:

import java.util.concurrent.locks.ReentrantReadWriteLock;

public class MyData
{
    int value;
    ReentrantReadWriteLock lock;

    public MyData(int value)
    {
        this.value = value;
        lock = new ReentrantReadWriteLock();
    }

    public int getValue()
    {
        lock.readLock().lock();
        try
        {
            try
            {
                Thread.sleep(2000);
            }
            catch (InterruptedException e)
            {
                Thread.currentThread().interrupt();
            }
            System.out.println("Thread #" + Thread.currentThread().threadId() +
                " reads value " + value);
            return value;
        }
        finally
        {
            lock.readLock().unlock();
        }
    }

    public void setValue(int value)
    {
        lock.writeLock().lock();
        try
        {
            try
            {
                Thread.sleep(2000);
            }
            catch (InterruptedException e)
            {
                Thread.currentThread().interrupt();
            }
            System.out.println("Thread #" + Thread.currentThread().threadId() +
                " sets value to " + (this.value + value));
            this.value += value;
        }
        finally
        {
            lock.writeLock().unlock();
        }
    }
}

Now, we define a thread class that tries to read it, and a thread class that tries to change its value:

public class ReadingThread extends Thread
{
    MyData sharedData;

    public ReadingThread(MyData sharedData)
    {
        this.sharedData = sharedData;
    }

    @Override
    public void run()
    {
        int value = sharedData.getValue();
    }
}

public class WritingThread extends Thread
{
    MyData sharedData;

    public WritingThread(MyData sharedData)
    {
        this.sharedData = sharedData;
    }

    @Override
    public void run()
    {
        sharedData.setValue(10);
    }
}

If we run a main program like this one:

MyData mds = new MyData(10);
ReadingThread[] threadsR = new ReadingThread[5];
WritingThread[] threadsW = new WritingThread[2];

for (int i = 0; i < threadsW.length; i++)
{
    threadsW[i] = new WritingThread(mds);
}

for (int i = 0; i < threadsR.length; i++)
{
    threadsR[i] = new ReadingThread(mds);
}

for (int i = 0; i < threadsW.length; i++)
{
    threadsW[i].start();
}
for (int i = 0; i < threadsR.length; i++)
{
    threadsR[i].start();
}

we will notice that several reading threads may execute the read section concurrently, whereas a writing thread needs exclusive access. The exact order and timing are not guaranteed, so the output may look like this (it can differ on every run):

Thread #9 sets value to 20
Thread #13 reads value 20
Thread #11 reads value 20
Thread #14 reads value 20
Thread #12 reads value 20
Thread #15 reads value 20
Thread #10 sets value to 30

Exercise 4:

Create a project called ReadersWritersLock and copy the previous example on it. Make changes to the code so that there are 10 reading threads (instead of 5), and each thread (reader or writer) will sleep a random number of seconds (between 1 and 10), and then it will do its job. This way, there should be some reading operations at the beginning, some in the middle of the two writings, and some at the end. Your output should look like this one:

Thread #13 reads value 10
Thread #11 reads value 10
Thread #9 sets value to 20
Thread #14 reads value 20
Thread #12 reads value 20
Thread #15 reads value 20
Thread #10 sets value to 30
Thread #16 reads value 30
Thread #18 reads value 30
...

4. The Fork/Join framework

When should we use it? Fork/Join is mainly aimed at CPU-bound problems that can be recursively divided into smaller independent computations. It is not normally the best choice for a large number of blocking I/O operations; virtual threads or other I/O-oriented designs are usually more suitable for that scenario.

The executor framework introduced in Java 5 lets us separate task submission from thread management. Since Java 7, Java also provides the Fork/Join framework for recursively divisible parallel computations.

With this framework, we can divide complex or big problems into smaller ones. The framework is based on two key operations: fork (arrange for a subtask to execute asynchronously) and join (wait for a task to complete and obtain its result, if any). Fork/Join works best when tasks can be split into mostly independent subtasks, reducing the need for shared mutable state.

The Fork/Join framework relies on ForkJoinPool (which manages and executes tasks) and ForkJoinTask (the base class for Fork/Join tasks). Two commonly used subclasses are RecursiveAction (for tasks that do not return a result) and RecursiveTask<V> (for tasks that return a result). These classes belong to the java.util.concurrent package.

4.1. Example: tasks that do not return any result

Let’s see how this framework can be used with the following example: we are going to create a list of video games, with their titles and prices. Then, we are going to look for a given title in the list, so that, if the list size is smaller than 5 video games, only one task will be needed, but if not, a task will be created to search a subset of up to 5 video games from the list.

Our VideoGame class would be like this one:

public class VideoGame 
{
    String title;
    float price;

    public VideoGame(String title, float price)
    {
        this.title = title;
        this.price = price;
    }

    public String getTitle() 
    {
        return title;
    }

    public float getPrice() 
    {
        return price;
    }
}

Our thread or task to search in the list would be like this one:

public class GameSearch extends RecursiveAction
{
    /* How many video games will each task be in charge of? */
    public static final int MAX_GAMES = 5;
    /* List of video games */
    ArrayList<VideoGame> list;

    /* First index of the list to search */
    int first;

    /* Last index of the list to search */
    int last;

    /* Text to be searched in the list */
    String text;

    public GameSearch(ArrayList<VideoGame> list, String text, int first,
        int last)
    {
        this.list = list;
        this.text = text;
        this.first = first;
        this.last = last;
    }

    @Override
    protected void compute()
    {
        if (last - first <= MAX_GAMES)
            search();
        else
        {
            int middle = (first + last) / 2;
            System.out.println("Creating 2 subtasks...");
            GameSearch s1 = new GameSearch(list, text, first, middle);
            GameSearch s2 = new GameSearch(list, text, middle, last);
            invokeAll(s1, s2);
        }
    }

    public void search()
    {
        for (int i = first; i < last; i++)
        {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            }
            if (list.get(i).getTitle().contains(text))
                System.out.println("Found at position " + i + ": " +
                    list.get(i).getTitle());
        }
    }
}

Notice that, when we extend RecursiveAction class, we need to define a compute method. This would be the equivalent to the run method in common threads. Inside this method, we check the size of the game list. If it is smaller than 5, we just call the search method to solve the problem. Otherwise, we divide the list in two parts and create two subtasks; each one will be in charge of searching in one half of the list.

We can also create a list of tasks, and call the invokeAll method passing that list as a parameter:

ArrayList<GameSearch> subtasks = new ArrayList<>();
...
subtasks.add(new GameSearch(...));
subtasks.add(new GameSearch(...));
subtasks.add(new GameSearch(...));

invokeAll(subtasks);

From our main program, we create the video game list, create a GameSearch task to look for the word “Assassin’s”, and launch it in the Fork/Join pool, as we did before with thread executors:

public static void main(String[] args) 
{
    ArrayList<VideoGame> list = new ArrayList<VideoGame>();
    list.add(new VideoGame("Assassin's Creed", 19.95f));
    list.add(new VideoGame("The last of us", 49.90f));
    list.add(new VideoGame("Fifa 14", 39.95f));
    list.add(new VideoGame("Far Cry 2", 14.95f));
    list.add(new VideoGame("Watchdogs", 59.95f));
    list.add(new VideoGame("Assassin's Creed II", 24.90f));
    list.add(new VideoGame("Far Cry 3", 39.50f));
    list.add(new VideoGame("Borderlands", 19.90f));

    GameSearch v = new GameSearch(list, "Assassin's", 0, list.size());
    ForkJoinPool pool = new ForkJoinPool();
    pool.invoke(v); // Waits until the task completes
    pool.shutdown();
}

Here we use pool.invoke(v), which starts the task and waits until it completes. This is preferable to repeatedly polling isDone() when the main thread has no other work to perform.

4.2. Example: tasks that return a result

How could we adapt the previous example so that tasks do not print anything to the output, and return a set or list of results found? We have to use a subclass of RecursiveTask instead of a subclass of RecursiveAction. When we extend RecursiveTask, we have to take into account that it is a parameterized class, this is, we need to provide the type of result that will be returned. So our GameSearch class from previous example would look like this one now:

public class GameSearch extends RecursiveTask<ArrayList<String>>
{
    /* How many video games will each task be in charge of? */
    public static final int MAX_GAMES = 5;
    /* List of video games */
    ArrayList<VideoGame> list;
    /* First index of the list to search */
    int first;
    /* Last index of the list to search */
    int last;
    /* Text to be searched in the list */
    String text;

    public GameSearch(ArrayList<VideoGame> list, String text, int first, 
    int last)
    {
        this.list = list;
        this.text = text;
        this.first = first;
        this.last = last;
    }

    @Override
    protected ArrayList<String> compute()
    {
        ArrayList<String> results = new ArrayList<String>();
        if (last - first <= MAX_GAMES)
            results = search();
        else
        {
            int middle = (first + last)/2;
            System.out.println("Creating 2 subtasks...");
            GameSearch s1 = new GameSearch(list, text, first, middle);
            GameSearch s2 = new GameSearch(list, text, middle, last);
            invokeAll(s1, s2);
            results = s1.join();
            ArrayList<String> aux = s2.join();
            results.addAll(aux);
        }
        return results;
    }

    public ArrayList<String> search()
    {
        ArrayList<String> results = new ArrayList<String>();
        for (int i = first; i < last; i++)
        {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return results;
            }
            if (list.get(i).getTitle().contains(text))
                results.add("Found at " + i + ": " + 
                    list.get(i).getTitle());
        }
        return results;
    }
}

We are going to return an ArrayList of String values as a result, each one containing an occurrence of the searched text. In search() we create the list of matching games and return it. In compute() we call search() directly if the range is small enough; otherwise, we split the work into two tasks, wait for both with invokeAll(), obtain their results with join(), and combine them.

Our main program will get the results after the main task has finished, and it will print them to the standard output:

public static void main(String[] args) 
{
    ...
    ForkJoinPool pool = new ForkJoinPool();
    ArrayList<String> results = pool.invoke(v);

    for (String result : results)
        System.out.println(result);

    pool.shutdown();
}

Exercise 5:

Create a project called ForkJoinFile. Create a text file in it, with some lines (at least 50, you can copy them from any source). Use the Fork/Join framework to create tasks that will check the contents of the text file (up to 10 lines for each task). The tasks must replace every occurrence of the word “java” with “Java” (of course, try to add some occurrences of the word “java” in the text file). At the end, main program will get the results of all the subtasks (i.e., the pieces of text with the replacements done), will join them and will rewrite the text file with the updated text.

4.3. Launching asynchronous subtasks

In the examples shown above, invokeAll() does not return until all the supplied subtasks are done. Sometimes we want to fork work first, continue doing useful work in the current task, and only join a subtask when its result is actually needed. We can do this explicitly with fork() and join(). For example:

@Override
protected void compute()
{
    if (last - first <= MAX_GAMES)
        search();
    else
    {
        int middle = (first + last)/2;
        System.out.println("Creating 2 subtasks...");
        GameSearch s1 = new GameSearch(list, text, first, middle);
        GameSearch s2 = new GameSearch(list, text, middle, last);
        s1.fork();
        s2.fork();
        // At this point, this task continues running its code
        ...
        // Wait for the 1st subtask to finish
        s1.join();
        ...
        // Wait for the 2nd subtask to finish
        s2.join();
    }
}