streamalism.org

1. Prehľad

V tomto rýchlom výučbe preskúmame základy semaforov a mutexov v Jave.

2. Semafor

Začneme s java.util.concurrent.Semaphore. Môžeme použiť semafory na obmedzenie počtu súbežných vlákien pristupujúcich k konkrétnemu prostriedku.

V nasledujúcom príklade implementujeme jednoduchý prihlasovací front na obmedzenie počtu používateľov v systéme:

trieda LoginQueueUsingSemaphore {súkromný semafor semafor; public LoginQueueUsingSemaphore (int slotLimit) {semafor = nový Semafor (slotLimit); } boolean tryLogin () {return semaphore.tryAcquire (); } void logout () {semaphore.release (); } int availableSlots () {return semaphore.availablePermits (); }}

Všimnite si, ako sme použili nasledujúce metódy:

• tryAcquire () - vrátiť true, ak je povolenie k dispozícii okamžite, a získať ho, inak vrátiť false, - získať () získa povolenie a blokovanie, kým nie je k dispozícii
• release () - uvoľniť povolenie
• availablePermits () - návratný počet aktuálnych povolení k dispozícii

Aby sme otestovali našu frontu na prihlásenie, najskôr sa pokúsime dosiahnuť limit a skontrolujeme, či bude blokovaný ďalší pokus o prihlásenie:

@Test public void givenLoginQueue_whenReachLimit_thenBlocked () {int sloty = 10; ExecutorService executorService = Executors.newFixedThreadPool (sloty); LoginQueueUsingSemaphore loginQueue = nový LoginQueueUsingSemaphore (sloty); IntStream.range (0, sloty). ForEach (užívateľ -> executorService.execute (loginQueue :: tryLogin)); executorService.shutdown (); assertEquals (0, loginQueue.availableSlots ()); assertFalse (loginQueue.tryLogin ()); }

Ďalej uvidíme, či sú po odhlásení k dispozícii nejaké sloty:

@Test public void givenLoginQueue_whenLogout_thenSlotsAvailable () {int sloty = 10; ExecutorService executorService = Executors.newFixedThreadPool (sloty); LoginQueueUsingSemaphore loginQueue = nový LoginQueueUsingSemaphore (sloty); IntStream.range (0, sloty). ForEach (užívateľ -> executorService.execute (loginQueue :: tryLogin)); executorService.shutdown (); assertEquals (0, loginQueue.availableSlots ()); loginQueue.logout (); assertTrue (loginQueue.availableSlots ()> 0); assertTrue (loginQueue.tryLogin ()); }

3. Časované Semafor

Ďalej si povieme niečo o Apache Commons TimedSemaphore. TimedSemaphore povoľuje niekoľko povolení ako jednoduchý semafor, ale v danom časovom období, po tomto období, dôjde k resetovaniu času a k uvoľneniu všetkých povolení.

Môžeme použiť TimedSemaphore vytvoriť jednoduchý front oneskorení nasledovne:

trieda DelayQueueUsingTimedSemaphore {private TimedSemaphore semafor; DelayQueueUsingTimedSemaphore (dlhé obdobie, int slotLimit) {semaphore = nový TimedSemaphore (obdobie, TimeUnit.SECONDS, slotLimit); } boolean tryAdd () {return semaphore.tryAcquire (); } int availableSlots () {return semaphore.getAvailablePermits (); }}

Keď použijeme oneskorenie frontu s časovou periódou jednej sekundy a po použití všetkých slotov do jednej sekundy, žiadny by nemal byť k dispozícii:

public void givenDelayQueue_whenReachLimit_thenBlocked () {int sloty = 50; ExecutorService executorService = Executors.newFixedThreadPool (sloty); DelayQueueUsingTimedSemaphore delayQueue = nový DelayQueueUsingTimedSemaphore (1, sloty); IntStream.range (0, sloty). ForEach (užívateľ -> executorService.execute (delayQueue :: tryAdd)); executorService.shutdown (); assertEquals (0, delayQueue.availableSlots ()); assertFalse (delayQueue.tryAdd ()); }

Ale po nejakom čase spánku semafor by mal resetovať a uvoľniť povolenia:

@ Test public void givenDelayQueue_whenTimePass_thenSlotsAvailable () vyvolá InterruptedException {int sloty = 50; ExecutorService executorService = Executors.newFixedThreadPool (sloty); DelayQueueUsingTimedSemaphore delayQueue = nový DelayQueueUsingTimedSemaphore (1, sloty); IntStream.range (0, sloty). ForEach (užívateľ -> executorService.execute (delayQueue :: tryAdd)); executorService.shutdown (); assertEquals (0, delayQueue.availableSlots ()); Závit. Spánok (1 000); assertTrue (delayQueue.availableSlots ()> 0); assertTrue (delayQueue.tryAdd ()); }

4. Semafor vs. Mutex

Mutex funguje podobne ako binárny semafor, môžeme ho použiť na implementáciu vzájomného vylúčenia.

V nasledujúcom príklade použijeme na zostavenie počítadla jednoduchý binárny semafor:

trieda CounterUsingMutex {private Semaphore mutex; počet súkromných int; CounterUsingMutex () {mutex = nový Semafor (1); počet = 0; } void increase () hodí InterruptedException {mutex.acquire (); this.count = this.count + 1; Závit. Spánok (1 000); mutex.release (); } int getCount () {return this.count; } boolean hasQueuedThreads () {return mutex.hasQueuedThreads (); }}

Keď sa veľa vlákien pokúsi získať prístup k počítadlu naraz, budú jednoducho zablokované v rade:

@ Test public void whenMutexAndMultipleThreads_thenBlocked () hodí InterruptedException {int count = 5; ExecutorService executorService = Executors.newFixedThreadPool (počet); CounterUsingMutex counter = nový CounterUsingMutex (); IntStream.range (0, count) .forEach (user -> executorService.execute (() -> {try {counter.increase ();} catch (InterruptedException e) {e.printStackTrace ();}})); executorService.shutdown (); assertTrue (counter.hasQueuedThreads ()); }

Keď počkáme, všetky vlákna vstúpia do počítadla a nezostanú žiadne vlákna vo fronte:

@ Test public void givenMutexAndMultipleThreads_ThenDelay_thenCorrectCount () vyvolá InterruptedException {int count = 5; ExecutorService executorService = Executors.newFixedThreadPool (počet); CounterUsingMutex counter = nový CounterUsingMutex (); IntStream.range (0, count) .forEach (user -> executorService.execute (() -> {try {counter.increase ();} catch (InterruptedException e) {e.printStackTrace ();}})); executorService.shutdown (); assertTrue (counter.hasQueuedThreads ()); Závit. Spánok (5 000); assertFalse (counter.hasQueuedThreads ()); assertEquals (count, counter.getCount ()); }

5. Záver

V tomto článku sme preskúmali základy semaforov v Jave.

Celý zdrojový kód je ako vždy k dispozícii na serveri GitHub.