streamalism.org

1. Oversikt

I denne artikkelen vil vi se på Phaser konstruere fra java.util.concurrent pakke. Det er en veldig lik konstruksjon til CountDownLatch som lar oss koordinere utførelse av tråder. I forhold til CountDownLatch, den har litt ekstra funksjonalitet.

De Phaser er en barriere som det dynamiske antall tråder må vente på før du fortsetter kjøringen. I CountDownLatch det tallet kan ikke konfigureres dynamisk og må oppgis når vi oppretter forekomsten.

2. Phaser API

De Phaser lar oss bygge logikk der tråder må vente på barrieren før de går til neste trinn for utførelse.

Vi kan koordinere flere utførelsesfaser ved å gjenbruke a Phaser forekomst for hver programfase. Hver fase kan ha et annet antall tråder som venter på å gå videre til en annen fase. Vi ser på et eksempel på bruk av faser senere.

For å delta i koordineringen, trenger tråden det registrere() seg selv med Phaser forekomst. Merk at dette bare øker antall registrerte parter, og vi kan ikke sjekke om den gjeldende tråden er registrert - vi må underklasse implementeringen for å støtte dette.

Tråden signaliserer at den ankom barrieren ved å ringe arriveAndAwaitAdvance (), som er en blokkeringsmetode. Når antall ankomne parter er lik antall registrerte parter, vil gjennomføringen av programmet fortsette, og fasetallet vil øke. Vi kan få gjeldende fasetall ved å ringe getPhase () metode.

Når tråden er ferdig med jobben, bør vi ringe arriveAndDeregister () metode for å signalisere at den gjeldende tråden ikke lenger skal redegjøres for i denne spesielle fasen.

3. Implementering av logikk ved hjelp av Phaser API

La oss si at vi ønsker å koordinere flere faser av handlinger. Tre tråder vil behandle den første fasen, og to tråder vil behandle den andre fasen.

Vi lager en LongRunningAction klasse som implementerer Kjørbar grensesnitt:

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 actionklassen vår er instansert, registrerer vi oss på Phaser eksempel bruker registrere() metode. Dette øker antall tråder som bruker den spesifikke Phaser.

Samtalen til arriveAndAwaitAdvance () vil føre til at den gjeldende tråden venter på barrieren. Når antallet ankomne parter blir det samme som antall registrerte parter, vil gjennomføringen fortsette.

Etter at behandlingen er ferdig, avregistrerer den gjeldende tråden seg selv ved å ringe arriveAndDeregister () metode.

La oss lage en prøvesak der vi starter tre LongRunningAction tråder og blokker på barrieren. Deretter, etter at handlingen er ferdig, oppretter vi to ekstra LongRunningAction tråder som vil utføre behandling av neste fase.

Når du lager Phaser eksempel fra hovedtråden, vi passerer 1 som argument. Dette tilsvarer å kalle registrere() metode fra gjeldende tråd. Vi gjør dette fordi når vi oppretter tre arbeidertråder, er hovedtråden en koordinator, og derfor Phaser må ha fire tråder registrert:

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

Fasen etter initialiseringen er lik null.

De Phaser klasse har en konstruktør der vi kan overføre en foreldreinstans til den. Det er nyttig i tilfeller der vi har et stort antall partier som vil oppleve enorme påstandskostnader for synkronisering. I slike situasjoner, forekomster av Fasere kan settes opp slik at grupper av underfaser deler en felles forelder.

La oss starte tre LongRunningAction handlingstråder, som vil vente på barrieren til vi vil kalle arriveAndAwaitAdvance () metode fra hovedtråden.

Husk at vi har initialisert Phaser med 1 og ringte registrere() tre ganger til. Nå har tre handlingstråder kunngjort at de har ankommet barrieren, så en samtale til arriveAndAwaitAdvance () er nødvendig - den fra hovedtråden:

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

Etter at denne fasen er fullført, vil getPhase () metoden vil returnere en fordi programmet er ferdig med å behandle det første trinnet i utførelsen.

La oss si at to tråder skal gjennomføre neste behandlingsfase. Vi kan utnytte Phaser for å oppnå det fordi det lar oss konfigurere dynamisk antall tråder som skal vente på barrieren. Vi starter to nye tråder, men disse vil ikke fortsette å utføre før samtalen til arriveAndAwaitAdvance () fra hovedtråden (samme som i forrige tilfelle):

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

Etter dette, getPhase () metoden vil returnere fasetall lik to. Når vi vil fullføre programmet vårt, må vi ringe arriveAndDeregister () metoden som hovedtråd er fortsatt registrert i Phaser. Når avregistreringen fører til at antallet registrerte parter blir null, blir Phaser er avsluttet. Alle anrop til synkroniseringsmetoder blokkerer ikke lenger og kommer tilbake umiddelbart.

Å kjøre programmet vil gi følgende utdata (full kildekode med utskriftslinjene finner du i kodelageret):

Dette er fase 0 Dette er fase 0 Dette er fase 0 Gjengetråd-2 før langvarig handling Trådtråd-1 før langgående handling Trådgjenging-3 før langvarig handling Dette er fase 1 Dette er fase 1 Trådgjenger-4 før lenge kjører handling Tråd tråd-5 før lang kjører handling

Vi ser at alle tråder venter på henrettelse til barrieren åpnes. Neste fase av utførelsen utføres bare når den forrige ble fullført.

4. Konklusjon

I denne opplæringen så vi på Phaser konstruere fra java.util.concurrent og vi implementerte koordineringslogikken med flere faser ved hjelp av Phaser klasse.

Implementeringen av alle disse eksemplene og kodebitene finnes i GitHub-prosjektet - dette er et Maven-prosjekt, så det skal være enkelt å importere og kjøre som det er.