ExecutorService, Callable, Future: tapşırıqların işə salınması

1. ExecutorService: axınları peşəkar kimi idarə edək

Niyə sadəcə new Thread ilə axınlar yaratmaq məsləhət deyil

İlk baxışda çoxaxınlılıq sadə görünür:

Thread t = new Thread(() -> {
    // nəsə edirik
});
t.start();

Bu yanaşma işləyir, amma tapşırıqlar çoxalanda tez bir zamanda yükə çevrilir. Hər new Thread() çağırışı yeni axın yaradır, onlarla və yüzlərlə axın isə sistemi yükləməyə başlayır. Üstəlik onları idarə etmək narahatdır: nə vaxt bitdiklərini izləmək, səhv olduqda nə etmək, necə dayandırmaq və yenidən istifadə etmək lazımdır.

Məhz burada səhnəyə ExecutorService çıxır — ağıllı axın dispetcheri. Sadəcə tapşırıqları ona verirsiniz, o isə hansı axınla və nə vaxt icra olunacağını özü qərar verir. Nəticədə hər şey daha sürətli, stabil və baş ağrısı olmadan işləyir.

ExecutorService necə işləyir

ExecutorService sadə, lakin effektiv prinsiplə işləyir.

• Onun daxilində axın hovuzu var — əvvəlcədən yaradılmış işçi axınların dəsti (sabit və ya dinamik).
• Tapşırıqlar növbəyə düşür və boş axınlar tərəfindən götürülür.
• Servis həyat dövrünü idarə edir: siz tamamlanmanı gözləyə, hovuzu səliqə ilə dayandıra və resursları azad edə bilirsiniz.

ExecutorService-in yaradılması

Ən çox istifadə olunan yol — Executors sinfinin fabrik metodlarından yararlanmaqdır:

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

ExecutorService executor = Executors.newFixedThreadPool(4); // 4 axın

• newFixedThreadPool(N) — N axından ibarət hovuz (əksər tapşırıqlar üçün uyğundur).
• newCachedThreadPool() — dinamik hovuz, tələbat olduqda axın yaradır (diqqət: tapşırıqlar seli zamanı yaddaşa “dirənə” bilərsiniz).
• newSingleThreadExecutor() — bir axın (ardıcıl icra).

Nümunə: Runnable tapşırığını ExecutorService ilə işə salmaq

executor.submit(() -> {
    System.out.println("Axın hovuzundan salam!");
});

ExecutorService ilə işiniz bitəndən sonra onu mütləq səliqə ilə tamamlayın:

executor.shutdown(); // Yeni tapşırıqların əlavə olunmasını qadağan edir, cari tapşırıqların bitməsini gözləyir

Vacibdir: Əgər shutdown() çağırmasanız, proqram yekunlaşmaya bilər — hovuzdakı axınlar yeni tapşırıqları gözləyəcəklər.

2. Runnable vs Callable: tapşırıqlar müxtəlifdir

Java 5-ə qədər, əgər siz axında nəyisə icra etmək istəyirdinizsə, Runnable interfeysinin reallaşdırılmasını yazırdınız. Bu, heç nə qaytarmayan və yoxlanılan istisnalar atmayan tapşırıqdır.

Runnable task = () -> {
    System.out.println("Sadəcə işləyirəm, heç nə qaytarmıram!");
};
executor.submit(task);

Callable: nəticə ilə (və istisnalarla) tapşırıq

Bəzən istəyirsiniz ki, tapşırıq sadəcə “nəsə etsin” deyə yox, həm də nəticə qaytarsın — məsələn, ədədlərin cəmi, hesablamanın nəticəsi, serverdən məlumatlar və s. Bunun üçün Callable<T> interfeysi yaradılıb.

import java.util.concurrent.Callable;

Callable<Integer> sumTask = () -> {
    int sum = 0;
    for (int i = 1; i <= 100; i++) sum += i;
    return sum;
};

• call() metodu T tipində nəticə qaytarır.
• call() metodu yoxlanılan istisna ata bilər.

Bənzətmə: Runnable — “get, qabları yu” (nəticə vacib deyil), Callable — “get, çay gətir və temperaturunun neçə olduğunu de” (nəticə vacibdir).

Callable-in işə salınması: nəticə almaq üçün executor.submit(...) istifadə edin. O, Future<T> obyektini qaytaracaq.

3. Future: nəticə üçün vəd

Future — gələcəkdə nəticə qaytarmağa “vəd”dir. Tapşırığı ExecutorService-ə göndərdiyiniz zaman Future alırsınız və daha sonra ondan nəticəni götürə, tapşırığın tamamlanıb-tamamlanmadığını öyrənə və ya onu ləğv edə bilərsiniz.

Future-un əsas metodları

• T get() — nəticəni əldə etmək (tapşırıq tamamlanana qədər gözləyir).
• boolean isDone() — tapşırıq tamamlanıbmı.
• boolean cancel(boolean mayInterruptIfRunning) — tapşırığı ləğv etməyə cəhd.
• boolean isCancelled() — tapşırıq ləğv olunubmu.

Nümunə: Callable-in işə salınması və nəticənin alınması

import java.util.concurrent.*;

public class ParallelSumApp {
    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(2);

        Callable<Integer> sumTask = () -> {
            int sum = 0;
            for (int i = 1; i <= 100; i++) sum += i;
            return sum;
        };

        Future<Integer> future = executor.submit(sumTask);

        System.out.println("Tapşırıq işə salındı, başqa işlər görmək olar...");

        // Nəticəni alırıq (tapşırıq hələ tamamlanmayıbsa, metod axını bloklayır)
        Integer result = future.get();
        System.out.println("Hesablamaların nəticəsi: " + result);

        executor.shutdown();
    }
}

• Tapşırıq axın hovuzuna göndərilir.
• Tapşırıq icra olunarkən, əsas axın başqa işlər görə bilər.
• Nəticə lazım olanda future.get() çağırırıq — tapşırıq hələ işləyirsə, axın gözləyəcək.
• Tapşırıq bitən kimi, nəticə qaytarılacaq.

4. Təcrübə: bir neçə tapşırıq, tamamlanmanı gözləmək

Tez-tez bir neçə tapşırığı eyni vaxtda işə salıb, onların hamısının tamamlanmasını gözləmək lazım olur. Məsələn, bir massivdəki məlumatları hissələrə bölür və hər hissənin cəmini ayrı tapşırıqda hesablayırsınız.

Nümunə: massiv elementlərinin hissələrlə cəmi

import java.util.*;
import java.util.concurrent.*;

public class ParallelArraySum {
    public static void main(String[] args) throws Exception {
        int[] array = new int[1000];
        Arrays.setAll(array, i -> i + 1); // 1-dən 1000-ə qədər ədədlərlə doldururuq

        ExecutorService executor = Executors.newFixedThreadPool(4);

        int chunkSize = array.length / 4;
        List<Future<Integer>> futures = new ArrayList<>();

        for (int i = 0; i < 4; i++) {
            int from = i * chunkSize;
            int to = (i == 3) ? array.length : (i + 1) * chunkSize;

            Callable<Integer> sumTask = () -> {
                int sum = 0;
                for (int j = from; j < to; j++) sum += array[j];
                System.out.println("Cəm " + from + " ilə " + (to - 1) + " arasında = " + sum);
                return sum;
            };

            futures.add(executor.submit(sumTask));
        }

        int totalSum = 0;
        for (Future<Integer> f : futures) {
            totalSum += f.get(); // Hər tapşırığı ardıcıl gözləyirik
        }

        System.out.println("Ümumi cəm: " + totalSum);

        executor.shutdown();
    }
}

Burada massiv 4 hissəyə bölünür. Hər hissə üçün Callable tapşırığı yaradılır və cəm hesablanır. Bütün tapşırıqlar ExecutorService-ə göndərilir, cavab olaraq Future qaytarılır. Sonda bütün tapşırıqların nəticələrini toplayırıq.

Real tapşırıqlarda hamısını birdən gözləmək üçün invokeAll-dan istifadə etmək rahatdır.

5. Future ilə işləyərkən səhvlərin emalı

future.get() çağırdığınızda, tapşırıq istisna ilə bitibsə, bu istisna ExecutionException kimi atılacaq. Bu vacibdir: tapşırıqda nəsə qaydasında getməyibsə, bunu yalnız get() çağırışı zamanı öyrənəcəksiniz.

Nümunə: istisnaların emalı

Callable<Integer> errorTask = () -> {
    throw new IllegalArgumentException("Nə isə qaydasında getmədi!");
};

Future<Integer> badFuture = executor.submit(errorTask);

try {
    badFuture.get();
} catch (ExecutionException e) {
    System.out.println("Tapşırıq səhvlə başa çatdı: " + e.getCause());
}

• Tapşırıq daxilində istisna atılır.
• get() çağırışında o, ExecutionException içinə “bükülür”.
• Həqiqi səbəbi getCause() vasitəsilə əldə etmək olar.

6. Faydalı nüanslar

Tapşırığı necə ləğv etmək

Future<?> f = executor.submit(() -> {
    while (true) {
        // Sonsuz iş
        if (Thread.currentThread().isInterrupted()) {
            System.out.println("Məni dayandırmağı xahiş etdilər!");
            break;
        }
    }
});

Thread.sleep(100); // Bir az gözləyək
f.cancel(true); // Tapşırığı ləğv etməyə cəhd edək

• cancel(true) tapşırıq hələ tamamlanmayıbsa, onu dayandırmağa cəhd edir.
• Tapşırıq daxilində Thread.currentThread().isInterrupted() yoxlanmalı və səliqə ilə tamamlanmalıdır.

shutdown və shutdownNow

shutdown() — yumşaq dayandırma: yeni tapşırıqların əlavə olunmasını qadağan edir və cari tapşırıqların sakitcə bitməsinə imkan verir. Ən çox istifadə edilən üsuldur.

shutdownNow() — sərt dayandırma: aktiv axınları dayandırmağa cəhd edir və hələ işə başlamamış tapşırıqların siyahısını qaytarır. Ehtiyatla tətbiq edin.

invokeAll və invokeAny

invokeAll(Collection<Callable<T>> tasks) ötürülən bütün tapşırıqları işə salır və onların hamısının tamamlanmasını gözləyir. Future siyahısı qaytarır.

invokeAny(Collection<Callable<T>> tasks) yalnız ilk uğurla tamamlanan tapşırığı gözləyir, onun nəticəsini qaytarır və qalanlarını ləğv edir. İlk uğurlu cavab vacib olanda uyğundur.

7. ExecutorService, Callable və Future ilə işləyərkən tipik səhvlər

Xəta №1: ExecutorService-i bağlamamaq. Əgər shutdown() çağırmağı unutsaq, proqram main bitdikdən sonra “asma” vəziyyətində qala bilər, çünki hovuzdakı axınlar yeni tapşırıqları gözləyirlər.

Xəta №2: Tapşırığı göndərən kimi nəticəni gözləmək. submit()-dən dərhal sonra get() çağırsaq, asinxronluğun üstünlüklərini itirəcəyik — axın hər halda gözləyəcək. Paralel faydalı iş görün və nəticəni həqiqətən lazım olanda istəyin.

Xəta №3: Tapşırıqlardakı istisnaları görməzdən gəlmək. ExecutionException emal olunmazsa, get() çağırışında tapşırıqda baş verən vacib səhvləri qaçıra bilərsiniz.

Xəta №4: Ortaq dəyişənlərin sinxronizasiya olmadan istifadəsi. Bir neçə tapşırıq eyni verilənlərlə işləyirsə — sinxronizasiya və ya axın-təhlükəsiz kolleksiyalar lazımdır.

Xəta №5: Həddindən artıq çox axın yaratmaq. Axın sayını prosessor nüvələrinin sayını xeyli üstələyən hovuz yaratmağa dəyməz — bu, icranı hətta ləngidə də bilər.

Xəta №6: Tapşırıqları ləğv etməyi unutmaq. Tapşırıq artıq lazım deyilsə, resurs sərf etməmək üçün onu cancel() vasitəsilə ləğv edin.