streamalism.org

1. Introduksjon

I denne artikkelen vil vi gi en guide til CountDownLatch klasse og demonstrere hvordan den kan brukes i noen få praktiske eksempler.

I hovedsak ved å bruke en CountDownLatch vi kan få en tråd til å blokkere til andre tråder har fullført en gitt oppgave.

2. Bruk i samtidig programmering

Enkelt sagt, a CountDownLatch har en disk felt, som du kan redusere etter behov. Vi kan deretter bruke den til å blokkere en ringetråd til den er telt ned til null.

Hvis vi gjorde noen parallell behandling, kunne vi sette i gang CountDownLatch med samme verdi for telleren som et antall tråder vi vil jobbe på tvers. Så kunne vi bare ringe nedtelling () etter at hver tråd er ferdig, garanterer at en avhengig tråd ringer avvente() vil blokkere til arbeidertrådene er ferdige.

3. Venter på at en gruppe tråder skal fullføres

La oss prøve ut dette mønsteret ved å lage et Arbeider og bruke en CountDownLatch felt for å signalisere når det er fullført:

offentlig klasse Arbeidere implementerer Runnable {private List outputScraper; private CountDownLatch countDownLatch; offentlig arbeider (List outputScraper, CountDownLatch countDownLatch) {this.outputScraper = outputScraper; this.countDownLatch = countDownLatch; } @ Override public void run () {doSomeWork (); outputScraper.add ("Counted down"); countDownLatch.countDown (); }}

La oss deretter lage en test for å bevise at vi kan få en CountDownLatch å vente på Arbeider tilfeller å fullføre:

@Test offentlig ugyldig nårParallelProcessing_thenMainThreadWillBlockUntilCompletion () kaster InterruptedException {List outputScraper = Collections.synchronizedList (new ArrayList ()); CountDownLatch countDownLatch = ny CountDownLatch (5); Listearbeidere = Stream .generate (() -> ny tråd (ny arbeider (outputScraper, countDownLatch))) .grense (5) .collect (toList ()); workers.forEach (tråd :: start); countDownLatch.await (); outputScraper.add ("Sperre frigitt"); assertThat (outputScraper) .containsExactly ("Counted down", "Counted down", "Counted down", "Counted down", "Counted down", "Latch released"); }

Naturligvis vil "Latch released" alltid være den siste utgangen - ettersom den er avhengig av CountDownLatch slippe.

Merk at hvis vi ikke ringte avvente(), ville vi ikke være i stand til å garantere bestillingen av utførelsen av trådene, så testen mislyktes tilfeldig.

4. Et basseng med tråder som venter på å begynne

Hvis vi tok det forrige eksemplet, men denne gangen startet tusenvis av tråder i stedet for fem, er det sannsynlig at mange av de tidligere vil være ferdigbehandlet før vi til og med har ringt start() på de senere. Dette kan gjøre det vanskelig å prøve å reprodusere et samtidighetsproblem, da vi ikke vil kunne få alle trådene våre til å kjøre parallelt.

For å komme rundt dette, la oss få CountdownLatch å jobbe annerledes enn i forrige eksempel. I stedet for å blokkere en overordnet tråd til noen barnetråder er ferdige, kan vi blokkere hver barnetråd til alle de andre har startet.

La oss endre vår løpe() metode slik at den blokkeres før behandlingen:

offentlig klasse WaitingWorker implementerer Runnable {private List outputScraper; private CountDownLatch readyThreadCounter; privat CountDownLatch callingThreadBlocker; private CountDownLatch completedThreadCounter; public WaitingWorker (List outputScraper, CountDownLatch readyThreadCounter, CountDownLatch callingThreadBlocker, CountDownLatch completeThreadCounter) {this.outputScraper = outputScraper; this.readyThreadCounter = readyThreadCounter; this.callingThreadBlocker = callingThreadBlocker; this.completedThreadCounter = fullførtThreadCounter; } @ Override public void run () {readyThreadCounter.countDown (); prøv {callingThreadBlocker.await (); doSomeWork (); outputScraper.add ("Counted down"); } fange (InterruptedException e) {e.printStackTrace (); } til slutt {completeThreadCounter.countDown (); }}}

La oss nå endre testen vår slik at den blokkerer til alle Arbeidere har startet, opphever blokkeringen av Arbeidere, og blokkerer til Arbeidere er ferdig:

@Test offentlig ugyldig nårDoingLotsOfThreadsInParallel_thenStartThemAtTheSameTime () kaster InterruptedException {List outputScraper = Collections.synchronizedList (new ArrayList ()); CountDownLatch readyThreadCounter = ny CountDownLatch (5); CountDownLatch callingThreadBlocker = ny CountDownLatch (1); CountDownLatch completedThreadCounter = ny CountDownLatch (5); Listearbeidere = Stream .generate (() -> ny tråd (ny WaitingWorker (outputScraper, readyThreadCounter, callingThreadBlocker, fullførtThreadCounter))) .limit (5) .collect (toList ()); workers.forEach (tråd :: start); readyThreadCounter.await (); outputScraper.add ("Arbeidere klare"); callingThreadBlocker.countDown (); completeThreadCounter.await (); outputScraper.add ("Workers complete"); assertThat (outputScraper) .containsExactly ("Workers ready", "Counted down", "Counted down", "Counted down", "Counted down", "Counted down", "Workers complete"); }

Dette mønsteret er veldig nyttig for å prøve å reprodusere samtidige feil, som kan brukes til å tvinge tusenvis av tråder til å prøve å utføre litt logikk parallelt.

5. Avslutning a CountdownLatch Tidlig

Noen ganger kan vi komme i en situasjon der Arbeidere avslutte ved en feil før du teller ned CountDownLatch. Dette kan føre til at det aldri når null og avvente() avslutter aldri:

@ Override public void run () {if (true) {throw new RuntimeException ("Å kjære, jeg er en BrokenWorker"); } countDownLatch.countDown (); outputScraper.add ("Counted down"); }

La oss endre vår tidligere test for å bruke en BrokenWorker, for å vise hvordan avvente() vil blokkere for alltid:

@Test offentlig ugyldig nårFailingToParallelProcess_thenMainThreadShouldGetNotGetStuck () kaster InterruptedException {List outputScraper = Collections.synchronizedList (new ArrayList ()); CountDownLatch countDownLatch = ny CountDownLatch (5); List workers = Stream .generate (() -> new Thread (new BrokenWorker (outputScraper, countDownLatch))) .limit (5) .collect (toList ()); workers.forEach (tråd :: start); countDownLatch.await (); }

Det er klart at dette ikke er oppførselen vi ønsker - det ville være mye bedre for applikasjonen å fortsette enn uendelig blokk.

For å komme deg rundt dette, la oss legge til et tidsavbruddsargument i kallet vårt avvente().

boolsk fullført = countDownLatch.await (3L, TimeUnit.SECONDS); assertThat (fullført) .isFalse ();

Som vi kan se, vil testen til slutt time out og avvente() vil returnere falsk.

6. Konklusjon

I denne hurtigveiledningen har vi demonstrert hvordan vi kan bruke en CountDownLatch for å blokkere en tråd til andre tråder er ferdigbehandlet.

Vi har også vist hvordan det kan brukes til å feilsøke problemer med samtidighet ved å sørge for at tråder kjører parallelt.

Implementeringen av disse eksemplene finner du på GitHub; dette er et Maven-basert prosjekt, så det skal være enkelt å kjøre som det er.