一、控制并发线程数Semaphore
生活中，我们过桥，如果桥就能过3个人，那么一次就只能走三个人，如果多了，那么就会有人掉河里了。这就是Semaphore控制人数过桥。（而此处就是Semaphore控制只能有特定数量的线程访问指定资源）。
二、继承与实现关系
public class Semaphore implements java.io.Serializable
三、Semaphore源码解析
自定义同步器
/**
*
* 自定义同步器继承AQS,使用AQS的状态state来控制同时访问的线程数(流量)
*/
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;
/**
* 如果剩余的同步状态值小于0或者当前的同步状态值为available
* 将当前同步状态值更新为remaining
*/
if (remaining < 0 ||
compareAndSetState(available, remaining))
//返回当前最新的同步状态值
return remaining;
}
}
//采用共享方式释放同步状态值
protected final boolean tryReleaseShared(int releases) {
//死循环
for (;;) {
//获取当前的同步状态值
int current = getState();
//计算如果释放同步状态值之后，得到的结果next
int next = current + releases;
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
//如果当前同步状态值为current，那么更新当前的同步状态值为next
if (compareAndSetState(current, next))
//返回true
return true;
}
}
//减少并发线程的数量
final void reducePermits(int reductions) {
//死循环
for (;;) {
//获取当前的同步状态值
int current = getState();
//计算出剩下的并发线程数
int next = current - reductions;
if (next > current) // underflow
throw new Error("Permit count underflow");
//如果当前同步状态值为current，那么更新当前的同步状态值为next
if (compareAndSetState(current, next))
return;
}
}
//将并发的线程数量调整为0
final int drainPermits() {
//死循环
for (;;) {
//获取当前的同步状态值
int current = getState();
/**
* 如果当前的同步状态值为0或者当前同步状态值等于current
* 那么将当前的同步状态值current更新为0
*/
if (current == 0 || compareAndSetState(current, 0))
return current;
}
}
}
解释①：Semaphore自定义同步器Sync采用AQS这样一个抽象模板类来实现自定义的同步器。
解释②：自定义的同步器Sync采用了AQS中的共享锁实现的。
解释③：非公平方式获取共享锁采用自旋+cas来获取同步状态值。
解释④：释放共享锁采用自旋+cas来释放更新同步状态值。
解释⑤：减少并发数量采用的还是自旋+cas来更新同步状态值。
解释⑥：将并发线程数量调整为0的策略就是自旋+cas更新同步状态值为0。
非公平同步器 / 公平同步器
/**
* 非公平同步器
*/
static final class NonfairSync extends Sync {
private static final long serialVersionUID = -2694183684443567898L;
//非公平同步器构造器
NonfairSync(int permits) {
super(permits);
}
//在共享模式下尝试获取同步状态值
protected int tryAcquireShared(int acquires) {
return nonfairTryAcquireShared(acquires);
}
}
/**
* 公平同步器
*/
static final class FairSync extends Sync {
private static final long serialVersionUID = 2014338818796000944L;
//公平同步器构造器
FairSync(int permits) {
super(permits);
}
//在共享模式下尝试获取同步状态值
protected int tryAcquireShared(int acquires) {
//死循环
for (;;) {
//当前队列是否有前驱
if (hasQueuedPredecessors())
return -1;
//获取当前的同步状态值
int available = getState();
//计算出剩下的并发线程数
int remaining = available - acquires;
/**
* 如果剩下的并发线程数小于0或者当前同步状态值等于available
* 那么将当前的同步状态值更新为remaining
*/
if (remaining < 0 ||
compareAndSetState(available, remaining))
//返回计算出的最新同步状态值
return remaining;
}
}
}
解释：非公平和公平同步器在尝试获取对象的共享锁都是采用了自旋+cas。
构造器
/**
* 创建一个指定并发线程数的非公平同步器构造器
*/
public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
/**
* 创建一个指定并发线程数、是否公平的同步器构造器
*/
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
方法
acquire方法
/**
* 在信号量Semaphore中获取一个许可
* 在获取一个许可前线程将会阻塞，否则线程被中断
* 整体许可数将减少1
*/
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
解释：采用aqs的共享锁的策略：首先是否允许尝试获取共享锁，以死循环的方式获取，判断当前线程节点的前驱p是否为头节点，如果为头节点就将当前线程节点设置为头节点，并断开与p的关系，否则继续循环。
tryAcquire方法
/**
* 仅在调用时此信号量存在一个可用许可，才从信号量获取许可。
*/
public boolean tryAcquire() {
return sync.nonfairTryAcquireShared(1) >= 0;
}
解释：
采用死循环+cas判断剩余的同步状态值是否大于等于0。
release方法
/**
* 释放一个许可，将其返回给信号量
*/
public void release() {
sync.releaseShared(1);
}
解释：采用AQS的释放共享锁：首先是否允许释放共享锁，然后以死循环的方式从头节点往后遍历，如果头节点不为null或者队列中的存在元素，那么就释放头节点，然后唤醒头节点的后继结点。
其他常用方法
/**
* 返回此信号量中当前可用的许可数
*/
public int availablePermits() {
return sync.getPermits();
}
/**
* 获取并返回立即可用的所有许可
*/
public int drainPermits() {
return sync.drainPermits();
}
/**
* 根据指定的缩减量减小可用许可的数目
*/
protected void reducePermits(int reduction) {
if (reduction < 0) throw new IllegalArgumentException();
sync.reducePermits(reduction);
}
/**
* 如果此信号量的公平设置为 true，则返回 true
*/
public boolean isFair() {
return sync instanceof FairSync;
}
/**
* 查询是否有线程正在等待获取
*/
public final boolean hasQueuedThreads() {
return sync.hasQueuedThreads();
}
/**
* 返回正在等待获取的线程的估计数目
*/
public final int getQueueLength() {
return sync.getQueueLength();
}
/**
* 返回一个 collection，包含可能等待获取的线程
*/
protected Collection<Thread> getQueuedThreads() {
return sync.getQueuedThreads();
}
应用例子
public class SemaphoreTest {
private static final int PERSON_NUM=4;
private static ExecutorService es=Executors.newFixedThreadPool(PERSON_NUM);
private static Semaphore s=new Semaphore(3,true);
public static void release(Semaphore s,String name){
s.release();
System.out.println(name+"已经离开桥了!");
}
public static void main(String[] args) {
es.execute(new Runnable(){
@Override
public void run() {
try {
s.acquire();
System.out.println("甲上桥了！");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
es.execute(new Runnable(){
@Override
public void run() {
try {
s.acquire();
System.out.println("乙上桥了！");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
es.execute(new Runnable(){
@Override
public void run() {
try {
s.acquire();
System.out.println("丙上桥了！");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
es.execute(new Runnable(){
@Override
public void run() {
try {
release(s,"甲");
s.acquire();
System.out.println("丁上桥了！");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}
结果：
甲上桥了！
乙上桥了！
丙上桥了！
甲已经离开桥了!
丁上桥了！
解释：虽然线程池里面有甲乙丙丁四个线程准备过桥Semaphore，但是Semaphore只能让三个人过桥，所以，甲没离开桥上时，丁是无法上桥的，所以甲离开之后，丁就可以上桥了。
阅读总结：
① Semaphore采用AQS的共享锁来实现自定义同步器。
② Semaphore控制并发线程数采用的是原子的方式控制同步状态值来实现。
③ 控制并发线程的数量就直接自旋+cas来更新同步状态值即可。
---------------------------该源码为jdk1.7版本的
