آشنایی با Semaphore در جاوا — راهنمای جامع

1	
2class LoginQueueUsingSemaphore {
3 
4    private Semaphore semaphore;
5 
6    public LoginQueueUsingSemaphore(int slotLimit) {
7        semaphore = new Semaphore(slotLimit);
8    }
9 
10    boolean tryLogin() {
11        return semaphore.tryAcquire();
12    }
13 
14    void logout() {
15        semaphore.release();
16    }
17 
18    int availableSlots() {
19        return semaphore.availablePermits();
20    }
21 
22}

1@Test
2public void givenLoginQueue_whenReachLimit_thenBlocked() {
3    int slots = 10;
4    ExecutorService executorService = Executors.newFixedThreadPool(slots);
5    LoginQueueUsingSemaphore loginQueue = new LoginQueueUsingSemaphore(slots);
6    IntStream.range(0, slots)
7      .forEach(user -> executorService.execute(loginQueue::tryLogin));
8    executorService.shutdown();
9 
10    assertEquals(0, loginQueue.availableSlots());
11    assertFalse(loginQueue.tryLogin());
12}

1@Test
2public void givenLoginQueue_whenLogout_thenSlotsAvailable() {
3    int slots = 10;
4    ExecutorService executorService = Executors.newFixedThreadPool(slots);
5    LoginQueueUsingSemaphore loginQueue = new LoginQueueUsingSemaphore(slots);
6    IntStream.range(0, slots)
7      .forEach(user -> executorService.execute(loginQueue::tryLogin));
8    executorService.shutdown();
9    assertEquals(0, loginQueue.availableSlots());
10    loginQueue.logout();
11 
12    assertTrue(loginQueue.availableSlots() > 0);
13    assertTrue(loginQueue.tryLogin());
14}

1class DelayQueueUsingTimedSemaphore {
2 
3    private TimedSemaphore semaphore;
4 
5    DelayQueueUsingTimedSemaphore(long period, int slotLimit) {
6        semaphore = new TimedSemaphore(period, TimeUnit.SECONDS, slotLimit);
7    }
8 
9    boolean tryAdd() {
10        return semaphore.tryAcquire();
11    }
12 
13    int availableSlots() {
14        return semaphore.getAvailablePermits();
15    }
16 
17}

1public void givenDelayQueue_whenReachLimit_thenBlocked() {
2    int slots = 50;
3    ExecutorService executorService = Executors.newFixedThreadPool(slots);
4    DelayQueueUsingTimedSemaphore delayQueue 
5      = new DelayQueueUsingTimedSemaphore(1, slots);
6     
7    IntStream.range(0, slots)
8      .forEach(user -> executorService.execute(delayQueue::tryAdd));
9    executorService.shutdown();
10 
11    assertEquals(0, delayQueue.availableSlots());
12    assertFalse(delayQueue.tryAdd());
13}

1@Test
2public void givenDelayQueue_whenTimePass_thenSlotsAvailable() throws InterruptedException {
3    int slots = 50;
4    ExecutorService executorService = Executors.newFixedThreadPool(slots);
5    DelayQueueUsingTimedSemaphore delayQueue = new DelayQueueUsingTimedSemaphore(1, slots);
6    IntStream.range(0, slots)
7      .forEach(user -> executorService.execute(delayQueue::tryAdd));
8    executorService.shutdown();
9 
10    assertEquals(0, delayQueue.availableSlots());
11    Thread.sleep(1000);
12    assertTrue(delayQueue.availableSlots() > 0);
13    assertTrue(delayQueue.tryAdd());
14}

1class CounterUsingMutex {
2 
3    private Semaphore mutex;
4    private int count;
5 
6    CounterUsingMutex() {
7        mutex = new Semaphore(1);
8        count = 0;
9    }
10 
11    void increase() throws InterruptedException {
12        mutex.acquire();
13        this.count = this.count + 1;
14        Thread.sleep(1000);
15        mutex.release();
16 
17    }
18 
19    int getCount() {
20        return this.count;
21    }
22 
23    boolean hasQueuedThreads() {
24        return mutex.hasQueuedThreads();
25    }
26}

1@Test
2public void whenMutexAndMultipleThreads_thenBlocked()
3 throws InterruptedException {
4    int count = 5;
5    ExecutorService executorService
6     = Executors.newFixedThreadPool(count);
7    CounterUsingMutex counter = new CounterUsingMutex();
8    IntStream.range(0, count)
9      .forEach(user -> executorService.execute(() -> {
10          try {
11              counter.increase();
12          } catch (InterruptedException e) {
13              e.printStackTrace();
14          }
15      }));
16    executorService.shutdown();
17 
18    assertTrue(counter.hasQueuedThreads());
19}

1@Test
2public void givenMutexAndMultipleThreads_ThenDelay_thenCorrectCount()
3 throws InterruptedException {
4    int count = 5;
5    ExecutorService executorService
6     = Executors.newFixedThreadPool(count);
7    CounterUsingMutex counter = new CounterUsingMutex();
8    IntStream.range(0, count)
9      .forEach(user -> executorService.execute(() -> {
10          try {
11              counter.increase();
12          } catch (InterruptedException e) {
13              e.printStackTrace();
14          }
15      }));
16    executorService.shutdown();
17 
18    assertTrue(counter.hasQueuedThreads());
19    Thread.sleep(5000);
20    assertFalse(counter.hasQueuedThreads());
21    assertEquals(count, counter.getCount());
22}