Concurrency

61. The Fork/Join Framework

ForkJoinPool and RecursiveTask/RecursiveAction implement divide-and-conquer parallelism using work stealing — idle threads steal tasks from busy threads' queues, maximising CPU utilisation for CPU-bound parallel computation.

Java 21 (core language)

Explanation

The Fork/Join framework (Java 7+) is designed for recursive, divide-and-conquer problems: split a big task into subtasks (fork), solve them in parallel, and combine results (join). The underlying ForkJoinPool uses work-stealing: each worker thread has a deque; idle threads steal from the tail of busy threads' deques.

RecursiveTask<T> for tasks that produce a result; RecursiveAction for tasks with no return value. Override compute(): if the problem is small enough, solve it directly; otherwise fork two subtasks and join their results.

ForkJoinPool.commonPool() is a shared pool used by parallel streams (stream().parallel()) and CompletableFuture.supplyAsync(). Blocking inside the common pool starves other tasks — prefer virtual threads for blocking work.

Example

import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;

public class ForkJoinDemo {
    // Parallel merge sort using fork/join
    static class MergeSort extends RecursiveTask<int[]> {
        private final int[] arr;
        private static final int THRESHOLD = 16;

        MergeSort(int[] arr) { this.arr = arr; }

        @Override
        protected int[] compute() {
            if (arr.length <= THRESHOLD) {
                // Base case: sort sequentially
                int[] sorted = arr.clone();
                java.util.Arrays.sort(sorted);
                return sorted;
            }
            int mid = arr.length / 2;
            MergeSort left  = new MergeSort(java.util.Arrays.copyOfRange(arr, 0, mid));
            MergeSort right = new MergeSort(java.util.Arrays.copyOfRange(arr, mid, arr.length));

            left.fork();               // start left in another thread
            int[] rightResult = right.compute(); // compute right in this thread
            int[] leftResult  = left.join();     // wait for left result

            return merge(leftResult, rightResult);
        }

        static int[] merge(int[] a, int[] b) {
            int[] result = new int[a.length + b.length];
            int i = 0, j = 0, k = 0;
            while (i < a.length && j < b.length)
                result[k++] = a[i] <= b[j] ? a[i++] : b[j++];
            while (i < a.length) result[k++] = a[i++];
            while (j < b.length) result[k++] = b[j++];
            return result;
        }
    }

    public static void main(String[] args) {
        int[] data = new int[1024];
        java.util.Random rng = new java.util.Random(42);
        for (int i = 0; i < data.length; i++) data[i] = rng.nextInt(10000);

        ForkJoinPool pool = ForkJoinPool.commonPool();
        int[] sorted = pool.invoke(new MergeSort(data));

        // Verify sorted
        for (int i = 0; i < sorted.length - 1; i++)
            if (sorted[i] > sorted[i+1]) { System.out.println("NOT SORTED!"); return; }
        System.out.println("Sorted " + sorted.length + " elements correctly");
        System.out.println("Pool parallelism: " + pool.getParallelism());
    }
}

Sources