Что такое RecursiveTask?

Что такое RecursiveTask

RecursiveTask — это абстрактный класс из Java framework Fork/Join, который используется для решения задач методом "разделяй и властвуй" (divide and conquer) с использованием параллелизма. RecursiveTask предназначена для задач, которые возвращают результат.

Это часть пакета java.util.concurrent и является основой для эффективной параллельной обработки в Java.

Основная идея Fork/Join

Оригинальная задача (размер: 1000000 элементов)
    |
    +---> Fork: разделяем на подзадачи
    |
    +---> Задача 1 (250000 элементов) → обрабатываем параллельно
    |     |
    |     +---> Fork: еще разделяем
    |     |     Задача 1a (125000) | Задача 1b (125000)
    |     |
    |     +---> Join: объединяем результаты
    |
    +---> Задача 2 (250000 элементов) → обрабатываем параллельно
    |
    +---> Задача 3 (250000 элементов) → обрабатываем параллельно
    |
    +---> Задача 4 (250000 элементов) → обрабатываем параллельно
    |
    +---> Join: собираем все результаты
    |
    v
Итоговый результат

Основная структура RecursiveTask

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

// RecursiveTask<V> - V это тип результата
public class MyTask extends RecursiveTask<Integer> {
    
    private int[] array;
    private int start;
    private int end;
    private static final int THRESHOLD = 1000;  // Порог для разделения
    
    public MyTask(int[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }
    
    @Override
    protected Integer compute() {
        // Если размер задачи достаточно маленький - вычисляем напрямую
        if (end - start <= THRESHOLD) {
            return computeDirectly();
        }
        
        // Иначе разделяем (fork) и объединяем (join)
        int mid = (start + end) / 2;
        
        MyTask leftTask = new MyTask(array, start, mid);
        MyTask rightTask = new MyTask(array, mid, end);
        
        leftTask.fork();   // Запускаем левую задачу в отдельном потоке
        int rightResult = rightTask.compute();  // Вычисляем правую задачу в текущем потоке
        int leftResult = leftTask.join();  // Ждем результата левой задачи
        
        return leftResult + rightResult;  // Объединяем результаты
    }
    
    private Integer computeDirectly() {
        int sum = 0;
        for (int i = start; i < end; i++) {
            sum += array[i];
        }
        return sum;
    }
}

// Использование
public class ForkJoinExample {
    public static void main(String[] args) {
        int[] array = new int[1_000_000];
        for (int i = 0; i < array.length; i++) {
            array[i] = i + 1;
        }
        
        // Создаем пул Fork/Join потоков
        ForkJoinPool pool = new ForkJoinPool();
        
        // Создаем и запускаем задачу
        MyTask task = new MyTask(array, 0, array.length);
        Integer result = pool.invoke(task);
        
        System.out.println("Sum: " + result);
        pool.shutdown();
    }
}

Сравнение: RecursiveTask vs RecursiveAction

// RecursiveTask<V> - возвращает результат
public class SumTask extends RecursiveTask<Long> {
    @Override
    protected Long compute() {
        // Вычисляем и возвращаем результат
        return result;
    }
}

// RecursiveAction - не возвращает результат
public class PrintTask extends RecursiveAction {
    @Override
    protected void compute() {
        // Выполняем побочные эффекты (вывод, запись в файл)
    }
}

Реальные примеры

1. Сумма элементов массива

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

public class ArraySumTask extends RecursiveTask<Long> {
    private final long[] numbers;
    private final int start;
    private final int end;
    private static final int THRESHOLD = 1000;
    
    public ArraySumTask(long[] numbers, int start, int end) {
        this.numbers = numbers;
        this.start = start;
        this.end = end;
    }
    
    @Override
    protected Long compute() {
        // Если размер маленький - вычисляем сразу
        if (end - start <= THRESHOLD) {
            long sum = 0;
            for (int i = start; i < end; i++) {
                sum += numbers[i];
            }
            return sum;
        }
        
        // Иначе разделяем
        int mid = (start + end) / 2;
        ArraySumTask leftTask = new ArraySumTask(numbers, start, mid);
        ArraySumTask rightTask = new ArraySumTask(numbers, mid, end);
        
        leftTask.fork();
        Long rightSum = rightTask.compute();
        Long leftSum = leftTask.join();
        
        return leftSum + rightSum;
    }
    
    public static void main(String[] args) {
        long[] numbers = new long[10_000_000];
        for (int i = 0; i < numbers.length; i++) {
            numbers[i] = i + 1;
        }
        
        ForkJoinPool pool = ForkJoinPool.commonPool();
        ArraySumTask task = new ArraySumTask(numbers, 0, numbers.length);
        Long result = pool.invoke(task);
        
        System.out.println("Sum: " + result);
    }
}

2. Поиск максимального значения в массиве

public class MaxValueTask extends RecursiveTask<Integer> {
    private int[] array;
    private int start;
    private int end;
    private static final int THRESHOLD = 5000;
    
    public MaxValueTask(int[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }
    
    @Override
    protected Integer compute() {
        if (end - start <= THRESHOLD) {
            // Базовый случай: вычисляем максимум
            int max = Integer.MIN_VALUE;
            for (int i = start; i < end; i++) {
                max = Math.max(max, array[i]);
            }
            return max;
        }
        
        // Рекурсивный случай
        int mid = (start + end) / 2;
        MaxValueTask leftTask = new MaxValueTask(array, start, mid);
        MaxValueTask rightTask = new MaxValueTask(array, mid, end);
        
        leftTask.fork();
        int rightMax = rightTask.compute();
        int leftMax = leftTask.join();
        
        return Math.max(leftMax, rightMax);
    }
    
    public static void main(String[] args) {
        int[] array = {3, 7, 2, 9, 1, 5, 8, 4, 6};
        
        ForkJoinPool pool = ForkJoinPool.commonPool();
        MaxValueTask task = new MaxValueTask(array, 0, array.length);
        Integer maxValue = pool.invoke(task);
        
        System.out.println("Max value: " + maxValue);
    }
}

3. Сортировка массива (Merge Sort)

public class MergeSortTask extends RecursiveTask<int[]> {
    private int[] array;
    private int start;
    private int end;
    private static final int THRESHOLD = 1000;
    
    public MergeSortTask(int[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }
    
    @Override
    protected int[] compute() {
        // Если размер маленький - используем обычную сортировку
        if (end - start <= THRESHOLD) {
            int[] temp = new int[end - start];
            System.arraycopy(array, start, temp, 0, end - start);
            java.util.Arrays.sort(temp);
            return temp;
        }
        
        // Разделяем
        int mid = (start + end) / 2;
        MergeSortTask leftTask = new MergeSortTask(array, start, mid);
        MergeSortTask rightTask = new MergeSortTask(array, mid, end);
        
        leftTask.fork();
        int[] rightSorted = rightTask.compute();
        int[] leftSorted = leftTask.join();
        
        // Объединяем отсортированные части
        return merge(leftSorted, rightSorted);
    }
    
    private int[] merge(int[] left, int[] right) {
        int[] result = new int[left.length + right.length];
        int i = 0, j = 0, k = 0;
        
        while (i < left.length && j < right.length) {
            result[k++] = (left[i] <= right[j]) ? left[i++] : right[j++];
        }
        
        while (i < left.length) {
            result[k++] = left[i++];
        }
        
        while (j < right.length) {
            result[k++] = right[j++];
        }
        
        return result;
    }
}

4. Обход дерева и подсчет узлов

public class TreeNode {
    public int value;
    public TreeNode left;
    public TreeNode right;
    
    public TreeNode(int value) {
        this.value = value;
    }
}

public class TreeCountTask extends RecursiveTask<Integer> {
    private TreeNode node;
    
    public TreeCountTask(TreeNode node) {
        this.node = node;
    }
    
    @Override
    protected Integer compute() {
        // Базовый случай: листовой узел
        if (node == null) {
            return 0;
        }
        
        if (node.left == null && node.right == null) {
            return 1;
        }
        
        // Рекурсивно обрабатываем поддеревья
        TreeCountTask leftTask = null;
        TreeCountTask rightTask = null;
        
        if (node.left != null) {
            leftTask = new TreeCountTask(node.left);
            leftTask.fork();
        }
        
        if (node.right != null) {
            rightTask = new TreeCountTask(node.right);
            rightTask.fork();
        }
        
        int leftCount = (leftTask == null) ? 0 : leftTask.join();
        int rightCount = (rightTask == null) ? 0 : rightTask.join();
        
        return 1 + leftCount + rightCount;
    }
}

Ключевые методы RecursiveTask

public abstract class RecursiveTask<V> extends ForkJoinTask<V> {
    
    // Основной метод, который переопределяем
    protected abstract V compute();
    
    // Запускает задачу асинхронно в отдельном потоке
    public final ForkJoinTask<V> fork()
    
    // Ждет завершения и возвращает результат
    public final V join()
    
    // Запускает и сразу ждет (синхронно)
    public final V invoke()
}

ForkJoinPool

public class ForkJoinPoolExample {
    public static void main(String[] args) {
        // Вариант 1: Используем общий пул (ForkJoinPool.commonPool())
        ForkJoinPool commonPool = ForkJoinPool.commonPool();
        MyTask task1 = new MyTask();
        commonPool.invoke(task1);
        
        // Вариант 2: Создаем кастомный пул с определенным количеством потоков
        ForkJoinPool customPool = new ForkJoinPool(
            Runtime.getRuntime().availableProcessors()  // количество потоков
        );
        MyTask task2 = new MyTask();
        customPool.invoke(task2);
        customPool.shutdown();
    }
}

Оптимизация: сбалансированное разделение

public class OptimizedTask extends RecursiveTask<Long> {
    private long[] data;
    private int start;
    private int end;
    
    // Динамически выбираем порог в зависимости от размера
    private static int computeThreshold() {
        int cpus = Runtime.getRuntime().availableProcessors();
        return Math.max(1000, 100_000 / cpus);  // Примерно 100K на процессор
    }
    
    @Override
    protected Long compute() {
        int threshold = computeThreshold();
        
        if (end - start <= threshold) {
            long sum = 0;
            for (int i = start; i < end; i++) {
                sum += data[i];
            }
            return sum;
        }
        
        // Разделяем пополам
        int mid = (start + end) / 2;
        OptimizedTask left = new OptimizedTask(data, start, mid);
        OptimizedTask right = new OptimizedTask(data, mid, end);
        
        left.fork();
        long rightResult = right.compute();
        long leftResult = left.join();
        
        return leftResult + rightResult;
    }
}

Когда использовать RecursiveTask

✓ Используй RecursiveTask когда:

• Задача имеет структуру "разделяй и властвуй"
• Разделение на подзадачи создает достаточно работы
• Задача возвращает результат
• Нужна высокая производительность на многопроцессорных системах

✗ НЕ используй RecursiveTask когда:

• Задача очень маленькая (overhead Fork/Join больше выигрыша)
• Нет явного разделения на подзадачи
• Нужна максимальная простота кода

Заключение

RecursiveTask — это мощный инструмент для параллельной обработки больших объемов данных с использованием стратегии "разделяй и властвуй". Fork/Join framework автоматически управляет потоками и балансирует нагрузку между ними, что делает код и параллелизм эффективными.