streamalism.org

1. Pregled

U ovom brzom vodiču istražit ćemo osnove semafora i muteksa u Javi.

2. Semafor

Počet ćemo s java.util.concurrent.Semaphore. Možemo koristiti semafore da ograničimo broj istodobnih niti koje pristupaju određenom resursu.

U sljedećem ćemo primjeru implementirati jednostavan red prijave kako bismo ograničili broj korisnika u sustavu:

klasa LoginQueueUsingSemaphore {privatni semafor Semaphore; javni LoginQueueUsingSemaphore (int slotLimit) {semaphore = novi Semaphore (slotLimit); } logička probaLogin () {povratak semafor.tryAcquire (); } void logout () {semaphore.release (); } int availableSlots () {vratiti semafor.availablePermissions (); }}

Primijetite kako smo koristili sljedeće metode:

• tryAcquire () - vrati true ako je dozvola odmah dostupna i pribavi je inače vrati false, ali steći() stekne dozvolu i blokira dok je ne postane dostupna
• release () - izdati dozvolu
• availablePermissions () - povratni broj dostupnih trenutnih dozvola

Da bismo testirali svoj red za prijavu, prvo ćemo pokušati doseći ograničenje i provjeriti hoće li sljedeći pokušaj prijave biti blokiran:

@Test javna praznina givenLoginQueue_whenReachLimit_thenBlocked () {int slots = 10; ExecutorService executorService = Izvršitelji.newFixedThreadPool (utora); LoginQueueUsingSemaphore loginQueue = novi LoginQueueUsingSemaphore (mjesta); IntStream.range (0, slots) .forEach (user -> executorService.execute (loginQueue :: tryLogin)); executorService.shutdown (); assertEquals (0, loginQueue.availableSlots ()); assertFalse (loginQueue.tryLogin ()); }

Dalje, vidjet ćemo jesu li dostupni utori nakon odjave:

@Test javna praznina givenLoginQueue_whenLogout_thenSlotsAvailable () {int slots = 10; ExecutorService executorService = Izvršitelji.newFixedThreadPool (utora); LoginQueueUsingSemaphore loginQueue = novi LoginQueueUsingSemaphore (mjesta); 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. Vrijeme Semafor

Dalje ćemo razgovarati o Apache Commons Tempirani semafor. Tempirani semafor dopušta brojne dozvole kao jednostavan Semafor, ali u određenom vremenskom razdoblju, nakon tog razdoblja resetiranje vremena i sve dozvole se oslobađaju.

Možemo koristiti Tempirani semafor za izgradnju jednostavnog reda odgode na sljedeći način:

klasa DelayQueueUsingTimedSemaphore {privatni semafor TimedSemaphore; DelayQueueUsingTimedSemaphore (dugo razdoblje, int slotLimit) {semafor = novi TimedSemaphore (period, TimeUnit.SECONDS, slotLimit); } logička probaAdd () {return semaphore.tryAcquire (); } int availableSlots () {return semaphore.getAvailablePermissions (); }}

Kada koristimo red odgode s jednom sekundom kao vremenskim razdobljem i nakon upotrebe svih mjesta unutar jedne sekunde, nijedan ne bi trebao biti dostupan:

javna praznina givenDelayQueue_whenReachLimit_thenBlocked () {int slots = 50; ExecutorService executorService = Izvršitelji.newFixedThreadPool (utora); DelayQueueUsingTimedSemaphore delayQueue = novi DelayQueueUsingTimedSemaphore (1, mjesta); IntStream.range (0, slots) .forEach (user -> executorService.execute (delayQueue :: tryAdd)); executorService.shutdown (); assertEquals (0, delayQueue.availableSlots ()); assertFalse (delayQueue.tryAdd ()); }

Ali nakon spavanja neko vrijeme, semafor bi trebao resetirati i pustiti dozvole:

@Test javna praznina givenDelayQueue_whenTimePass_thenSlotsAvailable () baca InterruptedException {int slots = 50; ExecutorService executorService = Izvršitelji.newFixedThreadPool (utora); DelayQueueUsingTimedSemaphore delayQueue = novi DelayQueueUsingTimedSemaphore (1, mjesta); IntStream.range (0, slots) .forEach (korisnik -> executorService.execute (delayQueue :: tryAdd)); executorService.shutdown (); assertEquals (0, delayQueue.availableSlots ()); Navoj.spavanje (1000); assertTrue (delayQueue.availableSlots ()> 0); assertTrue (delayQueue.tryAdd ()); }

4. Semafor protiv muteksa

Mutex djeluje slično binarnom semaforu, možemo ga koristiti za uzajamno isključivanje.

U sljedećem ćemo primjeru koristiti jednostavni binarni semafor za izgradnju brojača:

klasa CounterUsingMutex {privatni mutex Semaphore; broj privatnih int; CounterUsingMutex () {mutex = novi Semafor (1); broj = 0; } void povećanje () baca InterruptedException {mutex.acquire (); this.count = this.count + 1; Navoj.spavanje (1000); mutex.release (); } int getCount () {return this.count; } boolean hasQueuedThreads () {return mutex.hasQueuedThreads (); }}

Kada puno niti pokuša odjednom pristupiti brojaču, jednostavno će biti blokirani u redu čekanja:

@Test public void whenMutexAndMultipleThreads_thenBlocked () baca InterruptedException {int count = 5; ExecutorService executorService = Izvršitelji.newFixedThreadPool (count); CounterUsingMutex counter = novi CounterUsingMutex (); IntStream.range (0, count) .forEach (user -> executorService.execute (() -> {try {counter.increase ();} catch (InterruptedException e) {e.printStackTrace ();}}))); executorService.shutdown (); assertTrue (counter.hasQueuedThreads ()); }

Kad čekamo, sve će niti pristupiti brojaču, a niti jedna nit neće ostati u redu:

@Test javna praznina givenMutexAndMultipleThreads_ThenDelay_thenCorrectCount () baca InterruptedException {int count = 5; ExecutorService executorService = Izvršitelji.newFixedThreadPool (count); CounterUsingMutex counter = novi CounterUsingMutex (); IntStream.range (0, count) .forEach (user -> executorService.execute (() -> {try {counter.increase ();} catch (InterruptedException e) {e.printStackTrace ();}}))); executorService.shutdown (); assertTrue (counter.hasQueuedThreads ()); Navoj.spavanje (5000); assertFalse (counter.hasQueuedThreads ()); assertEquals (count, counter.getCount ()); }

5. Zaključak

U ovom smo članku istražili osnove semafora u Javi.

Kao i uvijek, puni izvorni kod dostupan je na GitHub-u.