Kihagyás

5. gyakorlat

A Runnable mellett egy másik interfészt is használhatunk arra, hogy egy adott kódrészletet külön szálon futtassunk: ez lesz a Callable. Ez az interfész rugalmasabb megoldást kínál a korábban megismert megközelítésekhez képest, mivel a call() metódusa nemcsak visszatérési értékkel rendelkezhet, hanem kivételt is dobhat, amit a hívó metódusnak kell lekezelnie vagy továbbítania. Az előbbi tulajdonság különösen előnyös abban az esetben, ha a párhuzamosan végrehajtott feladat egy számítás eredményét adja vissza, amelyet később felhasználhatunk.

A Callable interfész nem használható közvetlenül egy Thread-del (hiszen az egy Runnable interfészt vár paraméterként), de demonstrációs céllal, külön szál létrehozása nélkül is tudjuk futtatni (mivel lényegében csak egy végrehajtandó kódrészletet definiálunk).

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
public class CallableExample {
    public static void main(String[] args) {
        Callable<Integer> callableTask = new ExampleCallable();
        try {
            System.out.println("Return value: " + callableTask.call() + " executed by: " + Thread.currentThread());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

class ExampleCallable implements Callable<Integer> {

    @Override
    public Integer call()  { // a run() nem dobhat kivételt
        System.out.println(Thread.currentThread().getName() + " is counting..");
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        int result = (int) (Math.random() * 100);
        System.out.println(Thread.currentThread().getName() + "'s result: " + result);
        return result;
    }
}

ExecutorService

Az ExecutorService egy magasabb szintű API Java-ban, amely megkönnyíti a szálkezelést. Arra használható, hogy hatékonyan kezelje és újrahasznosítsa a szálakat, elkerülve az egyes szálak kézi létrehozásának és kezelésének nehézségeit.

Thread végállapot

Ahogy azt a korábbi órák egyikén megtanultuk, egy TERMINATED állapotban lévő szálat nem tudunk ismét elindíti, helyette új szálat kellene példányosítanunk. Egy Thread létrehozása erőforrásigényes, különösen ha sok szálra van szükség a végrehajtás során.

Az ExecutorService megoldja ezt a problémát egy ún. thread pool segítségével, amelyben előre létrehozott és újrahasznosítható szálak dolgoznak. Ezek a szálak fogják megkapni a Callable feladatokat az ExecutorService által meghatározott ütemezés alapján. Ha több feladatot adunk át futtatásra az ExecutorService-nek, mint amennyi szál a thread pool-ban elérhető, akkor a feladatokat egy belső várakozási sorban tárolja. Ha egy szál végzett egy feladattal, akkor nem kerül leállításra, hanem ismét munkára fogható. Ez egy hatékonyabb és egyszerűbb megközelítés, mint manuálisan létrehozni és kezelni az egyes szálakat.

Az Executors osztály segítségével különböző típusú thread poolokat hozhatunk létre:

  • Executors.newFixedThreadPool(int threads): Fix számú szálat tartalmazó pool kerül létrehozásra.

  • Executors.newCachedThreadPool(): Dinamikusan, a feladatok számától függően változó szálkészlet kerül létrehozásra.

  • Executors.newSingleThreadExecutor(): Egyetlen szállal dolgozó pool kerül létrehozásra.

Ismétlődő feladatok

A ScheduledThreadPoolExecutor egy olyan ExecutorService implementáció, amely időzített vagy ismétlődő feladatokat tud végrehajtani. Ez a megoldás különösen hasznos, ha például időszakos karbantartási feladatokat kell végezni, vagy a feladatokat adott időpontban kell elindítani.

ExecutorService legfontosabb metódusai:

  • shutdown(): A szálak szabályos leállítására szolgál, de a már beküldött (várakozó) és a futó feladatok végrehajtódnak.

  • submit(Callable<T> task): Elindít egy feladatot és azonnal visszatér, lehetővé téve, hogy a hívó szál folytassa a végrehajtást, miközben a háttérfeladat párhuzamosan fut (azaz nem blokkolja a hívó szálat!)

  • invokeAll(Collection<Callable<T>> tasks): A kollekcióban átadott feladatokat futtatja az elérhető szálakon párhuzamosan, és blokkolja a hívó szálat, amíg mindegyik feladat be nem fejeződik.

ExecutorService + Runnable

Az ExecutorService képes Runnable feladatok fogadására és végrehajtására is, és szintén automatikusan kezeli a szálak életciklusát, így nem kell manuálisan létrehozni a szálakat. 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExecutorServiceExample {
    public static void main(String[] args) {

        ExecutorService executor = Executors.newCachedThreadPool();

        for(int i=0; i<10; i++) {
            executor.submit(new MyCallable());
            try {
                Thread.sleep(25);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        executor.shutdown();
    }
}

class MyCallable implements Callable<Integer>{

    @Override
    public Integer call() {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println(Thread.currentThread().getName() + "-" + this);
        return 0;
    }
}

Future

A Future interfész lehetővé teszi, hogy kezeljük az aszinkron feladatok eredményeit. Az aszinkron programozásban a feladatokat háttérszálakon indítjuk el, így a hívó szál nem várja meg azok befejeződését, és folytathatja a saját munkáját. A Future segítségével később lekérhetjük a feladat eredményét, de a lekérés maga már blokkoló hívás lesz, ha a feladat még nem fejeződött be.

Párhuzamos vs. aszinkron programozás

A párhuzamos programozás azt jelenti, hogy több feladat egyszerre fut különböző szálakon. Bár a feladatok párhuzamosan futnak, ez nem feltétlenül jelent aszinkron működést, mert a hívó szál is blokkolódhat, amíg a párhuzamos feladatok befejeződnek.

Aszinkron programozás esetén a háttérben futó aszinkron feladatok nem blokkolják a hívó szálat, tehát miközben a háttérfeladatok futnak, a hívó szál tovább végezheti a saját dolgát. Az eredmény később, szükség esetén, blokkoló módon kérhető le a feladatok befejeződése után.

A korábban ismertetett submit(..) és invokeAll(..) metódusok az ExecutorService implementációk részeként Future objektumokat adnak vissza, amelyek a futó feladatok eredményének kezelésére szolgálnak. A Future-höz tartozó legfontosabb metódusok:

  • get(): Blokkolja a hívó szálat várva a feladat befejezésére, majd visszatér az eredménnyel.

  • isDone(): Ellenőrzi, hogy a feladat befejeződött-e.

  • cancel(): Megszakítja a futó feladatot.

  • isCancelled(): Ellenőrzi, hogy a feladatot megszakították-e.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class FutureExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(5);

        List<RandomCallable> callableList = new ArrayList<RandomCallable>();
        List<Future<Integer>> futures = new ArrayList<Future<Integer>>();

        for(int i=0; i<10; i++) {
            callableList.add(new RandomCallable());
        }

        try {
            futures = executor.invokeAll(callableList);
            int sum = 0;
            for(Future<Integer> future : futures) {
                sum += future.get();
            }
            System.out.println("Result: " + sum);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        executor.shutdown();
    }
}

class RandomCallable implements Callable<Integer>{
    @Override
    public Integer call() throws InterruptedException {
        Thread.sleep(500);
        int res = new Random().nextInt(100);
        System.out.println(this + " is counting.. " + res);
        return res;
    }
}

Virtuális szálak

A Project Loom célja, hogy a Java-ban elérhetővé tegye az ún. virtuális szálak használatát, amelyek kevesebb erőforrást igényelnek, mint a hagyományos szálak. Ennek köszönhetően lehetőség nyílik nagyszámú virtuális szál hatékony kezelésére. A virtuális szálak kevesebb memóriahasználattal járnak, mivel nem közvetlenül az operációs rendszer által kezelt natív szálakhoz kapcsolódnak. Ezeknek a szálak ütemezése a JVM-en belül történik, így nagyságrendekkel több szál indítható el.

A virtuális szálak kiválóan alkalmasak nagy számú egyidejű kapcsolat kezelésére, például HTTP szerverek, adatbázis-kezelő rendszerek, mikroszolgáltatások és aszinkron eseménykezelő rendszerek esetén. Fő előnyeik a kisebb memóriahasználat, jobb skálázhatóság és egyszerűbb programozási modell, különösen I/O-intenzív feladatoknál. Hátrányuk, hogy a blokkoló műveletek csökkenthetik a hatékonyságukat, és CPU-intenzív feladatok esetén nem feltétlenül jobbak a hagyományos szálaknál. Emellett egyes régebbi Java könyvtárak nem támogatják őket, ami kompatibilitási problémákat okozhat.

Amikor a Java futtatókörnyezet ütemez egy virtuális szálat, akkor azt hozzárendeli egy ún. platform thread-hez (hordozó szálhoz). A hordozó szál már az operációs rendszer által ütemezett hagyományos szál. Virtuális szálakat létre tudunk hozni ExecutorService segítségével: Executors.newVirtualThreadPerTaskExecutor(). Ez minden egyes feladathoz új virtuális szálat hoz létre.

Blokkoló műveletek

Amikor a Java futtatókörnyezet ütemez egy virtuális szálat, az hozzárendelésre kerül egy platformszálhoz, amelyet az operációs rendszer a szokásos módon ütemez. Miután a virtuális szál végrehajtott valamennyi kódot, lecsatlakozhat a hordozó szálról (tipikusan amikor blokkoló műveletet hajt végre). A hordozó szál így felszabadul, és más virtuális szálakat is kiszolgálhat.

Blokkoló művelet

Olyan művelet, amely megakadályozza a szál további végrehajtását, amíg egy külső feltétel teljesül. Ilyenek például az I/O műveletek (pl. fájlolvasás vagy hálózati kommunikáció) és a zárolási mechanizmusok közbeni várakoztatás.

Például ha egy program fájlt próbál beolvasni, akkor várnia kell, amíg az operációs rendszer betölti a fájlt a háttértárból a memóriába. Ez idő alatt a program nem folytathatja a futást, várakoznia kell, ami viszont megakadályozza a további műveletek végrehajtását.

Egy virtuális szál nem tud lecsatlakozni blokkoló műveletek közben, ha rögzítve van (pinned) a hordozó szálhoz. Esetünkben a rögzítés akkor fordul elő, amikor a virtuális szál végrehajtása egy synchronized blokkban vagy metódusban van éppen és várakozásra kényszerül. Így ha a rögzített virtuális szálhoz tartozó platformszál nem tud más virtuális szálakat kiszolgálni, ami csökkenti a párhuzamosan futtatható szálak számát, és ezzel ronthatja a skálázódást.

JEP 491

A Java Enhancement Proposal 491 célja ezt a korlátozást orvosolni azáltal, hogy a monitor kezelését a platformszálak helyett a virtuális szálakhoz rendeli. Ez lehetővé teszi, hogy a virtuális szálak a synchronized blokkok vagy metódusok végrehajtása közben is szabadon lecsatlakozzanak a platformszálatakról. Ezt várhatóan a 2025 márciusában publikálásra kerülő JDK 24 fogja tartalmazni először.

A jelenlegi monitor koncepciójának korlátozása orvosolható Lock vagy Semaphore használatával, mivel ezek nem kötődnek közvetlenül a platformszálakhoz, így elkerülhető a virtuális szálak rögzítése.

Gyakorló feladatok

Feladat

Írj egy programot a a Pi értékének közelítésére, amely virtuális szálakat használó ExecutorService-t, Future-t és Callable-t használ.

A Pi közelítését a geometriai valószínűség felhasználásval végezzük el. Ha egy egységnyi oldalhosszúságú négyzetbe egy minden oldalát érintő kört rajzolunk, majd az így kapott alakzat pontjait véletlenül választjuk ki elég sokszor, akkor a körre eső pontok száma úgy aránylik a négyzetre eső pontok számához, mint a kör területe a négyzet területéhez. Ezután a tört egyszerűsítésével és átrendezésével megkaphatjuk a Pi közelített értékét.

Tehát nincs más feladatunk, mint véletlenszerűen választani a négyzet területéről egy pontot a szálak segítségével, a generált pontok számát nyilvántartani (legyen N), majd megnézni, hogy a kör területére esett-e, és ha igen akkor azt külön is nyilvántartjuk (K). A program terminálásakor a 4*K/N művelet adja meg a Pi közelített értékét.

A MyCallable osztály Point típusú visszatérési értékkel rendelkezzen, amelynek az x és y adattagja reprezentálja a négyzetre eső pontot. Hozz létre százezer virtuális szálat, majd Future segítségével biztosítsd a szálak által végzett művelet eredményének felhasználását a main metódus lokális K és az N változóinak segítségével. A program írja ki a Pi közelített értékét, majd szabályosan termináljon. Egy pont a (0,0) középpontú kör területére esik, ha x^2 + y^2 <= r^2.

Ne használj a feladat által nem említett párhuzamos konstrukciót.

Feladat

Implementálj egy osztályt egy parkoló reprezentálására, ParkingLot néven, amely 10 kocsit képes befogadni. Készíts el az osztályhoz a következő metódusokat:

  • enter(): beenged egy kocsit a parkolóba, amennyiben van szabad hely, egyéb esetben a kocsinak várakoznia kell.

  • parking(): 300 ms-ig altatja a szál végrehajtását.

  • leave(): kiléptet egy kocsit a parkolóból, és értesíti az esetlegesen várakozókat a parkolóhely felszabadulásáról.

Egy számláló Semaphore vagy a Lock+Condition interfész segítségével biztosítsd, hogy ne tartózkodjon egyidőben a parkolóban a megengedettnél több jármű.

Készíts egy Car osztályt, amely a Callable interfészt valósítja meg Void visszatérési értékkel. Az osztály a konstruktorában egy ParkingLot objektumot kap. A szál feladata az enter(), parking() és a leave() metódusok meghívása a megadott sorrendben.

Teszteld a programot a main metódusban! Hozz létre egy ParkingLot objektumot és 100 párhuzamosan futó Car feladatot, illetve egy fix méretű ExecutorService-t 5 szállal.

Ne használj a feladat által nem említett párhuzamos konstrukciót.

Utolsó frissítés: 2025-03-23 20:14:57