The Fork/Join Framework is a Java implementation of the divide-and-conquer paradigm, designed to efficiently parallelize recursive algorithms using work-stealing.
1. Overview
What is Fork/Join?
- Part of
java.util.concurrentpackage (since Java 7) - Designed for recursive, divide-and-conquer algorithms
- Uses work-stealing algorithm for load balancing
- Ideal for CPU-intensive tasks
Key Components
ForkJoinPool- Specialized thread poolForkJoinTask- Base class for tasksRecursiveTask- Returns a resultRecursiveAction- No result returned
2. Basic Architecture
Work-Stealing Algorithm
Thread 1: [Task A] -> splits into [A1, A2, A3] Thread 2: (idle) steals A3 from Thread 1 Thread 3: (idle) steals A2 from Thread 1
3. ForkJoinPool
Creating ForkJoinPool
import java.util.concurrent.ForkJoinPool;
public class ForkJoinPoolCreation {
public static void main(String[] args) {
// Method 1: Use common pool (recommended for most cases)
ForkJoinPool commonPool = ForkJoinPool.commonPool();
// Method 2: Create custom pool
ForkJoinPool customPool = new ForkJoinPool(4); // 4 threads
// Method 3: Create with parallelism level and factory
ForkJoinPool configuredPool = new ForkJoinPool(
Runtime.getRuntime().availableProcessors(),
ForkJoinPool.defaultForkJoinWorkerThreadFactory,
null, // default handler
false // async mode?
);
System.out.println("Common pool parallelism: " + commonPool.getParallelism());
System.out.println("Available processors: " + Runtime.getRuntime().availableProcessors());
}
}
4. RecursiveTask (Returns Result)
Factorial Calculation Example
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;
public class FactorialTask extends RecursiveTask<Long> {
private final int start;
private final int end;
private static final int THRESHOLD = 5; // Sequential computation threshold
public FactorialTask(int n) {
this(1, n);
}
private FactorialTask(int start, int end) {
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
int length = end - start + 1;
// If problem is small enough, compute sequentially
if (length <= THRESHOLD) {
return computeSequentially();
}
// Split task into subtasks
int mid = start + (end - start) / 2;
FactorialTask leftTask = new FactorialTask(start, mid);
FactorialTask rightTask = new FactorialTask(mid + 1, end);
// Fork the left task (execute asynchronously)
leftTask.fork();
// Compute right task and wait for left task
Long rightResult = rightTask.compute();
Long leftResult = leftTask.join();
// Combine results
return leftResult * rightResult;
}
private Long computeSequentially() {
long result = 1;
for (int i = start; i <= end; i++) {
result *= i;
// Simulate some work
try { Thread.sleep(10); } catch (InterruptedException e) {}
}
System.out.println(Thread.currentThread().getName() +
" computed " + start + " to " + end);
return result;
}
public static void main(String[] args) {
ForkJoinPool pool = ForkJoinPool.commonPool();
FactorialTask task = new FactorialTask(10);
Long result = pool.invoke(task);
System.out.println("Factorial result: " + result);
System.out.println("Pool parallelism: " + pool.getParallelism());
}
}
Array Sum Calculation
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;
import java.util.Random;
public class ArraySumTask extends RecursiveTask<Long> {
private final int[] array;
private final int start;
private final int end;
private static final int THRESHOLD = 1000;
public ArraySumTask(int[] array) {
this(array, 0, array.length);
}
private ArraySumTask(int[] array, int start, int end) {
this.array = array;
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
int length = end - start;
if (length <= THRESHOLD) {
return computeSequentially();
}
int mid = start + length / 2;
ArraySumTask leftTask = new ArraySumTask(array, start, mid);
ArraySumTask rightTask = new ArraySumTask(array, mid, end);
leftTask.fork();
Long rightResult = rightTask.compute();
Long leftResult = leftTask.join();
return leftResult + rightResult;
}
private Long computeSequentially() {
long sum = 0;
for (int i = start; i < end; i++) {
sum += array[i];
}
return sum;
}
public static void main(String[] args) {
// Create large array
int[] array = new int[100000];
Random random = new Random();
for (int i = 0; i < array.length; i++) {
array[i] = random.nextInt(100);
}
// Parallel computation
long startTime = System.currentTimeMillis();
ForkJoinPool pool = ForkJoinPool.commonPool();
ArraySumTask task = new ArraySumTask(array);
Long parallelSum = pool.invoke(task);
long parallelTime = System.currentTimeMillis() - startTime;
// Sequential computation for comparison
startTime = System.currentTimeMillis();
long sequentialSum = 0;
for (int value : array) {
sequentialSum += value;
}
long sequentialTime = System.currentTimeMillis() - startTime;
System.out.println("Parallel sum: " + parallelSum + " in " + parallelTime + "ms");
System.out.println("Sequential sum: " + sequentialSum + " in " + sequentialTime + "ms");
System.out.println("Results match: " + (parallelSum == sequentialSum));
}
}
5. RecursiveAction (No Result)
Array Processing Example
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.ForkJoinPool;
import java.util.Random;
public class ArrayTransformAction extends RecursiveAction {
private final double[] array;
private final int start;
private final int end;
private static final int THRESHOLD = 10000;
public ArrayTransformAction(double[] array) {
this(array, 0, array.length);
}
private ArrayTransformAction(double[] array, int start, int end) {
this.array = array;
this.start = start;
this.end = end;
}
@Override
protected void compute() {
int length = end - start;
if (length <= THRESHOLD) {
transformSequentially();
return;
}
int mid = start + length / 2;
ArrayTransformAction leftTask = new ArrayTransformAction(array, start, mid);
ArrayTransformAction rightTask = new ArrayTransformAction(array, mid, end);
invokeAll(leftTask, rightTask); // Fork and wait for both
}
private void transformSequentially() {
for (int i = start; i < end; i++) {
// Apply some transformation
array[i] = Math.sqrt(Math.sin(array[i]) * Math.cos(array[i]));
}
System.out.println(Thread.currentThread().getName() +
" processed " + start + " to " + end);
}
public static void main(String[] args) {
double[] array = new double[100000];
Random random = new Random();
for (int i = 0; i < array.length; i++) {
array[i] = random.nextDouble() * 100;
}
ForkJoinPool pool = new ForkJoinPool();
ArrayTransformAction task = new ArrayTransformAction(array);
pool.invoke(task);
System.out.println("Array transformation completed");
// Verify some values
for (int i = 0; i < 5; i++) {
System.out.printf("array[%d] = %.4f%n", i, array[i]);
}
}
}
6. Fibonacci Sequence (Classic Example)
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;
public class FibonacciTask extends RecursiveTask<Long> {
private final int n;
public FibonacciTask(int n) {
this.n = n;
}
@Override
protected Long compute() {
if (n <= 1) {
return (long) n;
}
// For very small n, compute sequentially to avoid overhead
if (n <= 10) {
return computeSequentially(n);
}
FibonacciTask leftTask = new FibonacciTask(n - 1);
FibonacciTask rightTask = new FibonacciTask(n - 2);
leftTask.fork();
Long rightResult = rightTask.compute();
Long leftResult = leftTask.join();
return leftResult + rightResult;
}
private Long computeSequentially(int n) {
if (n <= 1) return (long) n;
long a = 0, b = 1, result = 0;
for (int i = 2; i <= n; i++) {
result = a + b;
a = b;
b = result;
}
return result;
}
public static void main(String[] args) {
int n = 25;
// Parallel computation
long startTime = System.currentTimeMillis();
ForkJoinPool pool = ForkJoinPool.commonPool();
FibonacciTask task = new FibonacciTask(n);
Long parallelResult = pool.invoke(task);
long parallelTime = System.currentTimeMillis() - startTime;
// Sequential computation
startTime = System.currentTimeMillis();
Long sequentialResult = new FibonacciTask(n).computeSequentially(n);
long sequentialTime = System.currentTimeMillis() - startTime;
System.out.println("Fibonacci(" + n + ") = " + parallelResult);
System.out.println("Parallel time: " + parallelTime + "ms");
System.out.println("Sequential time: " + sequentialTime + "ms");
System.out.println("Results match: " + (parallelResult.equals(sequentialResult)));
}
}
7. Advanced Features
invokeAll() Method
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;
import java.util.Arrays;
public class InvokeAllExample extends RecursiveTask<Long> {
private final int[] array;
private final int start;
private final int end;
private static final int THRESHOLD = 100;
public InvokeAllExample(int[] array, int start, int end) {
this.array = array;
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
int length = end - start;
if (length <= THRESHOLD) {
return Arrays.stream(array, start, end)
.asLongStream()
.sum();
}
int mid = start + length / 2;
InvokeAllExample leftTask = new InvokeAllExample(array, start, mid);
InvokeAllExample rightTask = new InvokeAllExample(array, mid, end);
// Fork both tasks and wait for them - more efficient than fork()/join()
invokeAll(leftTask, rightTask);
return leftTask.join() + rightTask.join();
}
public static void main(String[] args) {
int[] array = new int[1000];
Arrays.fill(array, 1);
ForkJoinPool pool = ForkJoinPool.commonPool();
InvokeAllExample task = new InvokeAllExample(array, 0, array.length);
Long result = pool.invoke(task);
System.out.println("Sum: " + result);
System.out.println("Expected: " + array.length);
}
}
Exception Handling in Fork/Join
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;
public class ExceptionHandlingTask extends RecursiveTask<Integer> {
private final int[] array;
private final int start;
private final int end;
public ExceptionHandlingTask(int[] array, int start, int end) {
this.array = array;
this.start = start;
this.end = end;
}
@Override
protected Integer compute() {
try {
if (end - start <= 1) {
if (array[start] < 0) {
throw new IllegalArgumentException("Negative number found: " + array[start]);
}
return array[start];
}
int mid = start + (end - start) / 2;
ExceptionHandlingTask leftTask = new ExceptionHandlingTask(array, start, mid);
ExceptionHandlingTask rightTask = new ExceptionHandlingTask(array, mid, end);
leftTask.fork();
int rightResult = rightTask.compute();
int leftResult = leftTask.join();
return leftResult + rightResult;
} catch (Exception e) {
// Re-throw as RuntimeException to propagate
throw new RuntimeException("Error in subtask", e);
}
}
public static void main(String[] args) {
int[] array = {1, 2, 3, -4, 5, 6}; // Contains negative number
ForkJoinPool pool = ForkJoinPool.commonPool();
ExceptionHandlingTask task = new ExceptionHandlingTask(array, 0, array.length);
try {
Integer result = pool.invoke(task);
System.out.println("Result: " + result);
} catch (RuntimeException e) {
System.out.println("Caught exception: " + e.getMessage());
System.out.println("Root cause: " + e.getCause().getMessage());
}
}
}
8. Performance Considerations
Threshold Tuning
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;
public class ThresholdTuning extends RecursiveTask<Long> {
private final long[] array;
private final int start;
private final int end;
private final int threshold;
public ThresholdTuning(long[] array, int start, int end, int threshold) {
this.array = array;
this.start = start;
this.end = end;
this.threshold = threshold;
}
@Override
protected Long compute() {
int length = end - start;
if (length <= threshold) {
long sum = 0;
for (int i = start; i < end; i++) {
sum += array[i];
}
return sum;
}
int mid = start + length / 2;
ThresholdTuning leftTask = new ThresholdTuning(array, start, mid, threshold);
ThresholdTuning rightTask = new ThresholdTuning(array, mid, end, threshold);
leftTask.fork();
Long rightResult = rightTask.compute();
Long leftResult = leftTask.join();
return leftResult + rightResult;
}
public static void main(String[] args) {
long[] array = new long[1000000];
for (int i = 0; i < array.length; i++) {
array[i] = i + 1;
}
int[] thresholds = {100, 1000, 10000, 100000};
for (int threshold : thresholds) {
long startTime = System.currentTimeMillis();
ForkJoinPool pool = ForkJoinPool.commonPool();
ThresholdTuning task = new ThresholdTuning(array, 0, array.length, threshold);
Long result = pool.invoke(task);
long duration = System.currentTimeMillis() - startTime;
System.out.printf("Threshold: %7d, Time: %4dms, Result: %d%n",
threshold, duration, result);
}
}
}
9. Real-World Example: File Processing
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;
import java.io.IOException;
import java.nio.file.*;
import java.util.ArrayList;
import java.util.List;
public class FileSearchTask extends RecursiveTask<List<Path>> {
private final Path directory;
private final String fileExtension;
private static final int THRESHOLD = 100; // Max directories to process sequentially
public FileSearchTask(Path directory, String fileExtension) {
this.directory = directory;
this.fileExtension = fileExtension;
}
@Override
protected List<Path> compute() {
List<Path> files = new ArrayList<>();
List<FileSearchTask> subtasks = new ArrayList<>();
try (DirectoryStream<Path> stream = Files.newDirectoryStream(directory)) {
for (Path entry : stream) {
if (Files.isDirectory(entry)) {
// Create subtask for subdirectory
FileSearchTask subtask = new FileSearchTask(entry, fileExtension);
subtask.fork();
subtasks.add(subtask);
} else if (entry.toString().endsWith(fileExtension)) {
files.add(entry);
}
}
} catch (IOException e) {
System.err.println("Error reading directory: " + e.getMessage());
}
// If too many subtasks, process some sequentially
if (subtasks.size() > THRESHOLD) {
// Process first half of subtasks sequentially
for (int i = 0; i < subtasks.size() / 2; i++) {
files.addAll(subtasks.get(i).compute());
}
// Wait for second half
for (int i = subtasks.size() / 2; i < subtasks.size(); i++) {
files.addAll(subtasks.get(i).join());
}
} else {
// Wait for all subtasks
for (FileSearchTask subtask : subtasks) {
files.addAll(subtask.join());
}
}
return files;
}
public static void main(String[] args) {
Path startDir = Paths.get("."); // Current directory
String extension = ".java";
ForkJoinPool pool = ForkJoinPool.commonPool();
FileSearchTask task = new FileSearchTask(startDir, extension);
long startTime = System.currentTimeMillis();
List<Path> foundFiles = pool.invoke(task);
long duration = System.currentTimeMillis() - startTime;
System.out.println("Found " + foundFiles.size() + " " + extension + " files in " + duration + "ms");
foundFiles.forEach(System.out::println);
}
}
10. Best Practices
1. Choose Appropriate Threshold
// Good threshold depends on: // - Task complexity // - Number of available processors // - Overhead of task creation private static final int THRESHOLD = Math.max(array.length / (Runtime.getRuntime().availableProcessors() * 4), 1000);
2. Avoid Blocking Operations
// Bad - blocking I/O in compute method
protected Long compute() {
// Reading from network/file - blocks thread
String data = readFromNetwork(); // AVOID THIS
return process(data);
}
// Good - use CompletableFuture for I/O
protected Long compute() {
CompletableFuture<String> future = CompletableFuture.supplyAsync(this::readFromNetwork);
return process(future.join());
}
3. Use invokeAll for Multiple Subtasks
// More efficient than sequential fork/join
protected Result compute() {
List<Subtask> subtasks = createSubtasks();
// Fork all and then join all
invokeAll(subtasks);
return combineResults(subtasks);
}
4. Handle Exceptions Properly
@Override
protected Result compute() {
try {
// computation
} catch (Exception e) {
// Log and rethrow or return default value
completeExceptionally(e);
return null;
}
}
Summary
The Fork/Join Framework is ideal for:
- Recursive algorithms that can be divided
- CPU-intensive tasks with minimal I/O
- Data parallelism scenarios
- Divide-and-conquer problems
Key Benefits:
- Automatic workload balancing
- Efficient use of multiple cores
- Minimal synchronization overhead
- Elegant recursive problem solving
When to Avoid:
- I/O-bound operations
- Tasks with dependencies
- Very small problem sizes
- Simple sequential operations