ForkJoinPool və RecursiveTask: rekursiv tapşırıqlar

1. ForkJoinPool: bu nədir və nə üçün lazımdır

ForkJoinPool — “böl və hökm sür” (divide and conquer) yanaşmasını reallaşdıran xüsusi axın hovuzudur. Onun vəzifəsi — böyük tapşırığı müstəqil subtapşırıqlara bölmək, onları paralel icra etmək və sonra nəticələri birləşdirmək mümkün olduqda işi maksimum səmərəli şəkildə paralelləşdirməkdir.

• Fork (bölmək) — tapşırıq subtapşırıqlara bölünür.
• Join (birləşdirmək) — subtapşırıqların nəticələri yekunlaşdırılır.

ForkJoinPool — Java-da paralel stream-lərin ürəyidir: siz list.parallelStream() yazanda, məhz o istifadə olunur. Lakin siz onu birbaşa da tətbiq edə bilərsiniz və daha çox nəzarət əldə edirsiniz.

ForkJoinPool nə zaman xüsusilə faydalıdır

ForkJoinPool asanlıqla müstəqil hissələrə bölünən tapşırıqlarda özünü ən yaxşı göstərir: məsələn, çox böyük massivlərin emalı — hər hissə ayrı-ayrılıqda işlənir, sonra nəticələr birləşdirilir.

• Tapşırığı asanlıqla müstəqil subtapşırıqlara bölmək olar: çeşidləmə, axtarış, cəmləmə.
• Subtapşırıqlar təxminən eyni həcmdədir və bir-birindən asılı deyil.
• Maksimal sürət üçün prosessorun bütün nüvələrindən istifadə etmək lazımdır.

+---------------------+
|   Böyük tapşırıq    |
+---------------------+
          |
          v
+---------+---------+
|  Subtapşırıq 1    |
|  Subtapşırıq 2    |
|  ...              |
+-------------------+
          |
          v
+---------+---------+
|  Nəticələr        |
+-------------------+

Məhz belə “böl və hökm sür” işləyir: böldük — paralel hesabladıq — birləşdirdik.

2. RecursiveTask və RecursiveAction: eyni medalın iki üzü

ForkJoinPool-da tapşırıqlar subtapşırıqlara bölünməyi və nəticələri birləşdirməyi bacaran xüsusi siniflərlə təsvir olunur. RecursiveTask<T> nəticə qaytarır, RecursiveAction — qaytarmır. Praktikada çox vaxt RecursiveTask istifadə olunur; məsələn, cəmi, maksimumu və ya sayını qaytarmaq üçün.

Belə tapşırıq yaratmaq üçün ondan miras alıb compute() metodunu reallaşdırırıq. Orada məntiqi təsvir edirik: tapşırıq kiçikdirsə — dərhal həll edirik; böyükdürsə — subtapşırıqlara bölür, onları fork() ilə paralel işə salır və join() vasitəsilə nəticələri birləşdiririk. Beləcə, təbii rekursiv paralelləşmə yaranır.

3. Sintaksis və nümunə: massiv cəminin paralel hesablanması

Gəlin böyük bir ədəd massivi olsun və bütün elementlərin cəmini tez hesablamaq istəyək.

Addım 1. Tapşırıq sinfi

import java.util.concurrent.RecursiveTask;

public class ArraySumTask extends RecursiveTask<Long> {
    private static final int THRESHOLD = 1_000; // Tapşırığın bölünmə həddi
    private final int[] array;
    private final int start, end;

    public ArraySumTask(int[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        // Tapşırıq kiçikdirsə — birbaşa hesablayırıq
        if (end - start <= THRESHOLD) {
            long sum = 0;
            for (int i = start; i < end; i++) {
                sum += array[i];
            }
            return sum;
        } else {
            // Tapşırığı iki subtapşırığa bölürük
            int mid = (start + end) / 2;
            ArraySumTask leftTask = new ArraySumTask(array, start, mid);
            ArraySumTask rightTask = new ArraySumTask(array, mid, end);

            // Subtapşırıqları paralel başladırıq
            leftTask.fork(); // Asinxron
            long rightResult = rightTask.compute(); // Sinxron
            long leftResult = leftTask.join(); // Solun bitməsini gözləyirik

            // Nəticəni birləşdiririk
            return leftResult + rightResult;
        }
    }
}

• Tapşırıq kiçikdirsə (hədd THRESHOLD-dan kiçikdirsə) — cəmi adi dövrlə hesablayırıq.
• Böyükdürsə — ikiyə bölürük, birini fork() ilə asinxron işə salırıq, digərini compute() ilə sinxron hesablayırıq, sonra join() ilə birləşdiririk.

Addım 2. Tapşırığın ForkJoinPool vasitəsilə işə salınması

import java.util.concurrent.ForkJoinPool;

public class ForkJoinSumDemo {
    public static void main(String[] args) {
        int[] numbers = new int[10_000_000];
        for (int i = 0; i < numbers.length; i++) {
            numbers[i] = 1; // Sadəlik üçün — cəm massiv uzunluğuna bərabər olmalıdır
        }

        ForkJoinPool pool = new ForkJoinPool(); // Defolt olaraq — nüvələrin sayına görə

        ArraySumTask task = new ArraySumTask(numbers, 0, numbers.length);

        long result = pool.invoke(task); // Tapşırığın işə salınması

        System.out.println("Massivin elementlərinin cəmi: " + result);
    }
}

Bu necə işləyir?

• ForkJoinPool hansı sayda axından istifadə etməyi özü müəyyən edir (adətən — nüvələrin sayına görə).
• Tapşırıq avtomatik olaraq subtapşırıqlara bölünür, hər biri ayrıca nüvədə icra oluna bilər.
• Məhsuldarlıq adətən ardıcıl koddakından yüksək olur (xüsusilə böyük verilənlər və çoxnüvəli sistemlərdə).

4. ForkJoinPool necə işləyir: bir az “qapağın altında”

Work-Stealing (işin “oğurlanması”)

ForkJoinPool “work‑stealing” mexanizmini reallaşdırır: hər hansı axında tapşırıq qurtaranda, o, başqa axından “iş” oğurlayır. Bu, yüklərin effektiv balanslaşdırılmasını və bütün nüvələrin səmərəli istifadəsini təmin edir.

Baza alqoritm

• Əsas tapşırıq subtapşırıqlara bölünür.
• Subtapşırıqlar ixtisaslaşmış növbələrə yerləşdirilir.
• Axınlar tapşırıqları öz növbələrindən götürür, onlar boşalanda — “qonşulardan” iş axtarırlar.
• Hər şey yerinə yetirildikdə, nəticələr birləşdirilir.

İş sxemi

flowchart TD
    A[Əsas tapşırıq] --> B1[Subtapşırıq 1]
    A --> B2[Subtapşırıq 2]
    B1 --> C1[Kiçik tapşırıq 1]
    B1 --> C2[Kiçik tapşırıq 2]
    B2 --> C3[Kiçik tapşırıq 3]
    B2 --> C4[Kiçik tapşırıq 4]
    C1 --> D[Nəticələrin birləşdirilməsi]
    C2 --> D
    C3 --> D
    C4 --> D

5. RecursiveAction — nəticə qaytarmağa ehtiyac yoxdursa

Əgər sadəcə nələri isə paralel icra etmək və nəticə qaytarmamaq lazımdırsa, RecursiveAction istifadə edin. Tipik nümunələr — massivi paralel doldurmaq, çap, “yerində” çeşidləmə və s.

import java.util.concurrent.RecursiveAction;

public class PrintTask extends RecursiveAction {
    private static final int THRESHOLD = 100;
    private final int[] array;
    private final int start, end;

    public PrintTask(int[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    protected void compute() {
        if (end - start <= THRESHOLD) {
            for (int i = start; i < end; i++) {
                System.out.print(array[i] + " ");
            }
        } else {
            int mid = (start + end) / 2;
            invokeAll(
                new PrintTask(array, start, mid),
                new PrintTask(array, mid, end)
            );
        }
    }
}

6. Təcrübə: massivdə maksimum dəyərin paralel axtarışı

import java.util.concurrent.RecursiveTask;

public class MaxFindTask extends RecursiveTask<Integer> {
    private static final int THRESHOLD = 1000;
    private final int[] array;
    private final int start, end;

    public MaxFindTask(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 = array[start];
            for (int i = start + 1; i < end; i++) {
                if (array[i] > max) max = array[i];
            }
            return max;
        } else {
            int mid = (start + end) / 2;
            MaxFindTask left = new MaxFindTask(array, start, mid);
            MaxFindTask right = new MaxFindTask(array, mid, end);
            left.fork();
            int rightResult = right.compute();
            int leftResult = left.join();
            return Math.max(leftResult, rightResult);
        }
    }
}

İşə salma:

import java.util.concurrent.ForkJoinPool;

public class ForkJoinMaxDemo {
    public static void main(String[] args) {
        int[] array = new int[5_000_000];
        for (int i = 0; i < array.length; i++) {
            array[i] = (int)(Math.random() * 1_000_000);
        }

        ForkJoinPool pool = new ForkJoinPool();
        MaxFindTask task = new MaxFindTask(array, 0, array.length);

        int max = pool.invoke(task);

        System.out.println("Maksimum dəyər: " + max);
    }
}

7. ForkJoinPool-un üstünlükləri və məhdudiyyətləri

Üstünlüklər

• Yükün avtomatik balanslaşdırılması. Work-stealing bütün nüvələri səmərəli istifadə etməyə imkan verir.
• Rahatlıq. Axınları əl ilə yaratmaq və idarə etmək lazım deyil.
• Yüksək məhsuldarlıq. Xüsusilə böyük tapşırıqlarda və çoxnüvəli sistemlərdə.
• Çeviklik. Tapşırıqları lazım olduğu qədər hissəyə bölmək olar.

Məhdudiyyətlər

• Subtapşırıqlar arasında güclü bağlılıq. Əgər subtapşırıqlar tez-tez bir-birini gözləyirsə, üstünlük azalır.
• Çox xırda tapşırıqlar. Bölmə/sinxronizasiya üzrə əlavə xərclər üstünlüyü “yeyə” bilər.
• Yan təsirlər. Ümumi dəyişənləri sinxronizasiya etmədən dəyişmək olmaz — race condition əldə edərsiniz.
• Tətbiqolunma. Müstəqil hissələrə bölünən tapşırıqlar üçün uyğundur.

8. ForkJoinPool və RecursiveTask ilə işləyərkən tipik səhvlər

Səhv №1: Tapşırığın çox xırda bölünməsi. Əgər hədd (THRESHOLD) çox kiçikdirsə, çoxlu xırda tapşırıqlar yaranacaq — onları yaratmaq və sinxronizasiya etmək xərcləri paralelləşmənin qazancından çox olacaq. Hədlə eksperimenti edin: optimal dəyərlər çox vaxt minlərlə və ya on minlərlə element səviyyəsində olur.

Səhv №2: Ümumi dəyişənlərin istifadəsi. Əgər subtapşırıqlar sinxronizasiya olmadan ümumi dəyişənə yazırsa — məlumat yarışları (race condition) yaranacaq. Nəticəni compute() vasitəsilə qaytarın və yalnız join()-da birləşdirin.

Səhv №3: fork/join-un yanlış istifadəsi. fork() və ya join() çağırmağı yaddan çıxartdınızsa — subtapşırıq paralel başlamayacaq və ya nəticə “itiriləcək”. Çağırışların ardıcıllığına diqqətlə nəzarət edin.

Səhv №4: ForkJoinTask-ı ForkJoinPool-dan kənarda işə salmaq. Əgər sadəcə tapşırığın compute()-unu çağırsanız, o, cari axında, paralelləşmə olmadan icra olunacaq. Həqiqi effekt üçün pool.invoke() və ya pool.submit() istifadə edin.

Səhv №5: İstisnaları nəzərə almamaq. Əgər tapşırıqda istisna baş verərsə, o, join() və ya invoke() zamanı üzə çıxacaq. Səhvləri emal etməyi unutmayın.

Səhv №6: Bloklayan tapşırıqlar üçün ForkJoinPool-dan istifadə. ForkJoinPool tez-tez bloklanan (I/O gözləyən və s.) tapşırıqlar üçün yaxşı uyğun deyil. Belə hallarda ExecutorService istifadə etmək daha məqsədəuyğundur.