ExecutorService, Callable, Future: uruchamianie zadań

1. ExecutorService: zarządzamy wątkami jak profesjonaliści

Dlaczego nie warto po prostu tworzyć wątków przez new Thread

Na początku pracy z wielowątkowością wszystko wydaje się proste:

Thread t = new Thread(() -> {
    // coś robimy
});
t.start();

Taki sposób działa, ale szybko staje się uciążliwy, gdy zadań robi się dużo. Każde wywołanie new Thread() tworzy nowy wątek, a dziesiątki czy setki wątków zaczynają przeciążać system. Do tego dochodzi niewygodne zarządzanie: trzeba pilnować, kiedy się kończą, co robić przy błędach, jak je zatrzymywać i ponownie wykorzystywać.

Właśnie tutaj na scenę wchodzi ExecutorService — sprytny dyspozytor wątków. Po prostu przekazujesz mu zadania, a on sam decyduje, którym wątkiem i kiedy je wykonać. W efekcie wszystko działa szybciej, stabilniej i bez bólu głowy.

Jak działa ExecutorService

ExecutorService działa według prostego, lecz skutecznego mechanizmu.

• W środku ma pulę wątków — z góry utworzony zestaw wątków roboczych (stały lub dynamiczny).
• Zadania trafiają do kolejki i są podejmowane przez wolne wątki.
• Serwis zarządza cyklem życia: możesz poczekać na zakończenie, poprawnie zatrzymać pulę i zwolnić zasoby.

Tworzenie ExecutorService

Najczęstszy sposób — użyć metod fabrycznych z klasy Executors:

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

ExecutorService executor = Executors.newFixedThreadPool(4); // 4 wątki

• newFixedThreadPool(N) — pula z N wątków (odpowiednia dla większości zadań).
• newCachedThreadPool() — dynamiczna pula, tworzy wątki w razie potrzeby (uwaga: przy lawinie zadań można wyczerpać pamięć).
• newSingleThreadExecutor() — jeden wątek (sekwencyjne wykonywanie).

Przykład: uruchomienie Runnable przez ExecutorService

executor.submit(() -> {
    System.out.println("Pozdrowienia z puli wątków!");
});

Gdy skończysz pracę z ExecutorService, należy go poprawnie zakończyć:

executor.shutdown(); // Zabrania dodawania nowych zadań, czeka na zakończenie bieżących

Ważne: Jeśli nie wywołasz shutdown(), program może się nie zakończyć — wątki z puli będą czekać na nowe zadania.

2. Runnable vs Callable: zadania są różne

Przed Java 5, jeśli chciałeś coś wykonać w wątku, pisałeś implementację interfejsu Runnable. To zadanie, które nic nie zwraca i nie wyrzuca wyjątków sprawdzanych.

Runnable task = () -> {
    System.out.println("Po prostu pracuję, nic nie zwracam!");
};
executor.submit(task);

Callable: zadanie z wynikiem (i wyjątkami)

Czasem chcemy, aby zadanie nie tylko „coś robiło”, ale też zwracało wynik — na przykład sumę liczb, rezultat obliczeń, dane z serwera. Do tego powstał interfejs Callable<T>.

import java.util.concurrent.Callable;

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

• Metoda call() zwraca wynik typu T.
• Metoda call() może wyrzucać wyjątek sprawdzany.

Analogia: Runnable — „idź i pozmywaj naczynia” (wynik nie jest ważny), Callable — „idź, przynieś herbatę i powiedz, jaką ma temperaturę” (wynik jest ważny).

Uruchamianie Callable: aby otrzymać wynik, użyj executor.submit(...). Zwróci on obiekt Future<T>.

3. Future: obietnica rezultatu

Future to „obietnica” zwrotu wyniku w przyszłości. Gdy wysyłasz zadanie do ExecutorService, otrzymujesz Future, z którego później możesz pobrać wynik, sprawdzić, czy zadanie się zakończyło, lub je anulować.

Najważniejsze metody Future

• T get() — pobiera wynik (czeka, aż zadanie się zakończy).
• boolean isDone() — czy zadanie zostało zakończone.
• boolean cancel(boolean mayInterruptIfRunning) — próba anulowania zadania.
• boolean isCancelled() — czy zadanie zostało anulowane.

Przykład: uruchomienie Callable i pobranie wyniku

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("Zadanie uruchomione, można robić coś jeszcze...");

        // Pobieramy wynik (metoda blokuje wątek, jeśli zadanie jeszcze się nie zakończyło)
        Integer result = future.get();
        System.out.println("Wynik obliczeń: " + result);

        executor.shutdown();
    }
}

• Zadanie trafia do puli wątków.
• Gdy zadanie się wykonuje, wątek główny może robić coś innego.
• Gdy wynik jest potrzebny, wywołujemy future.get() — wątek poczeka, jeśli zadanie jest w toku.
• Gdy tylko zadanie się zakończy, wynik zostanie zwrócony.

4. Praktyka: wiele zadań i oczekiwanie na zakończenie

Często trzeba uruchomić kilka zadań naraz i poczekać, aż wszystkie się zakończą. Na przykład przetwarzamy tablicę danych, dzielimy ją na części i liczymy sumę każdej części w osobnym zadaniu.

Przykład: suma elementów tablicy w częściach

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); // Wypełniamy liczbami od 1 do 1000

        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("Suma od " + from + " do " + (to - 1) + " = " + sum);
                return sum;
            };

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

        int totalSum = 0;
        for (Future<Integer> f : futures) {
            totalSum += f.get(); // Czekamy na każde zadanie po kolei
        }

        System.out.println("Suma całkowita: " + totalSum);

        executor.shutdown();
    }
}

Tutaj tablica jest dzielona na 4 części. Dla każdej części tworzone jest zadanie (Callable), które liczy sumę. Wszystkie zadania są wysyłane do ExecutorService, zwracane są Future. Na końcu zbieramy wyniki wszystkich zadań i sumujemy.

W realnych zadaniach wygodnie jest użyć invokeAll, aby od razu poczekać na wykonanie wszystkich zadań.

5. Obsługa błędów podczas pracy z Future

Gdy wywołujesz future.get(), jeśli zadanie zakończyło się wyjątkiem, zostanie on opakowany i rzucony jako ExecutionException. To ważne: jeśli w zadaniu coś poszło nie tak, dowiesz się o tym dopiero przy wywołaniu get().

Przykład: obsługa wyjątków

Callable<Integer> errorTask = () -> {
    throw new IllegalArgumentException("Coś poszło nie tak!");
};

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

try {
    badFuture.get();
} catch (ExecutionException e) {
    System.out.println("Zadanie zakończyło się błędem: " + e.getCause());
}

• Wewnątrz zadania wyrzucany jest wyjątek.
• Przy wywołaniu get() jest „opakowywany” w ExecutionException.
• Prawdziwą przyczynę można uzyskać przez getCause().

6. Przydatne niuanse

Jak anulować zadanie

Future<?> f = executor.submit(() -> {
    while (true) {
        // Nieskończona praca
        if (Thread.currentThread().isInterrupted()) {
            System.out.println("Poproszono mnie o zakończenie!");
            break;
        }
    }
});

Thread.sleep(100); // Poczekajmy chwilkę
f.cancel(true); // Spróbujmy anulować zadanie

• cancel(true) próbuje przerwać zadanie, jeśli jeszcze się nie zakończyło.
• Wewnątrz zadania warto sprawdzać Thread.currentThread().isInterrupted() i poprawnie się kończyć.

shutdown vs shutdownNow

shutdown() — łagodne zatrzymanie: zabrania dodawania nowych zadań i pozwala aktualnym spokojnie się zakończyć. Używane najczęściej.

shutdownNow() — twarde zatrzymanie: próbuje przerwać aktywne wątki i zwraca listę zadań, które nie zdążyły wystartować. Stosuj ostrożnie.

invokeAll i invokeAny

invokeAll(Collection<Callable<T>> tasks) uruchamia wszystkie przekazane zadania i czeka, aż wszystkie się zakończą. Zwraca listę Future.

invokeAny(Collection<Callable<T>> tasks) czeka tylko na pierwsze pomyślnie wykonane zadanie, zwraca jego wynik i anuluje pozostałe. Wygodne, gdy liczy się pierwsza udana odpowiedź.

7. Typowe błędy przy pracy z ExecutorService, Callable i Future

Błąd nr 1: Niezamykanie ExecutorService. Jeśli zapomnisz wywołać shutdown(), program może „wisieć” po zakończeniu main, ponieważ wątki puli czekają na nowe zadania.

Błąd nr 2: Oczekiwanie na wynik zaraz po wysłaniu zadania. Jeśli od razu po submit() wywołasz get(), nie zyskasz zalet asynchroniczności — wątek i tak będzie czekał. Rób pożyteczną pracę równolegle i pobieraj wynik wtedy, gdy jest naprawdę potrzebny.

Błąd nr 3: Ignorowanie wyjątków w zadaniach. Jeśli nie obsłużysz ExecutionException przy wywołaniu get(), możesz przeoczyć ważne błędy, które wystąpiły w zadaniu.

Błąd nr 4: Używanie wspólnych zmiennych modyfikowalnych bez synchronizacji. Jeśli kilka zadań pracuje na tych samych danych — potrzebna jest synchronizacja lub kolekcje bezpieczne dla wątków.

Błąd nr 5: Tworzenie zbyt dużej liczby wątków. Nie warto robić puli z liczbą wątków znacznie większą niż liczba rdzeni procesora — może to nawet spowolnić wykonanie.

Błąd nr 6: Zapominanie o anulowaniu zadań. Jeśli zadanie nie jest już potrzebne, anuluj je przez cancel(), aby nie marnować zasobów.