streamalism.org

1. Bemutatkozás

Ebben a cikkben útmutatást adunk a CountDownLatch osztályban, és néhány gyakorlati példában mutassa be, hogyan használható.

Lényegében az a használatával CountDownLatch egy szál blokkolását okozhatjuk, amíg más szálak nem teljesítik az adott feladatot.

2. Használat a párhuzamos programozásban

Egyszerűen fogalmazva: a CountDownLatch van egy számláló mező, amelyet csökkenthet, ahogyan azt igényeljük. Ezután blokkolhatunk egy hívó szálat, amíg le nem számoljuk nulláig.

Ha valamilyen párhuzamos feldolgozást végeznénk, akkor a CountDownLatch a számlálóval megegyező értékkel, mint számos szálon, amelyeken át akarunk dolgozni. Akkor hívhatnánk visszaszámlálás() miután minden szál befejeződött, garantálva, hogy egy függő szál hívjon várják() addig blokkol, amíg a munkásszálak be nem fejeződnek.

3. Várakozás a szálkészlet befejezésére

Próbáljuk ki ezt a mintát a Munkás és a CountDownLatch jelző mező, ha elkészült:

public class Munkavállaló Runnable {private List outputScraper; privát CountDownLatch countDownLatch; public Worker (List outputScraper, CountDownLatch countDownLatch) {this.outputScraper = outputScraper; this.countDownLatch = countDownLatch; } @Orride public void run () {doSomeWork (); outputScraper.add ("Visszaszámolva"); countDownLatch.countDown (); }}

Ezután hozzunk létre egy tesztet annak igazolására, hogy a CountDownLatch megvárni a Munkás kitöltendő példányok:

@Test public void, amikorParallelProcessing_thenMainThreadWillBlockUntilCompletion () dobja az InterruptedException {List outputScraper = Collections.synchronizedList (new ArrayList ()); CountDownLatch countDownLatch = new CountDownLatch (5); Munkavállalók listája = Stream .generate (() -> new Thread (new Worker (outputScraper, countDownLatch))) .limit (5) .collect (toList ()); dolgozók.minden (Szál :: kezdet); countDownLatch.await (); outputScraper.add ("Zár kioldva"); assertThat (outputScraper) .containsExactly ("Counted down", "Counted down", "Counted down", "Counted down", "Counted down", "Retesz oldva"); }

Természetesen mindig a „Kinyitott retesz” lesz az utolsó kimenet - mivel ez függ a programtól CountDownLatch felszabadító.

Vegye figyelembe, hogy ha nem hívtunk várják(), nem tudnánk garantálni a szálak végrehajtásának sorrendjét, így a teszt véletlenszerűen kudarcot vallana.

4. A szálkészlet kezdetre vár

Ha az előző példát vettük, de ezúttal több ezer szálat indítottunk el öt helyett, akkor valószínűleg a korábbiak közül sokan befejezik a feldolgozást, mielőtt még felhívtuk volna Rajt() a későbbieken. Ez megnehezítheti a párhuzamossági probléma kipróbálását és reprodukálását, mivel nem tudjuk elérni, hogy az összes szálunk párhuzamosan futjon.

Hogy ezt megkerüljük, keressük meg a CountdownLatch hogy az előző példánál eltérően működjön. Ahelyett, hogy blokkolnánk egy szülőszálat, amíg néhány gyermekszál nem fejeződik be, blokkolhatunk minden egyes szálszálat, amíg az összes többi el nem indul.

Módosítsuk a fuss() módszer, így blokkolja a feldolgozás előtt:

public class WaitingWorker megvalósítja a Runnable {private List outputScraper; privát CountDownLatch readyThreadCounter; privát CountDownLatch hívásThreadBlocker; privát CountDownLatch befejezettThreadCounter; public WaitingWorker (List outputScraper, CountDownLatch readyThreadCounter, CountDownLatch callingThreadBlocker, CountDownLatch completeThreadCounter) {this.outputScraper = outputScraper; this.readyThreadCounter = readyThreadCounter; this.callingThreadBlocker = callingThreadBlocker; this.completedThreadCounter = befejezettThreadCounter; } @Orride public void run () {readyThreadCounter.countDown (); próbáld meg a {callingThreadBlocker.await (); Dolgozz valamit(); outputScraper.add ("Visszaszámolva"); } catch (InterruptedException e) {e.printStackTrace (); } végül {completeThreadCounter.countDown (); }}}

Most módosítsuk a tesztünket, hogy blokkolja az összes Munkások megkezdték, feloldja a Munkavállalók, majd blokkol, amíg a Munkások befejezte:

@Test public void whenDoingLotsOfThreadsInParallel_thenStartThemAtTheSameTime () thrumps InterruptedException {List outputScraper = Collections.synchronizedList (new ArrayList ()); CountDownLatch readyThreadCounter = new CountDownLatch (5); CountDownLatch hívásThreadBlocker = új CountDownLatch (1); CountDownLatch completeThreadCounter = new CountDownLatch (5); Dolgozók listája = Stream .generate (() -> new Thread (új WaitingWorker (outputScraper, readyThreadCounter, callingThreadBlocker, completeThreadCounter)))) .limit (5) .collect (toList ()); dolgozók.minden (Szál :: kezdet); readyThreadCounter.await (); outputScraper.add ("Munkavállalók készen állnak"); callingThreadBlocker.countDown (); completeThreadCounter.await (); outputScraper.add ("A dolgozók teljesek"); assertThat (outputScraper) .containsExactly ("Munkavállalók készen", "Visszaszámolva", "Visszaszámolva", "Visszaszámolva", "Visszaszámolva", "Visszaszámolva", "A dolgozók teljesek"); }

Ez a minta nagyon hasznos a párhuzamossági hibák reprodukálásához, mivel arra használható, hogy több ezer szálat arra kényszerítsen, hogy párhuzamosan hajtsanak végre valamilyen logikát.

5. Megszüntetése a CountdownLatch Korai

Néha olyan helyzetbe kerülhetünk, amikor a Munkások hibásan megszakítja, mielőtt visszaszámolja a CountDownLatch. Ez azt eredményezheti, hogy soha nem éri el a nullát, és várják() soha nem szűnik meg:

@Orride public void run () {if (true) {dobj új RuntimeException-t ("Ó, kedvesem, BrokenWorker vagyok"); } countDownLatch.countDown (); outputScraper.add ("Visszaszámolva"); }

Módosítsuk korábbi tesztünket az a használatához BrokenWorker, annak bemutatása érdekében várják() örökre blokkolni fogja:

@Test public void whenFailingToParallelProcess_thenMainThreadShouldGetNotGetStuck () thrumps InterruptedException {List outputScraper = Collections.synchronizedList (new ArrayList ()); CountDownLatch countDownLatch = new CountDownLatch (5); Munkavállalók listája = Stream .generate (() -> new Thread (új BrokenWorker (outputScraper, countDownLatch))) .limit (5) .collect (toList ()); dolgozók.minden (Szál :: kezdet); countDownLatch.await (); }

Nyilvánvaló, hogy nem ezt a viselkedést szeretnénk - sokkal jobb lenne, ha az alkalmazást folytatnánk, mintsem végtelenül blokkolnánk.

Ennek kiküszöbölése érdekében tegyünk hozzá időtúllépés-argumentumot a hívásunkhoz várják().

logikai érték befejezve = countDownLatch.await (3L, TimeUnit.SECONDS); assertThat (befejezve) .isFalse ();

Mint láthatjuk, a teszt végül időtúllépésnek indul és várják() vissza fog térni hamis.

6. Következtetés

Ebben a gyors útmutatóban bemutattuk, hogyan használhatjuk a CountDownLatch egy szál blokkolásához, amíg a többi szál befejezi a feldolgozást.

Megmutattuk azt is, hogyan lehet a párhuzamossági problémák hibakeresésében segíteni azáltal, hogy a szálak párhuzamosan futnak.

Ezeknek a példáknak a megvalósítása megtalálható a GitHub oldalon; ez egy Maven-alapú projekt, ezért könnyen futtathatónak kell lennie.