streamalism.org

1. Oversigt

I denne artikel vil vi se på Phaser konstruere fra java.util.concurrent pakke. Det er en meget lignende konstruktion til CountDownLatch der giver os mulighed for at koordinere udførelsen af ​​tråde. I sammenligning med CountDownLatch, det har nogle ekstra funktioner.

Det Phaser er en barriere, hvor det dynamiske antal tråde skal vente, før de fortsætter udførelsen. I CountDownLatch dette nummer kan ikke konfigureres dynamisk og skal leveres, når vi opretter forekomsten.

2. Phaser API

Det Phaser giver os mulighed for at opbygge logik, hvor tråde skal vente på barrieren, inden de går til næste trin i udførelsen.

Vi kan koordinere flere udførelsesfaser ved at genbruge en Phaser eksempel for hver programfase. Hver fase kan have et forskelligt antal tråde, der venter på at gå videre til en anden fase. Vi ser på et eksempel på brug af faser senere.

For at deltage i koordineringen skal tråden gøre det Tilmeld() sig selv med Phaser eksempel. Bemærk, at dette kun øger antallet af registrerede parter, og vi kan ikke kontrollere, om den aktuelle tråd er registreret - vi bliver nødt til at underklasse implementeringen for at understøtte dette.

Tråden signaliserer, at den ankom til barrieren ved at ringe til arriveAndAwaitAdvance (), som er en blokeringsmetode. Når antallet af ankomne parter er lig med antallet af registrerede parter, fortsætter gennemførelsen af ​​programmet, og fasetallet stiger. Vi kan få det aktuelle fasetal ved at ringe til getPhase () metode.

Når tråden er færdig med sit job, skal vi ringe til arriveAndDeregister () metode til at signalere, at den aktuelle tråd ikke længere skal redegøres for i denne særlige fase.

3. Implementering af logik ved hjælp af Phaser API

Lad os sige, at vi ønsker at koordinere flere faser af handlinger. Tre tråde behandler den første fase, og to tråde behandler den anden fase.

Vi opretter en LongRunningAction klasse, der implementerer Kan køres grænseflade:

klasse LongRunningAction implementerer Runnable {private String threadName; privat Phaser ph; LongRunningAction (String threadName, Phaser ph) {this.threadName = threadName; this.ph = ph; ph.register (); } @ Override public void run () {ph.arriveAndAwaitAdvance (); prøv {Thread.sleep (20); } fange (InterruptedException e) {e.printStackTrace (); } ph.arriveAndDeregister (); }}

Når vores aktionsklasse er instantieret, tilmelder vi os til Phaser eksempel ved hjælp af Tilmeld() metode. Dette øger antallet af tråde ved hjælp af det specifikke Phaser.

Opkaldet til arriveAndAwaitAdvance () får den aktuelle tråd til at vente på barrieren. Når antallet af ankomne parter bliver det samme som antallet af registrerede parter, fortsætter udførelsen, som allerede nævnt.

Når behandlingen er udført, afregistrerer den aktuelle tråd sig selv ved at ringe til arriveAndDeregister () metode.

Lad os oprette en test case, hvor vi starter tre LongRunningAction tråde og blok på barrieren. Derefter opretter vi to yderligere, når handlingen er afsluttet LongRunningAction tråde, der udfører behandling af den næste fase.

Når du opretter Phaser eksempel fra hovedtråden, vi passerer 1 som et argument. Dette svarer til at kalde Tilmeld() metode fra den aktuelle tråd. Vi gør dette, fordi hovedtråden, når vi opretter tre arbejdstråde, er en koordinator og derfor Phaser skal have fire tråde registreret:

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

Fasen efter initialiseringen er lig med nul.

Det Phaser klasse har en konstruktør, hvor vi kan videregive en overordnet instans til den. Det er nyttigt i tilfælde, hvor vi har et stort antal parter, der vil opleve massive synkroniseringsomkostningsomkostninger. I sådanne situationer, tilfælde af Fasere kan indstilles, så grupper af underfaser deler en fælles forælder.

Lad os derefter starte tre LongRunningAction handlingstråde, som venter på barrieren, indtil vi kalder arriveAndAwaitAdvance () metode fra hovedtråden.

Husk, at vi har initialiseret vores Phaser med 1 og kaldte Tilmeld() tre gange mere. Nu har tre handlingstråde meddelt, at de er ankommet til barrieren, så endnu et opkald af arriveAndAwaitAdvance () er nødvendig - den fra hovedtråden:

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

Efter afslutningen af ​​denne fase er getPhase () metode returnerer en, fordi programmet er færdigbehandlet det første trin i udførelsen.

Lad os sige, at to tråde skal gennemføre den næste fase af behandlingen. Vi kan udnytte Phaser for at opnå det, fordi det giver os mulighed for dynamisk at konfigurere antallet af tråde, der skal vente på barrieren. Vi starter to nye tråde, men disse vil ikke fortsætte med at udføre, før opkaldet til arriveAndAwaitAdvance () fra hovedtråden (samme som i det foregående tilfælde):

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

Efter dette, getPhase () metode returnerer fasetallet lig med to. Når vi vil afslutte vores program, skal vi ringe til arriveAndDeregister () metode, da hovedtråden stadig er registreret i Phaser. Når afmeldingen får antallet af registrerede parter til at blive nul, Phaser er afsluttet. Alle opkald til synkroniseringsmetoder blokerer ikke længere og vender straks tilbage.

Kørsel af programmet vil producere følgende output (fuld kildekode med udskrivningslinjens udsagn kan findes i kodelageret):

Dette er fase 0 Dette er fase 0 Dette er fase 0 Trådgevind-2 før langvarig handling Trådgevind-1 før langvarig handling Trådgevind-3 før langvarig handling Dette er fase 1 Dette er fase 1 Trådgevind-4 inden længe kørende handling Tråd tråd-5 inden lang kørsel

Vi ser, at alle tråde venter på udførelse, indtil barrieren åbner. Den næste fase af udførelsen udføres kun, når den forrige er færdig.

4. Konklusion

I denne vejledning kiggede vi på Phaser konstruere fra java.util.concurrent og vi implementerede koordinationslogikken med flere faser ved hjælp af Phaser klasse.

Implementeringen af ​​alle disse eksempler og kodestykker findes i GitHub-projektet - dette er et Maven-projekt, så det skal være let at importere og køre, som det er.