streamalism.org

1. Oversikt

I denne artikkelen vil vi se på en av de mest nyttige konstruksjonene java.util.concurrent for å løse det samtidige produsent-forbrukerproblemet. Vi ser på et API av BlockingQueue grensesnitt og hvordan metoder fra det grensesnittet gjør det lettere å skrive samtidige programmer.

Senere i artikkelen vil vi vise et eksempel på et enkelt program som har flere produsenttråder og flere forbrukertråder.

2. BlockingQueue Typer

Vi kan skille mellom to typer BlockingQueue:

• ubegrenset kø - kan vokse nesten på ubestemt tid
• avgrenset kø - med maksimal kapasitet definert

2.1. Ubundet kø

Å lage ubegrensede køer er enkelt:

BlockingQueue blockingQueue = ny LinkedBlockingDeque ();

Kapasiteten til blockingQueue vil bli satt til Heltall.MAX_VALUE. Alle operasjoner som legger til et element i den ubegrensede køen, vil aldri blokkere, og dermed kan den vokse til en veldig stor størrelse.

Det viktigste når du designer et produsent-forbrukerprogram ved hjelp av ubegrenset BlockingQueue er at forbrukerne skal kunne konsumere meldinger så raskt som produsenter legger til meldinger i køen. Ellers kan minnet fylles opp, og vi får et Tomt for minne unntak.

2.2. Avgrenset kø

Den andre typen køer er den avgrensede køen. Vi kan opprette slike køer ved å overføre kapasiteten som et argument til en konstruktør:

BlockingQueue blockingQueue = ny LinkedBlockingDeque (10);

Her har vi en blockingQueue som har en kapasitet lik 10. Det betyr at når en produsent prøver å legge til et element i en allerede full kø, avhengig av en metode som ble brukt til å legge det til (by på(), legge til() eller sette()), vil den blokkere til det blir ledig plass til å sette inn objekt. Ellers vil operasjonene mislykkes.

Å bruke avgrenset kø er en god måte å designe samtidige programmer på, fordi når vi setter inn et element i en allerede full kø, må operasjonen vente til forbrukerne tar igjen og gjør litt plass tilgjengelig i køen. Det gir oss struping uten noen anstrengelse fra vår side.

3. BlockingQueue API

Det er to typer metoder i BlockingQueue grensesnitt – metoder som er ansvarlige for å legge til elementer i en kø og metoder som henter disse elementene. Hver metode fra de to gruppene oppfører seg forskjellig i tilfelle køen er full / tom.

3.1. Legge til elementer

• legge til() - returnerer ekte hvis innsetting var vellykket, ellers kaster en IllegalStateException
• sette() - setter inn det angitte elementet i en kø, og venter på et gratis spor hvis det er nødvendig
• by på() - returnerer ekte hvis innsetting var vellykket, ellers falsk
• tilbud (E e, lang tidsavbrudd, TimeUnit-enhet) - prøver å sette inn element i en kø og venter på et tilgjengelig spor innen en spesifisert tidsavbrudd

3.2. Henter elementer

• ta() - venter på et hodeelement i en kø og fjerner det. Hvis køen er tom, blokkeres den og venter på at et element blir tilgjengelig
• meningsmåling (lang tidsavbrudd, TimeUnit-enhet) - henter og fjerner hodet på køen, og venter opp til den angitte ventetiden om nødvendig for at et element blir tilgjengelig. Returnerer null etter en timeout

Disse metodene er de viktigste byggesteinene fra BlockingQueue grensesnitt når man bygger produsent-forbrukerprogrammer.

4. Flertrådet produsent-forbrukereksempel

La oss lage et program som består av to deler - en produsent og en forbruker.

Produsenten vil produsere et tilfeldig tall fra 0 til 100 og vil sette nummeret i et BlockingQueue. Vi har fire produsenttråder og bruker sette() metode for å blokkere til det er ledig plass i køen.

Det viktige å huske er at vi må stoppe forbrukertrådene våre fra å vente på at et element skal vises i kø på ubestemt tid.

En god teknikk for å signalisere fra produsent til forbruker at det ikke er flere meldinger å behandle, er å sende en spesiell melding kalt en giftpille. Vi må sende så mange giftpiller som vi har forbrukere. Så når en forbruker tar den spesielle giftpillemeldingen fra en kø, vil den fullføre kjennelsen.

La oss se på en produsentklasse:

offentlig klasse NumbersProducer implementerer Runnable {private BlockingQueue numbersQueue; privat sluttgiftPill; privat endelig int poisonPillPerProducer; public NumbersProducer (BlockingQueue numbersQueue, int poisonPill, int poisonPillPerProducer) {this.numbersQueue = numbersQueue; this.poisonPill = poisonPill; this.poisonPillPerProducer = poisonPillPerProducer; } public void run () {try {genererNumre (); } fange (InterruptedException e) {Thread.currentThread (). interrupt (); }} privat tomrom generereNumre () kaster InterruptedException {for (int i = 0; i <100; i ++) {numbersQueue.put (ThreadLocalRandom.current (). nextInt (100)); } for (int j = 0; j <poisonPillPerProducer; j ++) {numbersQueue.put (poisonPill); }}}

Vår produsentkonstruktør tar som argument BlockingQueue som brukes til å koordinere behandlingen mellom produsent og forbruker. Vi ser den metoden generereNumre () vil sette 100 elementer i kø. Det krever også giftpillemelding for å vite hvilken type melding som må settes i kø når henrettelsen skal være ferdig. Den meldingen må settes poisonPillPerProducer ganger inn i en kø.

Hver forbruker vil ta et element fra en BlockingQueue ved hjelp av ta() metoden slik at den blokkeres til det er et element i en kø. Etter å ha tatt en Heltall Fra en kø sjekker det om meldingen er en giftpille, hvis ja, er kjøringen av en tråd ferdig. Ellers vil det skrive ut resultatet på standardutdata sammen med gjeldende trådnavn.

Dette vil gi oss innsikt i det indre arbeidet til forbrukerne våre:

offentlig klasse NumbersConsumer implementerer Runnable {private BlockingQueue kø; privat sluttgiftPill; public NumbersConsumer (BlockingQueue kø, int poisonPill) {this.queue = kø; this.poisonPill = poisonPill; } public void run () {try {while (true) {Integer number = queue.take (); if (number.equals (poisonPill)) {return; } System.out.println (Thread.currentThread (). GetName () + "resultat:" + nummer); }} fange (InterruptedException e) {Thread.currentThread (). interrupt (); }}}

Det viktige å legge merke til er bruken av en kø. Samme som i produsentkonstruktøren, blir en kø sendt som et argument. Vi kan gjøre det fordi BlockingQueue kan deles mellom tråder uten eksplisitt synkronisering.

Nå som vi har produsent og forbruker, kan vi starte vårt program. Vi må definere køens kapasitet, og vi setter den til 100 elementer.

Vi ønsker å ha 4 produsenttråder, og et antall forbrukertråder vil være lik antall tilgjengelige prosessorer:

int BUNDET = 10; int N_PRODUCERS = 4; int N_CONSUMERS = Runtime.getRuntime (). availableProcessors (); int poisonPill = Heltall.MAX_VALUE; int poisonPillPerProducer = N_CONSUMERS / N_PRODUCERS; int mod = N_CONSUMERS% N_PRODUCERS; BlockingQueue queue = new LinkedBlockingQueue (BOUND); for (int i = 1; i <N_PRODUCERS; i ++) {ny tråd (ny NumbersProducer (kø, poisonPill, poisonPillPerProducer)). start (); } for (int j = 0; j <N_CONSUMERS; j ++) {new Thread (new NumbersConsumer (queue, poisonPill)). start (); } ny tråd (ny NumbersProducer (kø, poisonPill, poisonPillPerProducer + mod)). start (); 

BlockingQueue er opprettet ved hjelp av konstruksjon med kapasitet. Vi skaper fire produsenter og N-forbrukere. Vi spesifiserer at giftpillemeldingen skal være en Heltall.MAX_VALUE fordi en slik verdi aldri vil bli sendt av produsenten vår under normale arbeidsforhold. Det viktigste å legge merke til her er at BlockingQueue brukes til å koordinere arbeidet mellom dem.

Når vi kjører programmet, vil 4 produsenttråder være tilfeldige Heltall i en BlockingQueue og forbrukere vil ta disse elementene fra køen. Hver tråd skriver ut til standardutdata navnet på tråden sammen med et resultat.

5. Konklusjon

Denne artikkelen viser en praktisk bruk av BlockingQueue og forklarer metoder som brukes til å legge til og hente elementer fra den. Vi har også vist hvordan du bygger et flertrådet produsent-forbrukerprogram ved hjelp av BlockingQueue å koordinere arbeidet mellom produsenter og forbrukere.

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