From 20669dc6a1fb6de452c0da5bf1f49a8d37eb0fd3 Mon Sep 17 00:00:00 2001 From: Doha2012 Date: Mon, 3 Jul 2017 17:34:41 +0200 Subject: [PATCH] semaphores (#2198) * minor logging fix * spring security sso * use basic auth * use form login * cleanup * cleanup * final cleanup * second client app for sso * spring boot bootstrap * add logic * cleanup * add simple controller * add thymeleaf and security * minor fix * minor fix * add more boot properties * fix live test * fix live test * minor fix * semaphores --- .../semaphores/CounterUsingMutex.java | 31 ++++ .../DelayQueueUsingTimedSemaphore.java | 23 +++ .../semaphores/LoginQueueUsingSemaphore.java | 25 ++++ .../semaphores/SemaphoresManualTest.java | 140 ++++++++++++++++++ 4 files changed, 219 insertions(+) create mode 100644 core-java/src/main/java/com/baeldung/concurrent/semaphores/CounterUsingMutex.java create mode 100644 core-java/src/main/java/com/baeldung/concurrent/semaphores/DelayQueueUsingTimedSemaphore.java create mode 100644 core-java/src/main/java/com/baeldung/concurrent/semaphores/LoginQueueUsingSemaphore.java create mode 100644 core-java/src/test/java/com/baeldung/concurrent/semaphores/SemaphoresManualTest.java diff --git a/core-java/src/main/java/com/baeldung/concurrent/semaphores/CounterUsingMutex.java b/core-java/src/main/java/com/baeldung/concurrent/semaphores/CounterUsingMutex.java new file mode 100644 index 0000000000..8ef9cc6ea6 --- /dev/null +++ b/core-java/src/main/java/com/baeldung/concurrent/semaphores/CounterUsingMutex.java @@ -0,0 +1,31 @@ +package com.baeldung.concurrent.semaphores; + +import java.util.concurrent.Semaphore; + +public class CounterUsingMutex { + + private final Semaphore mutex; + private int count; + + public CounterUsingMutex() { + mutex = new Semaphore(1); + count = 0; + } + + public void increase() throws InterruptedException { + mutex.acquire(); + this.count = this.count + 1; + Thread.sleep(1000); + mutex.release(); + + } + + public int getCount() { + return this.count; + } + + public boolean hasQueuedThreads() { + return mutex.hasQueuedThreads(); + } + +} diff --git a/core-java/src/main/java/com/baeldung/concurrent/semaphores/DelayQueueUsingTimedSemaphore.java b/core-java/src/main/java/com/baeldung/concurrent/semaphores/DelayQueueUsingTimedSemaphore.java new file mode 100644 index 0000000000..a27b490fde --- /dev/null +++ b/core-java/src/main/java/com/baeldung/concurrent/semaphores/DelayQueueUsingTimedSemaphore.java @@ -0,0 +1,23 @@ +package com.baeldung.concurrent.semaphores; + +import java.util.concurrent.TimeUnit; + +import org.apache.commons.lang3.concurrent.TimedSemaphore; + +public class DelayQueueUsingTimedSemaphore { + + private final TimedSemaphore semaphore; + + public DelayQueueUsingTimedSemaphore(long period, int slotLimit) { + semaphore = new TimedSemaphore(period, TimeUnit.SECONDS, slotLimit); + } + + public boolean tryAdd() { + return semaphore.tryAcquire(); + } + + public int availableSlots() { + return semaphore.getAvailablePermits(); + } + +} diff --git a/core-java/src/main/java/com/baeldung/concurrent/semaphores/LoginQueueUsingSemaphore.java b/core-java/src/main/java/com/baeldung/concurrent/semaphores/LoginQueueUsingSemaphore.java new file mode 100644 index 0000000000..38dbc53ad3 --- /dev/null +++ b/core-java/src/main/java/com/baeldung/concurrent/semaphores/LoginQueueUsingSemaphore.java @@ -0,0 +1,25 @@ +package com.baeldung.concurrent.semaphores; + +import java.util.concurrent.Semaphore; + +public class LoginQueueUsingSemaphore { + + private final Semaphore semaphore; + + public LoginQueueUsingSemaphore(int slotLimit) { + semaphore = new Semaphore(slotLimit); + } + + public boolean tryLogin() { + return semaphore.tryAcquire(); + } + + public void logout() { + semaphore.release(); + } + + public int availableSlots() { + return semaphore.availablePermits(); + } + +} diff --git a/core-java/src/test/java/com/baeldung/concurrent/semaphores/SemaphoresManualTest.java b/core-java/src/test/java/com/baeldung/concurrent/semaphores/SemaphoresManualTest.java new file mode 100644 index 0000000000..9686bf6444 --- /dev/null +++ b/core-java/src/test/java/com/baeldung/concurrent/semaphores/SemaphoresManualTest.java @@ -0,0 +1,140 @@ +package com.baeldung.concurrent.semaphores; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.stream.IntStream; + +import org.junit.Test; + +public class SemaphoresManualTest { + + // ========= login queue ====== + + @Test + public void givenLoginQueue_whenReachLimit_thenBlocked() { + final int slots = 10; + final ExecutorService executorService = Executors.newFixedThreadPool(slots); + final LoginQueueUsingSemaphore loginQueue = new LoginQueueUsingSemaphore(slots); + IntStream.range(0, slots) + .forEach(user -> executorService.execute(new Runnable() { + @Override + public void run() { + loginQueue.tryLogin(); + } + })); + executorService.shutdown(); + + assertEquals(0, loginQueue.availableSlots()); + assertFalse(loginQueue.tryLogin()); + } + + @Test + public void givenLoginQueue_whenLogout_thenSlotsAvailable() { + final int slots = 10; + final ExecutorService executorService = Executors.newFixedThreadPool(slots); + final LoginQueueUsingSemaphore loginQueue = new LoginQueueUsingSemaphore(slots); + IntStream.range(0, slots) + .forEach(user -> executorService.execute(new Runnable() { + @Override + public void run() { + loginQueue.tryLogin(); + } + })); + executorService.shutdown(); + + assertEquals(0, loginQueue.availableSlots()); + loginQueue.logout(); + assertTrue(loginQueue.availableSlots() > 0); + assertTrue(loginQueue.tryLogin()); + } + + // ========= delay queue ======= + + @Test + public void givenDelayQueue_whenReachLimit_thenBlocked() { + final int slots = 50; + final ExecutorService executorService = Executors.newFixedThreadPool(slots); + final DelayQueueUsingTimedSemaphore delayQueue = new DelayQueueUsingTimedSemaphore(1, slots); + IntStream.range(0, slots) + .forEach(user -> executorService.execute(new Runnable() { + @Override + public void run() { + delayQueue.tryAdd(); + } + })); + executorService.shutdown(); + + assertEquals(0, delayQueue.availableSlots()); + assertFalse(delayQueue.tryAdd()); + } + + @Test + public void givenDelayQueue_whenTimePass_thenSlotsAvailable() throws InterruptedException { + final int slots = 50; + final ExecutorService executorService = Executors.newFixedThreadPool(slots); + final DelayQueueUsingTimedSemaphore delayQueue = new DelayQueueUsingTimedSemaphore(1, slots); + IntStream.range(0, slots) + .forEach(user -> executorService.execute(new Runnable() { + @Override + public void run() { + delayQueue.tryAdd(); + } + })); + executorService.shutdown(); + + assertEquals(0, delayQueue.availableSlots()); + Thread.sleep(1000); + assertTrue(delayQueue.availableSlots() > 0); + assertTrue(delayQueue.tryAdd()); + } + + // ========== mutex ======== + + @Test + public void whenMutexAndMultipleThreads_thenBlocked() throws InterruptedException { + final int count = 5; + final ExecutorService executorService = Executors.newFixedThreadPool(count); + final CounterUsingMutex counter = new CounterUsingMutex(); + IntStream.range(0, count) + .forEach(user -> executorService.execute(new Runnable() { + @Override + public void run() { + try { + counter.increase(); + } catch (final InterruptedException e) { + e.printStackTrace(); + } + } + })); + executorService.shutdown(); + + assertTrue(counter.hasQueuedThreads()); + } + + @Test + public void givenMutexAndMultipleThreads_ThenDelay_thenCorrectCount() throws InterruptedException { + final int count = 5; + final ExecutorService executorService = Executors.newFixedThreadPool(count); + final CounterUsingMutex counter = new CounterUsingMutex(); + IntStream.range(0, count) + .forEach(user -> executorService.execute(new Runnable() { + @Override + public void run() { + try { + counter.increase(); + } catch (final InterruptedException e) { + e.printStackTrace(); + } + } + })); + executorService.shutdown(); + assertTrue(counter.hasQueuedThreads()); + Thread.sleep(5000); + assertFalse(counter.hasQueuedThreads()); + assertEquals(count, counter.getCount()); + } +}