java多线程解说【拾柒】_并发工具类：Semaphore

上篇文章：java多线程解说【拾陆】_并发工具类：CyclicBarrier

本文介绍一下最后一个并发工具类Semaphore。

Semaphore

Semaphore(信号量)是用来控制同时访问特定资源的线程数量，通过协调各个线程以保证合理的使用公共资源。

它同样通过设置一个数字，当请求的数量达到指定的数字时，就将请求拦截；直到有线程释放了资源后，则放一个请求进入。

Semaphore提供了2个构造器：

public Semaphore(int permits) {
sync = new NonfairSync(permits);
}

public Semaphore(int permits, boolean fair) {
sync = (fair)? new FairSync(permits) : new NonfairSync(permits);
}

后一个构造方法是支持公平锁的，也就是先等待的可先进入。

Semaphore通过acquire()来获取一个许可，若无许可能够获得，则会一直等待，直到获得许可；release()用来释放许可。注意，在释放许可之前，必须先获获得许可。在获取许可之前，还可以调用availablePermits()方法来方法得到可用的许可数目；还可以通过getQueueLength()来获取在排队获取许可证的线程数。

一个例子

public class TestSemaphore {

public static void main(String[] args) {
int N = 8;            //工人数
Semaphore semaphore = new Semaphore(5); //机器数目
for(int i=0;i<N;i++)
new WorkerThread(i,semaphore).start();
}

static class WorkerThread extends Thread{
private int num;
private Semaphore semaphore;
public WorkerThread(int num,Semaphore semaphore){
this.num = num;
this.semaphore = semaphore;
}

@Override
public void run() {
try {
semaphore.acquire();
System.out.println("工人"+this.num+"占用一个机器在生产...");
Thread.sleep(2000);
System.out.println("工人"+this.num+"释放出机器");
semaphore.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

}

执行结果为

工人1占用一个机器在生产...
工人4占用一个机器在生产...
工人5占用一个机器在生产...
工人0占用一个机器在生产...
工人2占用一个机器在生产...
工人1释放出机器
工人3占用一个机器在生产...
工人2释放出机器
工人6占用一个机器在生产...
工人4释放出机器
工人7占用一个机器在生产...
工人5释放出机器
工人0释放出机器
工人3释放出机器
工人6释放出机器
工人7释放出机器

常用API

// 创建具有给定的许可数和非公平的公平设置的 Semaphore。
Semaphore(int permits)
// 创建具有给定的许可数和给定的公平设置的 Semaphore。
Semaphore(int permits, boolean fair)

// 从此信号量获取一个许可，在提供一个许可前一直将线程阻塞，否则线程被中断。
void acquire()
// 从此信号量获取给定数目的许可，在提供这些许可前一直将线程阻塞，或者线程已被中断。
void acquire(int permits)
// 从此信号量中获取许可，在有可用的许可前将其阻塞。
void acquireUninterruptibly()
// 从此信号量获取给定数目的许可，在提供这些许可前一直将线程阻塞。
void acquireUninterruptibly(int permits)
// 返回此信号量中当前可用的许可数。
int availablePermits()
// 获取并返回立即可用的所有许可。
int drainPermits()
// 返回一个 collection，包含可能等待获取的线程。
protected Collection<Thread> getQueuedThreads()
// 返回正在等待获取的线程的估计数目。
int getQueueLength()
// 查询是否有线程正在等待获取。
boolean hasQueuedThreads()
// 如果此信号量的公平设置为 true，则返回 true。
boolean isFair()
// 根据指定的缩减量减小可用许可的数目。
protected void reducePermits(int reduction)
// 释放一个许可，将其返回给信号量。
void release()
// 释放给定数目的许可，将其返回到信号量。
void release(int permits)
// 返回标识此信号量的字符串，以及信号量的状态。
String toString()
// 仅在调用时此信号量存在一个可用许可，才从信号量获取许可。
boolean tryAcquire()
// 仅在调用时此信号量中有给定数目的许可时，才从此信号量中获取这些许可。
boolean tryAcquire(int permits)
// 如果在给定的等待时间内此信号量有可用的所有许可，并且当前线程未被中断，则从此信号量获取给定数目的许可。
boolean tryAcquire(int permits, long timeout, TimeUnit unit)
// 如果在给定的等待时间内，此信号量有可用的许可并且当前线程未被中断，则从此信号量获取一个许可。
boolean tryAcquire(long timeout, TimeUnit unit)

实现分析

Semaphore内部也维护了一个sync对象，Sync是一个继承于AQS的抽象类，有两个子类：公平实现FairSync和非公平实现NonfairSync。默认为NonfairSync。公平信号量和非公平信号量的释放信号量的机制是一样的，不同的是它们获取信号量的机制。线程在尝试获取信号量许可时，对于公平信号量而言，如果当前线程不在CLH队列的头部，则排队等候；而对于非公平信号量而言，无论当前线程是不是在CLH队列的头部，它都会直接获取信号量。该差异具体的体现在，它们的tryAcquireShared()函数的实现不同。(详见之前分析CLH的文章)

核心实现也在于重写了tryReleaseShared()方法，通过对ASQ中state变量(用作信号量)的CAS操作来保证只能有指定数量线程获得操作。