streamalism.org

1. Oversikt

I denne raske opplæringen vil vi utforske det grunnleggende om semaforer og mutexer i Java.

2. Semafor

Vi begynner med java.util.concurrent.Semaphore. Vi kan bruke semaforer for å begrense antall samtidige tråder som får tilgang til en bestemt ressurs.

I det følgende eksemplet vil vi implementere en enkel påloggingskø for å begrense antall brukere i systemet:

klasse LoginQueueUsingSemaphore {private Semaphore semaphore; offentlig LoginQueueUsingSemaphore (int slotLimit) {semaphore = new Semaphore (slotLimit); } boolsk tryLogin () {return semaphore.tryAcquire (); } ugyldig avlogging () {semaphore.release (); } int availableSlots () {return semaphore.availablePermits (); }}

Legg merke til hvordan vi brukte følgende metoder:

• tryAcquire () - returner sann hvis en tillatelse er tilgjengelig umiddelbart og skaffe den ellers returnere falsk, men tilegne() får tillatelse og sperring til en er tilgjengelig
• frigjør () - frigjør tillatelse
• availablePermits () - retur antall nåværende tilgjengelige tillatelser

For å teste påloggingskøen vår, vil vi først prøve å nå grensen og sjekke om neste påloggingsforsøk blir blokkert:

@Test offentlig ugyldig givenLoginQueue_whenReachLimit_thenBlocked () {int slots = 10; ExecutorService executorService = Executors.newFixedThreadPool (slots); LoginQueueUsingSemaphore loginQueue = ny LoginQueueUsingSemaphore (slots); IntStream.range (0, slots) .forEach (user -> executorService.execute (loginQueue :: tryLogin)); executorService.shutdown (); assertEquals (0, loginQueue.availableSlots ()); assertFalse (loginQueue.tryLogin ()); }

Deretter vil vi se om noen spor er tilgjengelige etter en avlogging:

@Test offentlig ugyldig givenLoginQueue_whenLogout_thenSlotsAvailable () {int slots = 10; ExecutorService executorService = Executors.newFixedThreadPool (slots); LoginQueueUsingSemaphore loginQueue = ny LoginQueueUsingSemaphore (slots); IntStream.range (0, slots) .forEach (user -> executorService.execute (loginQueue :: tryLogin)); executorService.shutdown (); assertEquals (0, loginQueue.availableSlots ()); loginQueue.logout (); assertTrue (loginQueue.availableSlots ()> 0); assertTrue (loginQueue.tryLogin ()); }

3. Tidsbestemt Semafor

Deretter vil vi diskutere Apache Commons TimedSemaphore. TimedSemaphore tillater et antall tillatelser som en enkel semafor, men i en gitt tidsperiode, etter denne perioden tilbakestilles tiden og alle tillatelser frigjøres.

Vi kan bruke TimedSemaphore å bygge en enkel forsinkelseskø som følger:

class DelayQueueUsingTimedSemaphore {private TimedSemaphore semaphore; DelayQueueUsingTimedSemaphore (long period, int slotLimit) {semaphore = new TimedSemaphore (period, TimeUnit.SECONDS, slotLimit); } boolsk tryAdd () {return semaphore.tryAcquire (); } int availableSlots () {return semaphore.getAvailablePermits (); }}

Når vi bruker en forsinkelseskø med ett sekund som tidsperiode og etter å ha brukt alle sporene innen ett sekund, bør ingen være tilgjengelige:

offentlig ugyldig givenDelayQueue_whenReachLimit_thenBlocked () {int slots = 50; ExecutorService executorService = Executors.newFixedThreadPool (slots); DelayQueueUsingTimedSemaphore delayQueue = new DelayQueueUsingTimedSemaphore (1, slots); IntStream.range (0, slots) .forEach (user -> executorService.execute (delayQueue :: tryAdd)); executorService.shutdown (); assertEquals (0, delayQueue.availableSlots ()); assertFalse (delayQueue.tryAdd ()); }

Men etter å ha sovet en stund, semaforen skal tilbakestilles og frigjøre tillatelsene:

@Test offentlig ugyldighet givenDelayQueue_whenTimePass_thenSlotsAvailable () kaster InterruptedException {int slots = 50; ExecutorService executorService = Executors.newFixedThreadPool (slots); DelayQueueUsingTimedSemaphore delayQueue = new DelayQueueUsingTimedSemaphore (1, slots); IntStream.range (0, slots) .forEach (user -> executorService.execute (delayQueue :: tryAdd)); executorService.shutdown (); assertEquals (0, delayQueue.availableSlots ()); Tråd. Søvn (1000); assertTrue (delayQueue.availableSlots ()> 0); assertTrue (delayQueue.tryAdd ()); }

4. Semaphore vs. Mutex

Mutex fungerer på samme måte som en binær semafor, vi kan bruke den til å implementere gjensidig ekskludering.

I det følgende eksemplet bruker vi en enkel binær semafor til å bygge en teller:

klasse CounterUsingMutex {privat Semaphore mutex; privat int count; CounterUsingMutex () {mutex = new Semaphore (1); telle = 0; } ugyldig økning () kaster InterruptedException {mutex.acquire (); this.count = this.count + 1; Tråd. Søvn (1000); mutex.release (); } int getCount () {returner this.count; } boolsk hasQueuedThreads () {return mutex.hasQueuedThreads (); }}

Når mange tråder prøver å få tilgang til disken på en gang, de blir ganske enkelt blokkert i kø:

@Test offentlig ugyldig nårMutexAndMultipleThreads_thenBlocked () kaster InterruptedException {int count = 5; ExecutorService executorService = Executors.newFixedThreadPool (count); CounterUsingMutex counter = new CounterUsingMutex (); IntStream.range (0, count) .forEach (user -> executorService.execute (() -> {try {counter.increase ();} catch (InterruptedException e) {e.printStackTrace ();}})); executorService.shutdown (); assertTrue (counter.hasQueuedThreads ()); }

Når vi venter, vil alle tråder få tilgang til telleren og ingen tråder igjen i køen:

@Test offentlig ugyldig givenMutexAndMultipleThreads_ThenDelay_thenCorrectCount () kaster InterruptedException {int count = 5; ExecutorService executorService = Executors.newFixedThreadPool (count); CounterUsingMutex teller = ny CounterUsingMutex (); IntStream.range (0, count) .forEach (user -> executorService.execute (() -> {try {counter.increase ();} catch (InterruptedException e) {e.printStackTrace ();}})); executorService.shutdown (); assertTrue (counter.hasQueuedThreads ()); Tråd. Søvn (5000); assertFalse (counter.hasQueuedThreads ()); assertEquals (count, counter.getCount ()); }

5. Konklusjon

I denne artikkelen utforsket vi det grunnleggende om semaforer i Java.

Som alltid er hele kildekoden tilgjengelig på GitHub.