Fork/Join framework в Java

Last updated: 7 мая 2025 г.

fork/join framework – для разбиения задачи на подзадачи и чтобы эти подзадачи выполнялись в отдельных потоках.

Также эти подзадачи могут тоже делиться на подзадачи и опять же чтобы эти подзадачи выполнялись в отдельных потоках.

Повторяться этот процесс деления может пока подзадача не станет необходимо мала.

Есть пул потоков. В один из его потоков ставиться большая задача.

Эта задача и подзадачи в будущем являют собой выполнения метода compile, который нужно переопределить

Переопределяется он всегда похожим образом:

if задача или подзадача уже необходимо мала, возвращаем что-то из метода compile.

else – делим задачу или подзадачу на подзадачи.

Деление на подзадачи в else происходит особым рекурсивным образом с помощью методов fork и join.

fork добавляет подзадачу в пул потоков и любой свободный поток может ее подхватить для выполнения. fork вызывает метод compile подзадачи, к которой fork был применен.

join ждет пока эта форкнутая подзадача выполнится.

Внимание джойном останавливается выполнение кода, но поток не блокируется, поток может выполнять другую подзадачу из пула потоков.

join возвращает значение из compile, который fork вызвал.

Ясное дело, compile будет вызываться рекурсивно форком и соответственно джойны будут копиться пока не дойдет до наименьшей подзадачи.

И джойны, понятное дело, в обратном направлении начнут освобождаться.

Помещенные задачи форком в любом потоке в пул потоков могут быть взяты любым свободным потоком для выполнения.

В этом и есть профит, то что задача распараллеливается и любой поток может взять любую подзадачу для выполнения, даже если она была сгенерирована другим потоком.

Пример программы:

1import java.util.concurrent.ExecutionException;
2import java.util.concurrent.ForkJoinPool;
3import java.util.concurrent.RecursiveTask;
4
5public class ForkJoinExample {
6    static volatile int j;
7    public static void main(final String[] arguments)
8        throws InterruptedException,
9        ExecutionException {
10        //узнать количество ядер компьютера
11        //(в каждом ядре эффективно выполнять 1 поток)
12        int nThreads = Runtime.getRuntime().availableProcessors();
13        System.out.println(nThreads);
14
15        int[] numbers = new int[1000];
16        //запишем в массив числа от 1 до 1000
17        for(int i = 0; i < numbers.length; i++) {
18            numbers[i] = i+1;
19        }
20
21        //Будем складывать числа от 1 до 1000
22        //эффективным образом с помощью fork/join
23
24        // сюда передаем количество
25        // ядер (размер пула потоков)
26        ForkJoinPool forkJoinPool = new ForkJoinPool(nThreads);
27        // invoke запускает задачу на выполнение
28        // то есть вызывает compile.
29        Long result = forkJoinPool.invoke(
30            new Sum(numbers,0,numbers.length));
31        //пока invoke не вернул значение дальше
32        //в методе мейн ничего не выполняется
33        System.out.println(result);
34    }
35
36    static class Sum extends RecursiveTask< Long > {
37        int low;
38        int high;
39        int[] array;
40
41        Sum(int[] array, int low, int high) {
42            this.array = array;
43            this.low = low;
44            this.high = high;
45        }
46
47        protected Long compute() {
48            //if задача или подзадача уже необходимо мала
49            //возвращаем что-то из метода compile
50            if(high – low <= 10) {
51                long sum = 0;
52
53                for(int i = low; i < high; ++i)
54                    sum += array[i];
55                return sum;
56            }
57            //else - делим задачу или подзадачу
58            //на подзадачи (в данном случае на 2 половины)
59            else {
60                int mid = low + (high - low) / 2;
61                Sum left = new Sum(array, low, mid);
62                Sum right = new Sum(array, mid, high);
63                // добавляем одну половину задачи (подзадачу)
64                // в пул потоков
65                left.fork();
66                // добавляем вторую половину задачи (подзадачу)
67                // в пул потоков
68                right.fork();
69                // ждем когда compile в форке завершиться
70                long leftResult = left.join();
71                // ждем когда compile в форке завершиться
72                long rightResult = right.join();
73                return leftResult + rightResult;
74            }
75        }
76    }
77}