streamalism.org

1. Prehľad

V tomto článku sa pozrieme na Phaser konštruovať z java.util.concurrent balíček. Je to veľmi podobný konštrukt ako CountDownLatch čo nám umožňuje koordinovať vykonávanie vlákien. V porovnaní s CountDownLatch, má niektoré ďalšie funkcie.

The Phaser je bariéra, na ktorú musí dynamický počet vlákien čakať pred pokračovaním v spustení. V CountDownLatch toto číslo nie je možné nakonfigurovať dynamicky a je potrebné ho uviesť pri vytváraní inštancie.

2. Phaser API

The Phaser nám umožňuje budovať logiku, v ktorej vlákna musia pred prechodom na ďalší krok vykonania počkať na bariére.

Môžeme koordinovať viac fáz vykonania, opätovné použitie a Phaser pre každú fázu programu. Každá fáza môže mať iný počet vlákien čakajúcich na postup do inej fázy. Neskôr sa pozrieme na príklad použitia fáz.

Ak sa chcete zúčastniť koordinácie, vlákno musí Registrovať() sám s Phaser inštancia. Toto iba zvyšuje počet registrovaných strán a nemôžeme skontrolovať, či je aktuálne vlákno zaregistrované - na podporu toho by sme museli podtriedu implementácie.

Vlákno signalizuje, že k bariére dorazilo volaním comeAndAwaitAdvance (), čo je blokovacia metóda. Ak sa počet prichádzajúcich strán rovná počtu registrovaných účastníkov, bude vykonávanie programu pokračovaťa počet fáz sa zvýši. Aktuálne číslo fázy môžeme získať zavolaním na getPhase () metóda.

Keď vlákno dokončí svoju prácu, mali by sme zavolať comeAndDeregister () metóda signalizujúca, že súčasné vlákno by sa už v tejto konkrétnej fáze nemalo zohľadňovať.

3. Implementácia logiky pomocou Phaser API

Povedzme, že chceme koordinovať viac fáz akcií. Tri vlákna spracujú prvú fázu a dve vlákna druhú fázu.

Vytvoríme LongRunningAction trieda, ktorá implementuje Spustiteľné rozhranie:

trieda LongRunningAction implementuje Runnable {private String threadName; súkromný phaser ph; LongRunningAction (reťazec threadName, phaser ph) {this.threadName = threadName; this.ph = ph; ph.register (); } @Override public void run () {ph.arriveAndAwaitAdvance (); skus {Thread.sleep (20); } catch (InterruptedException e) {e.printStackTrace (); } ph.arriveAndDeregister (); }}

Po vytvorení inštancie našej triedy akcií sa registrujeme na serveri Phaser napríklad pomocou Registrovať() metóda. To zvýši počet vlákien pomocou tohto konkrétneho Phaser.

Výzva na comeAndAwaitAdvance () spôsobí, že súčasné vlákno počká na bariéru. Ako už bolo uvedené, v prípade, že sa počet doručených strán zhoduje s počtom zaregistrovaných strán, bude vykonávanie pokračovať.

Po dokončení spracovania sa súčasné vlákno odhlási z registrácie volaním súboru comeAndDeregister () metóda.

Vytvorme testovací prípad, v ktorom začneme traja LongRunningAction závity a blokovať na bariére. Ďalej po dokončení akcie vytvoríme ďalšie dve LongRunningAction vlákna, ktoré vykonajú spracovanie ďalšej fázy.

Pri tvorbe Phaser napríklad z hlavného vlákna, prechádzame 1 ako argument. Toto je ekvivalent volania na Registrovať() metóda z aktuálneho vlákna. Robíme to preto, lebo keď vytvárame tri pracovné vlákna, hlavným vláknom je koordinátor, a teda Phaser musí mať zaregistrované štyri vlákna:

ExecutorService executorService = Executors.newCachedThreadPool (); Phaser ph = nový Phaser (1); assertEquals (0, ph.getPhase ());

Fáza po inicializácii sa rovná nule.

The Phaser trieda má konštruktor, v ktorom do nej môžeme odovzdať nadradenú inštanciu. Je to užitočné v prípadoch, keď máme veľký počet strán, ktoré by mali obrovské náklady na synchronizáciu. V takýchto situáciách sa vyskytujú prípady: Phasery je možné nastaviť tak, aby skupiny subfázových zariadení zdieľali spoločného rodiča.

Ďalej začnime tri LongRunningAction akčné vlákna, ktoré budú čakať na bariére, kým nezavoláme comeAndAwaitAdvance () metóda z hlavného vlákna.

Majte na pamäti, že sme inicializovali naše Phaser s 1 a zavolal Registrovať() ešte trikrát. Teraz tri akčné vlákna oznámili, že prišli k bariére, takže ešte jedna výzva comeAndAwaitAdvance () je potrebný - ten z hlavného vlákna:

executorService.submit (nový LongRunningAction ("thread-1", ph)); executorService.submit (nový LongRunningAction ("thread-2", ph)); executorService.submit (nový LongRunningAction ("thread-3", ph)); ph.arriveAndAwaitAdvance (); assertEquals (1, ph.getPhase ());

Po ukončení tejto fázy getPhase () metóda vráti jednu, pretože program dokončil spracovanie prvého kroku vykonania.

Povedzme, že ďalšiu fázu spracovania by mali viesť dve vlákna. Môžeme využiť Phaser aby sme to dosiahli, pretože nám to umožňuje dynamicky konfigurovať počet vlákien, ktoré by mali čakať na bariéru. Začíname používať dve nové vlákna, ktoré sa však budú spúšťať až po zavolaní súboru comeAndAwaitAdvance () z hlavného vlákna (rovnaké ako v predchádzajúcom prípade):

executorService.submit (nový LongRunningAction ("thread-4", ph)); executorService.submit (nový LongRunningAction ("thread-5", ph)); ph.arriveAndAwaitAdvance (); assertEquals (2, ph.getPhase ()); ph.arriveAndDeregister ();

Po tomto getPhase () metóda vráti číslo fázy rovné dvom. Ak chceme dokončiť náš program, musíme zavolať na comeAndDeregister () metóda ako hlavné vlákno je stále zaregistrovaná v Phaser. Ak zrušenie registrácie spôsobí, že počet registrovaných strán bude nulový, použije sa hodnota Phaser je ukončený. Všetky volania na metódy synchronizácie sa už nebudú blokovať a okamžite sa vrátia.

Spustenie programu vyprodukuje nasledujúci výstup (celý zdrojový kód s príkazmi tlačového riadku nájdete v úložisku kódov):

Toto je fáza 0 To je fáza 0 To je fáza 0 Niť-vlákno-2 pred dlhodobým pôsobením Niťové-vlákno-1 pred dlhým bežiacim účinkom Niťové-vlákno-3 pred dlhodobým pôsobením To je fáza 1 Toto je fáza 1 Niťové vlákno-4 pred dlhým behom prebiehajúca akcia Vlákno-vlákno 5 pred dlhotrvajúcou akciou

Vidíme, že všetky vlákna čakajú na vykonanie, kým sa bariéra neotvorí. Ďalšia fáza vykonania sa uskutoční až po úspešnom dokončení predchádzajúcej.

4. Záver

V tomto tutoriáli sme sa pozreli na Phaser konštruovať z java.util.concurrent a implementovali sme koordinačnú logiku s viacerými fázami pomocou Phaser trieda.

Implementáciu všetkých týchto príkladov a útržkov kódu nájdete v projekte GitHub - jedná sa o projekt Maven, takže by malo byť ľahké ho importovať a spustiť tak, ako je.