Kihagyás

4. gyakorlat

A java.util.concurrent csomag lehetővé teszi a hatékonyabb, skálázhatóbb és biztonságosabb párhuzamos programozást. Péládul a szemaforok segítségével dinamikusan vezérelhető, hogy hány szál férhet hozzá egy adott erőforráshoz: egy adatbáziskapcsolat-kezelő osztályban egy számláló szemafor biztosíthatja, hogy egyszerre maximum n darab kapcsolat legyen nyitva, ahelyett, hogy minden szál egyszerre próbálna hozzáférni az adatokhoz.

Szinkronizáció szemaforral

A korábbi gyakorlaton megismert szemafort szinkronizációra is fel lehet használni. Ezen a ponton ez azért is hasznos lehet, mert korábban csak a join() segítségével tudunk egy speciális szinkronizációt előidézni, azonban ilyen esetben az egyik folyamat mindenképpen terminált (hiszen erre várt a hívó szál).

Folyamatinterakciók: szinkronizáció

A szinkronizációban egy küldő és egy fogadó folyamat vesz részt, mindkettőnek van egy szinkronizációs pontja. A fogadó nem lépheti át a saját szinkronizációs pontját, amíg a küldő el nem érte a sajátját, mivel a fogadónak szüksége van a küldő által előállított adatokra. Ha a fogadó ér hamarabb a szinkronizációs ponthoz, várakozik a küldőre. Ha a küldő ér oda előbb, kétféle viselkedés lehetséges: vagy bevárja a fogadót (szimmetrikus), vagy mindkét folyamat áthaladhat várakozás nélkül (asszimmetrikus).

Ahhoz, hogy a szinkronizációt megvalósítsuk, Semaphore kezdőértékét 0-ra állítjuk. A küldő szál szinkronizációs pontjához elhelyezzük a release(), a fogadó szinkronizációs pontjánál pedig az acquire() hívást. 

 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.concurrent.Semaphore;

public class SemaphoreExample {

    static void print(long id, int iteration) {
        System.out.println(id + ": " + iteration);
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(0);

        new Thread(() -> {
            print(Thread.currentThread().threadId(), 1);

            System.out.println(Thread.currentThread().threadId() + " is waiting.");
            try {
                semaphore.acquire();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            print(Thread.currentThread().threadId(), 2);
            print(Thread.currentThread().threadId(), 3);
            print(Thread.currentThread().threadId(), 4);
            print(Thread.currentThread().threadId(), 5);
            print(Thread.currentThread().threadId(), 6);
        }).start();

        new Thread(() -> {
            print(Thread.currentThread().threadId(), 1);
            print(Thread.currentThread().threadId(), 2);
            print(Thread.currentThread().threadId(), 3);
            print(Thread.currentThread().threadId(), 4);
            print(Thread.currentThread().threadId(), 5);
            System.out.println(Thread.currentThread().threadId() + " is sending a signal.");
            semaphore.release();
            print(Thread.currentThread().threadId(), 6);
        }).start();
    }
}

Szimmetrikus vagy asszimmetrikus?

Módosítsuk a programot. Milyen viselkedést tapasztalunk, ha a küldő szál ér előbb a saját szinkronizációs pontjába?

Szinkronizáció CyclicBarrier-rel

Szintén egy szinkronizációs mechanizmust biztosít a CyclicBarrier, amely kifejezetten a szálak koordinációjára szolgál, úgy, hogy tetszőleges mennyiségű szál várja be egymást egy közös szinkronizációs pontban. Ez a szinkronizációs pont lehetővé teszi, hogy a szálak csak akkor folytathassák a munkát, ha minden szál elérte a megadott pontot. A CyclicBarrier neve arra utal, hogy a szinkronizáció ciklikusan ismételhető, azaz többször is felhasználható a szálak közötti szinkronizáció megvalósításához.

A CyclicBarrier hatékonyabb és könnyebben alkalmazható a szálak szinkronizációjára, mint a Semaphore, mivel erre a feladatra specializálódott. Ezen felül a CyclicBarrier lehetővé teszi, hogy egy opcionális műveletet is végrehajtsunk, amikor minden szál elérte a közös pontot, például egy összegzést vagy ellenőrzést, ami még rugalmasabbá teszi a használatát.

A barrier létrehozásakor meghatározzuk, hogy hány szálat kell bevárni a szinkronizációs pontban. Amikor egy szál eléri ezt a pontot, meghívja az await() metódust. Ez a metódus csökkenti a barrier számlálóját, és addig blokkolja a szálat. Amint az utolsó szál is eléri az ún. barrier pontot, azaz a barrier számlálója nullára csökken, a szálak folytathatják a végrehajtást.

A barrier objektum példányosításakor megadható egy Runnable is opcionálisan. Fontos, hogy bár a művelet ekkor fut le, amikor az utolsó szál is meghívta az await()-et. A szálak csak akkor folytathatja a végrehajtást ha a Runnable művelet befejeződött. Ez biztosítja, hogy a közös művelet (például összegzés vagy ellenőrzés) befejeződjön, mielőtt bármelyik szál folytatná a saját végrehajtását.

 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
45
46
47
48
49
50
51
52
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierExample {
    public static void main(String[] args) {
        int numberOfThreads = 5;

        CyclicBarrier barrier = new CyclicBarrier(5, new BarrierTask());

        for (int i = 0; i < numberOfThreads; i++) {
            Thread thread = new Thread(new Task(barrier));
            thread.start();
        }
    }
}

class BarrierTask implements Runnable {
    @Override
    public void run() {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        System.out.println("A barrier feladata csak egyszer futott le.");
    }
}

class Task extends Thread {

    private final CyclicBarrier barrier;

    public Task(CyclicBarrier barrier) {
        this.barrier = barrier;
    }

    @Override
    public void run() {
        try {
            System.out.println(this.getName() + " dolgozik...");
            Thread.sleep(new Random().nextInt(1000)); 

            System.out.println(this.getName() + " a szinkronizációs pontban..(" +  (barrier.getNumberWaiting()+1) +")");
            barrier.await();

            System.out.println(this.getName() + " folytatja a munkát.");
        } catch (InterruptedException | BrokenBarrierException e) {
            e.printStackTrace();
        }
    }
}

Atomic változók

Az atomic változók célja, hogy szálbiztos műveleteket biztosítsanak egyszerű adattípusokon (például számlálókon), anélkül hogy kölcsönös kizárást (például synchronized blokkokat) kellene alkalmazni. Az atomic változók belső mechanizmusukkal garantálják, hogy az adott műveletek atomi módon végrehajtásra kerül. Mivel nem használnak hagyományos zárolási mechanizmusokat, hatékonyabb alternatívát kínálnak, és segítenek elkerülni a zárólás teljesítménybeli "költségét", miközben biztosítják, hogy a műveletek szálbiztos módon történjenek.

Számos atomic osztály elérhető különböző adattípusok kezelésére. Ezek közül néhány (a többiről itt lehet olvasni):

  • AtomicInteger: Szálbiztos egész számok kezelésére.

  • AtomicLong: Szálbiztos hosszú egész számok kezelésére.

  • AtomicBoolean: Szálbiztos logikai értékek kezelésére.

  • AtomicReference: Szálbiztos objektum referenciák kezelésére.

Az atomic osztályok olyan metódusokat biztosítanak, amelyek garantálják a szálbiztos műveletek végrehajtását:

  • get(): Az aktuális érték lekérdezése.

  • set(newValue): Az aktuális érték beállítása.

  • incrementAndGet(): Az érték növelése 1-gyel, majd az új érték visszaadása (csak AtomicInteger és AtomicLong esetén).

  • decrementAndGet(): Az érték csökkentése 1-gyel, majd az új érték visszaadása (csak AtomicInteger és AtomicLong esetén).

  • addAndGet(delta): Egy adott érték hozzáadása, majd az új érték visszaadása (csak AtomicInteger és AtomicLong esetén).

  • compareAndSet(expected, update): Feltételes frissítés, amely az update értékét csak akkor állítja be, ha az aktuális érték megegyezik az expected értékkel.

 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
45
46
47
48
49
50
51
52
53
import java.util.concurrent.atomic.AtomicInteger;

public class AtomicCounter {

    int number1;
    AtomicInteger number2 = new AtomicInteger();

    public void incNumberOne() {
        number1++;
    }

    public void incNumberTwo() {
        number2.incrementAndGet();
    }

    public static void main(String[] args) {
        AtomicCounter ac = new AtomicCounter();

        for (int i = 0; i < 10_000; i++) {
            Thread myThread = new AtomicThread(ac);
            myThread.start();
        }

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("Final result (number1): " + ac.number1);
        System.out.println("Final result (number2): " + ac.number2.get());
    }
}

class AtomicThread extends Thread {
    AtomicCounter ac;

    AtomicThread(AtomicCounter ac) {
        this.ac = ac;
    }

    @Override
    public void run() {
        try {
            Thread.sleep(300);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        ac.incNumberOne();
        ac.incNumberTwo();
    }
}

Gyakorló feladatok

Feladat

Hozz létre egy tömböt, ami kettő lebegőpontos értéket tud tárolni és inicializáld negatív számokkal.

Továbbá hozz létre két szálat: az egyik módosítsa a tömb első értékét egy tetszőleges számmal, majd szinkronizáció segítségével jelezzen a másik szálnak, hogy elkészült a módosítással, míg a a másik szál a szinkronizációt követően adja hozzá a tömb első értékét a másodikhoz.

A main szál várja be a létrehozott szálak terminálását, majd írja ki a tömb tartalmát. Ne használj altatást és kölcsönös kizárást. A szinkronizációt szemafor segítségével valósítsd meg.

Feladat

Hozz létre egy Resource nevű osztályt, amelynek konstruktorában egy egész szám típusú paramétert fogad, amely a szálak számát jelöli. A konstruktor feladata, hogy inicializálja és feltöltse az array nevű adattagot, amely egy egész számokat tartalmazó tömb. A tömb mérete a szálak számának százszorosával lesz egyenlő (szálak száma × 100), a szálak számát paraméterként kapja meg a konstruktor.

Ezután hozz létre egy Calculator osztályt, amely a Thread osztályból származik és implementálja a run() metódust. A Calculator osztály konstruktora négy paramétert fogad:

  • Egy Resource típusú objektum referenciáját, amely az egész számokat tartalmazó tömböt tárolja,

  • Egy egész számot, amely az intervallum kezdőértékét jelöli,

  • Egy másik egész számot, amely az intervallum végértékét adja meg.

  • Végezetül egy CyclicBarrier objektumra mutató referenciát.

A run() metódus feladata, hogy végigiteráljon a megadott intervallumon belül található számokon az array tömbben, és az összeget eltárolja a Calculator osztály sum nevű adattagjában.

A fő programban ne a join() metódussal biztosítsuk a szálak bevárását, hanem egy CyclicBarrier objektum segítségével, amely példányosításakor egy Runnable objektumot is megkap. Ez a Runnable az összes létrehozott szálat tartalmazó kollekcióra mutató referenciát kap, és a feladata, hogy össszegezze a Calculator szálak részszámításait. Ne használj kölcsönös kizárást vagy altatást.

Feladat

Hozz létre egy AtomicReference<String> változót, kezdőértéke legyen "Idle".

Hozz létre 10 szálat, amelyek mindegyike végrehajtja a következő lépéseket:

  • Ha az érték "Idle", akkor módosítja azt "InProgress"-ra, egyébként várakozik az erőforrásra. Ezt tevékeny várakozással végzi, tehát egy ciklus segítségével folyamatosan ellenőrzi az adattag értékét.

  • "InProgress" állapotban altatás segítségével várakoztatjuk a szálat 100 ms-ig, majd visszaállítjuk az értéket "Idle"-ra.

A megoldáshoz kizárólag a compareAndSet és a set metódusokat használd.

A műveletek elvégzét követően termináljon a program.

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