多线程控制并发数目工具类Semaphore

前言

在多线程编码过程中，我们会用到多线程来提升运行效率。比如我们的Executors创建线程池，程序尽可能的压榨CPU资源来提升我们程序吞吐量。但是过度的使用线程，也会将我们CPU资源榨干，从而让我们系统不能正常的提供服务。故今天我们引入JUC并发包下面的semaphore并发类，该类可以同时允许定量线程执行从而达到控制并发的目的。

Semaphore原理

Semaphore并发类提供了两个核心方法：acquire()方法和release()方法。acquire()方法表示获取一个许可，如果没有则等待，release()方法则是释放对应的许可。Semaphore维护了当前访问的个数，通过提供同步机制来控制同时访问的个数。

Semaphore源码解析

内部继承AQS保证同步

Semaphore 与AQS关系图：

进入java.util.concurrent 包下 Semaphore 并发类查看源码：

//继承AQS同步阻塞队列来实现同步功能
abstract static class Sync extends AbstractQueuedSynchronizer {private static final long serialVersionUID = 1192457210091910933L;Sync(int permits) {setState(permits);}final int getPermits() {return getState();}final int nonfairTryAcquireShared(int acquires) {for (;;) {int available = getState();int remaining = available - acquires;if (remaining < 0 ||compareAndSetState(available, remaining))return remaining;}}

如上所示，首先Semaphore并发内部类继承AbstractQueuedSynchronizer 同步阻塞类，所以可以得出Semaphore是通过AQS机制来保证同步。AQS不明白的同学可以看我之前的博文，大概就是内部一个state状态，获取到资源 state 就加一，释放资源 state 就减一，当 state == 0 的时候表示阻塞队列中的其他线程可以获取该资源。

继续查看源码：

/*** Creates a {@code Semaphore} with the given number of* permits and nonfair fairness setting.** @param permits the initial number of permits available.*        This value may be negative, in which case releases*        must occur before any acquires will be granted.*/
public Semaphore(int permits) {sync = new NonfairSync(permits);
}/*** Creates a {@code Semaphore} with the given number of* permits and the given fairness setting.** @param permits the initial number of permits available.*        This value may be negative, in which case releases*        must occur before any acquires will be granted.* @param fair {@code true} if this semaphore will guarantee*        first-in first-out granting of permits under contention,*        else {@code false}*/
public Semaphore(int permits, boolean fair) {sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}

如源码所示，Semaphore构造方法默认提供非公平锁同步构造方法，也提供了一个用户自定义是否公平锁的构造方法。是否公平锁同步直接影响阻塞队列中的哪个线程可以获取资源，公平锁是按照阻塞顺序获取资源，非公平锁是多个线程争抢资源。当然构造方法必须传入并发许可的总数，这个总数直接影响后续我们可以同时获取多少个许可。

acquire获取许可

查看Semaphore 获取许可的源码：

//是否可以获取许可
public boolean tryAcquire() {return sync.nonfairTryAcquireShared(1) >= 0;
}
//是否可以获取许可，并提供超时获取机制
public boolean tryAcquire(long timeout, TimeUnit unit)throws InterruptedException {return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}
//是否可以获取多个许可，并提供超时获取机制
public boolean tryAcquire(int permits, long timeout, TimeUnit unit)throws InterruptedException {if (permits < 0) throw new IllegalArgumentException();return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
}
//提供一个无参获取许可方法，默认获取一个许可
public void acquire() throws InterruptedException {sync.acquireSharedInterruptibly(1);
}//不间断的获取许可，如果获取不到就阻塞
public void acquireUninterruptibly() {sync.acquireShared(1);
}
//不间断的获取多个许可，如果获取不到就阻塞
public void acquireUninterruptibly(int permits) {if (permits < 0) throw new IllegalArgumentException();sync.acquireShared(permits);
}

如源码所示，我获取许可有很多的方法。有是否可以获取许可，超时是否可以获取许可。当然还有获取许可的方法，以及阻塞获取许可的多个方法，这些方法本质上都是调用父AQS中的获取资源许可的方法，同学们可以选择自己适用的方法进行调用。

release释放许可

查看Semaphore 释放许可的源码：

//默认释放一个许可
public void release() {sync.releaseShared(1);
}
//同时释放多个许可
public void release(int permits) {if (permits < 0) throw new IllegalArgumentException();sync.releaseShared(permits);
}
//调用父AQS的释放资源的方法
public final boolean releaseShared(int arg) {if (tryReleaseShared(arg)) {doReleaseShared();return true;}return false;
}

如源码所示，Semaphore提供了多个释放许可的方法，我们可以根据实际需要选择释放许可的方法。
值得注意的是release() 方法调用了自己内部类Sync的释放资源方法，而Sync又是继承AQS阻塞队列并调用了父类doReleaseShared() 释放资源的方法。

实战演示

上面博文我们介绍了Semaphore并发类是一个同步类，它继承了AQS阻塞队列。Semaphore主要提供了acquire() 获取许可与release() 释放许可的方法，通过这两个方法我们可以实现线程并发数目的功能。下面我们简单模拟一下限制并发数目。

1、Semaphore测试类
测试类提供一个定量5的线程池，用countdownlatch 让主线程等待子线程执行完成，Semaphore保证每次只有3个线程执行。为了保证测试结果可以追溯性，我们在业务逻辑中让线程睡眠3s。

/*** Semaphore test* @author senfel* @date 2023/4/6 11:38 * @return*/
@SpringBootTest
class ConcurrentApplicationTests {/*** 日期格式*/private static DateTimeFormatter formatter = DateTimeFormatter.ofPattern("HH:mm:ss");@Testvoid testSemaphore() throws Exception {//创建线程为5的线程池ExecutorService executorService = Executors.newFixedThreadPool(5);//同时运行3个线程运行Semaphore semaphore = new Semaphore(3);//等待执行完成CountDownLatch countDownLatch = new CountDownLatch(10);for(int i = 0;i<10;i++) {executorService.execute(new Runnable() {@Overridepublic void run() {try{//获取许可semaphore.acquire();//业务逻辑executeFun();//释放许可semaphore.release();countDownLatch.countDown();}catch (Exception e){e.printStackTrace();}}});}countDownLatch.await();System.err.println("执行完成");executorService.shutdown();}/*** 执行业务方法* @author senfel* @date 2023/4/6 11:13* @return void*/private void executeFun() {try {String startTime = formatter.format(LocalDateTime.now());Thread.sleep(3000);System.err.println("当前执行线程："+Thread.currentThread().getName()+",开始时间："+startTime+",结束时间"+formatter.format(LocalDateTime.now()));} catch (Exception e) {e.printStackTrace();}}}

2、执行测试方法
执行testSemaphore() 方法，我们可以在控制台看到如下结果：

当前执行线程：pool-1-thread-1,开始时间：11:37:00,结束时间11:37:03
当前执行线程：pool-1-thread-3,开始时间：11:37:00,结束时间11:37:03
当前执行线程：pool-1-thread-2,开始时间：11:37:00,结束时间11:37:03
当前执行线程：pool-1-thread-2,开始时间：11:37:03,结束时间11:37:06
当前执行线程：pool-1-thread-1,开始时间：11:37:03,结束时间11:37:06
当前执行线程：pool-1-thread-3,开始时间：11:37:03,结束时间11:37:06
当前执行线程：pool-1-thread-1,开始时间：11:37:06,结束时间11:37:09
当前执行线程：pool-1-thread-4,开始时间：11:37:06,结束时间11:37:09
当前执行线程：pool-1-thread-2,开始时间：11:37:06,结束时间11:37:09
当前执行线程：pool-1-thread-5,开始时间：11:37:09,结束时间11:37:12
执行完成

根据执行结果我们可以发现每3s有3个线程执行并输出，得出结论Semaphore可以保证每次只能同时3个线程执行！！！！

总结

多线程控制并发数目工具类Semaphore内部继承AQS抽象阻塞队列，并继承了其同步获取、释放资源的方法。Semaphore提供了acquire()获取许可、release()释放许可方法，底层当然调用的是AQS内部方法来满足同步以保证限制并发线程运行数目。