JUC并发工具包之Semaphore

Posted mrcharleshu

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了JUC并发工具包之Semaphore相关的知识,希望对你有一定的参考价值。

Semaphore (JDK)

我们使用semaphore去限制获取特定资源的并发线程数量。
下面的例子中,我们实现了一个简单的登录队列来限制登入系统的用户数量:

class LoginQueueUsingSemaphore {
 
    private Semaphore semaphore;
 
    public LoginQueueUsingSemaphore(int slotLimit) {
        semaphore = new Semaphore(slotLimit);
    }
 
    boolean tryLogin() {
        return semaphore.tryAcquire();
    }
 
    void logout() {
        semaphore.release();
    }
 
    int availableSlots() {
        return semaphore.availablePermits();
    }
}

注意下我们使用这些方法的方式:

  • tryAcquire():如果还有可用的permit(构造方法传入的,表示限制的线程数量)则立即返回true,否则返回false,但是acquire()方法会以阻塞的方式获取一个permit。
  • release():释放一个permit。
  • availablePermits():返回当前可用的permit数量。

我们来测试一下我们的登录队列,我们首先使用完所有的permit,然后再获取一个看看是否会被阻塞:

@Test
public void givenLoginQueue_whenReachLimit_thenBlocked() {
    int slots = 10;
    ExecutorService executorService = Executors.newFixedThreadPool(slots);
    LoginQueueUsingSemaphore loginQueue = new LoginQueueUsingSemaphore(slots);
    IntStream.range(0, slots)
      .forEach(user -> executorService.execute(loginQueue::tryLogin));
    executorService.shutdown();
 
    assertEquals(0, loginQueue.availableSlots());
    assertFalse(loginQueue.tryLogin());
}

现在我们logout()一下看看是否有可用的permit:

@Test
public void givenLoginQueue_whenLogout_thenSlotsAvailable() {
    int slots = 10;
    ExecutorService executorService = Executors.newFixedThreadPool(slots);
    LoginQueueUsingSemaphore loginQueue = new LoginQueueUsingSemaphore(slots);
    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());
}

结果显而易见。

Timed Semaphore (Apache Commons)

我们现在看看ApacheCommons下实现的TimedSemaphore。TimedSemaphore允许在既定的时间内维护一定数量的Semaphore(这段时间内和JDK实现的Semaphore效果一样),当时间过去后会释放所有的permits。

我们可以使用TimedSemaphore来构建一个简单的延时队列:

class DelayQueueUsingTimedSemaphore {
 
    private TimedSemaphore semaphore;
 
    DelayQueueUsingTimedSemaphore(long period, int slotLimit) {
        semaphore = new TimedSemaphore(period, TimeUnit.SECONDS, slotLimit);
    }
 
    boolean tryAdd() {
        return semaphore.tryAcquire();
    }
 
    int availableSlots() {
        return semaphore.getAvailablePermits();
    }
}

现在我们设置超时时间1秒,在1秒钟之内使用完所有的permit再次尝试获取的时候就会没有可用的permit:

public void givenDelayQueue_whenReachLimit_thenBlocked() {
    int slots = 50;
    ExecutorService executorService = Executors.newFixedThreadPool(slots);
    DelayQueueUsingTimedSemaphore delayQueue 
      = new DelayQueueUsingTimedSemaphore(1, slots);
     
    IntStream.range(0, slots)
      .forEach(user -> executorService.execute(delayQueue::tryAdd));
    executorService.shutdown();
 
    assertEquals(0, delayQueue.availableSlots());
    assertFalse(delayQueue.tryAdd());
}

但是把线程休眠1秒后,这时候semaphore会重置并释放所有的permits

@Test
public void givenDelayQueue_whenTimePass_thenSlotsAvailable() throws InterruptedException {
    int slots = 50;
    ExecutorService executorService = Executors.newFixedThreadPool(slots);
    DelayQueueUsingTimedSemaphore delayQueue = new DelayQueueUsingTimedSemaphore(1, slots);
    IntStream.range(0, slots)
      .forEach(user -> executorService.execute(delayQueue::tryAdd));
    executorService.shutdown();
 
    assertEquals(0, delayQueue.availableSlots());
    Thread.sleep(1000);
    assertTrue(delayQueue.availableSlots() > 0);
    assertTrue(delayQueue.tryAdd());
}

Semaphore vs. Mutex

Mutex像是一个二进制的Semaphore,我们可以使用它来实现互斥。
在下面的这个例子中,我们使用一个permit为1的Semaphore来构建一个计数器:

class CounterUsingMutex {
 
    private Semaphore mutex;
    private int count;
 
    CounterUsingMutex() {
        mutex = new Semaphore(1);
        count = 0;
    }
 
    void increase() throws InterruptedException {
        mutex.acquire();
        this.count = this.count + 1;
        Thread.sleep(1000);
        mutex.release();
 
    }
 
    int getCount() {
        return this.count;
    }
 
    boolean hasQueuedThreads() {
        return mutex.hasQueuedThreads();
    }
}    

当大量线程同时来操作counter的时候,他们都会在队列中阻塞:

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

我们把线程休眠一会后,所有的线程都将能操作counter,这时队列中就没有等待排队的线程了。

@Test
public void givenMutexAndMultipleThreads_ThenDelay_thenCorrectCount()
 throws InterruptedException {
    int count = 5;
    ExecutorService executorService
     = Executors.newFixedThreadPool(count);
    CounterUsingMutex counter = new CounterUsingMutex();
    IntStream.range(0, count)
      .forEach(user -> executorService.execute(() -> {
          try {
              counter.increase();
          } catch (InterruptedException e) {
              e.printStackTrace();
          }
      }));
    executorService.shutdown();
 
    assertTrue(counter.hasQueuedThreads());
    Thread.sleep(5000);
    assertFalse(counter.hasQueuedThreads());
    assertEquals(count, counter.getCount());
}

CodeRepo

完整代码在这



以上是关于JUC并发工具包之Semaphore的主要内容,如果未能解决你的问题,请参考以下文章

JUC并发编程 共享模式之工具 JUC Semaphore(信号量) -- 介绍 & 使用

JUC并发编程 共享模式之工具 JUC Semaphore(信号量)-- 限制对共享资源的使用 改进数据库连接池

JUC系列LOCK框架系列之四 同步工具类Semaphore

JUC并发编程 共享模式之工具 JUC CountdownLatch(倒计时锁) -- CountdownLatch应用(等待多个线程准备完毕( 可以覆盖上次的打印内)等待多个远程调用结束)(代码片段

java高并发系列 - 第15天:JUC中的Semaphore,最简单的限流工具类,必备技能

JUC回顾之-Semaphore底层实现和原理