streamalism.org

1. Overzicht

In deze korte tutorial verkennen we de basisprincipes van semaforen en mutexen in Java.

2. Semafoor

We beginnen met java.util.concurrent.Semaphore. We kunnen semaforen gebruiken om het aantal gelijktijdige threads dat toegang heeft tot een specifieke bron te beperken.

In het volgende voorbeeld zullen we een eenvoudige inlogwachtrij implementeren om het aantal gebruikers in het systeem te beperken:

klasse LoginQueueUsingSemaphore {private Semaphore semafoor; openbare LoginQueueUsingSemaphore (int slotLimit) {semaphore = new Semaphore (slotLimit); } boolean tryLogin () {retourneer semafoor.tryAcquire (); } void logout () {semaphore.release (); } int availableSlots () {return semaphore.availablePermits (); }}

Merk op hoe we de volgende methoden hebben gebruikt:

• tryAcquire () - retourneer true als een vergunning onmiddellijk beschikbaar is en verkrijg deze anders retourneer false, maar verkrijgen() verwerft een vergunning en blokkeert totdat er een beschikbaar is
• release () - geef een vergunning vrij
• availablePermits () - retourneer het aantal huidige beschikbare vergunningen

Om onze inlogwachtrij te testen, zullen we eerst proberen de limiet te bereiken en controleren of de volgende inlogpoging wordt geblokkeerd:

@Test openbare leegte gegevenLoginQueue_whenReachLimit_thenBlocked () {int slots = 10; ExecutorService executorService = Executors.newFixedThreadPool (slots); LoginQueueUsingSemaphore loginQueue = nieuwe LoginQueueUsingSemaphore (slots); IntStream.range (0, slots) .forEach (gebruiker -> executorService.execute (loginQueue :: tryLogin)); executorService.shutdown (); assertEquals (0, loginQueue.availableSlots ()); assertFalse (loginQueue.tryLogin ()); }

Vervolgens zullen we zien of er slots beschikbaar zijn na uitloggen:

@Test openbare ongeldig gegevenLoginQueue_whenLogout_thenSlotsAvailable () {int slots = 10; ExecutorService executorService = Executors.newFixedThreadPool (slots); LoginQueueUsingSemaphore loginQueue = nieuwe LoginQueueUsingSemaphore (slots); IntStream.range (0, slots) .forEach (gebruiker -> executorService.execute (loginQueue :: tryLogin)); executorService.shutdown (); assertEquals (0, loginQueue.availableSlots ()); loginQueue.logout (); assertTrue (loginQueue.availableSlots ()> 0); assertTrue (loginQueue.tryLogin ()); }

3. Getimed Semafoor

Vervolgens zullen we Apache Commons bespreken GetimedSemaphore. GetimedSemaphore staat een aantal vergunningen toe als een eenvoudige semafoor, maar in een bepaalde periode, na deze periode wordt de tijd gereset en worden alle vergunningen vrijgegeven.

We kunnen gebruiken GetimedSemaphore om als volgt een eenvoudige vertragingswachtrij op te bouwen:

klasse DelayQueueUsingTimedSemaphore {privé TimedSemaphore semafoor; DelayQueueUsingTimedSemaphore (lange periode, int slotLimit) {semafoor = nieuwe TimedSemaphore (periode, TimeUnit.SECONDS, slotLimit); } boolean tryAdd () {retourneer semafoor.tryAcquire (); } int availableSlots () {return semaphore.getAvailablePermits (); }}

Wanneer we een vertragingswachtrij gebruiken met één seconde als tijdsperiode en nadat we alle slots binnen één seconde hebben gebruikt, zou er geen beschikbaar moeten zijn:

openbare leegte gegevenDelayQueue_whenReachLimit_thenBlocked () {int slots = 50; ExecutorService executorService = Executors.newFixedThreadPool (slots); DelayQueueUsingTimedSemaphore delayQueue = nieuwe DelayQueueUsingTimedSemaphore (1, slots); IntStream.range (0, slots) .forEach (gebruiker -> executorService.execute (delayQueue :: tryAdd)); executorService.shutdown (); assertEquals (0, delayQueue.availableSlots ()); assertFalse (delayQueue.tryAdd ()); }

Maar na een tijdje slapen, de seinpaal zou moeten resetten en de vergunningen moeten vrijgeven:

@Test openbare ongeldig gegevenDelayQueue_whenTimePass_thenSlotsAvailable () gooit InterruptedException {int slots = 50; ExecutorService executorService = Executors.newFixedThreadPool (slots); DelayQueueUsingTimedSemaphore delayQueue = nieuwe DelayQueueUsingTimedSemaphore (1, slots); IntStream.range (0, slots) .forEach (gebruiker -> executorService.execute (delayQueue :: tryAdd)); executorService.shutdown (); assertEquals (0, delayQueue.availableSlots ()); Thread.sleep (1000); assertTrue (delayQueue.availableSlots ()> 0); assertTrue (delayQueue.tryAdd ()); }

4. Semafoor versus Mutex

Mutex werkt op dezelfde manier als een binaire semafoor, we kunnen het gebruiken om wederzijdse uitsluiting te implementeren.

In het volgende voorbeeld gebruiken we een eenvoudige binaire semafoor om een ​​teller te bouwen:

klasse CounterUsingMutex {privé Semaphore mutex; privé int tellen; CounterUsingMutex () {mutex = nieuwe Semaphore (1); count = 0; } void Increase () gooit InterruptedException {mutex.acquire (); this.count = this.count + 1; Thread.sleep (1000); mutex.release (); } int getCount () {retourneer this.count; } boolean hasQueuedThreads () {terugkeer mutex.hasQueuedThreads (); }}

Als veel threads tegelijkertijd toegang proberen te krijgen tot de balie, ze worden gewoon geblokkeerd in een wachtrij:

@Test openbare leegte whenMutexAndMultipleThreads_thenBlocked () gooit InterruptedException {int count = 5; ExecutorService executorService = Executors.newFixedThreadPool (aantal); CounterUsingMutex counter = nieuwe CounterUsingMutex (); IntStream.range (0, aantal) .forEach (gebruiker -> executorService.execute (() -> {probeer {counter.increase ();} catch (InterruptedException e) {e.printStackTrace ();}})); executorService.shutdown (); assertTrue (counter.hasQueuedThreads ()); }

Als we wachten, krijgen alle threads toegang tot de teller en staan ​​er geen threads meer in de wachtrij:

@Test openbare leegte gegevenMutexAndMultipleThreads_ThenDelay_thenCorrectCount () gooit InterruptedException {int count = 5; ExecutorService executorService = Executors.newFixedThreadPool (aantal); CounterUsingMutex counter = nieuwe CounterUsingMutex (); IntStream.range (0, aantal) .forEach (gebruiker -> executorService.execute (() -> {probeer {counter.increase ();} catch (InterruptedException e) {e.printStackTrace ();}})); executorService.shutdown (); assertTrue (counter.hasQueuedThreads ()); Thread.sleep (5000); assertFalse (counter.hasQueuedThreads ()); assertEquals (count, counter.getCount ()); }

5. Conclusie

In dit artikel hebben we de basisprincipes van semaforen in Java onderzocht.

Zoals altijd is de volledige broncode beschikbaar op GitHub.