Semaphore是计数信号量。Semaphore管理一系列许可证。每一个acquire方法阻塞,直到有一个许可证能够得到而后拿走一个许可证;每一个release方法增长一个许可证,这可能会释放一个阻塞的acquire方法。然而,其实并无实际的许可证这个对象,Semaphore只是维持了一个可得到许可证的数量。java
Semaphore能够用于作流量控制,特别公用资源有限的应用场景,好比数据库链接。假若有一个需求,要读取几万个文件的数据,由于都是IO密集型任务,咱们能够启动几十个线程并发的读取,可是若是读到内存后,还须要存储到数据库中,而数据库的链接数只有10个,这时咱们必须控制只有十个线程同时获取数据库链接保存数据,不然会报错没法获取数据库链接。这个时候,咱们就可使用Semaphore来作流控,代码以下:数据库
package org.java.base.thread;并发
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;idepublic class SemaphoreTest {ui
private static final int THREAD_COUNT = 30;spa
private static ExecutorService threadPool = Executors
.newFixedThreadPool(THREAD_COUNT);.netprivate static Semaphore s = new Semaphore(10);线程
public static void main(String[] args) {
for (int i = 0; i < THREAD_COUNT; i++) {
threadPool.execute(new Runnable() {
@Override
public void run() {
try {
s.acquire();
System.out.println(“save data”);
s.release();
} catch (InterruptedException e) {
}
}
});
}codethreadPool.shutdown();
}
}对象
在代码中,虽然有30个线程在执行,可是只容许10个并发的执行。Semaphore的构造方法Semaphore(int permits) 接受一个整型的数字,表示可用的许可证数量。Semaphore(10)表示容许10个线程获取许可证,也就是最大并发数是10。Semaphore的用法也很简单,首先线程使用Semaphore的acquire()获取一个许可证,使用完以后调用release()归还许可证。还能够用tryAcquire()方法尝试获取许可证。
Semaphore还提供一些其余方法:
Semaphore有两种模式,公平模式和非公平模式。公平模式就是调用acquire的顺序就是获取许可证的顺序,遵循FIFO;而非公平模式是抢占式的,也就是有可能一个新的获取线程刚好在一个许可证释放时获得了这个许可证,而前面还有等待的线程。
Semaphore有两个构造方法,以下:
public Semaphore(int permits) { sync = new NonfairSync(permits); } public Semaphore(int permits, boolean fair) { sync = fair ? new FairSync(permits) : new NonfairSync(permits); }
从上面能够看到两个构造方法,都必须提供许可的数量,第二个构造方法能够指定是公平模式仍是非公平模式,默认非公平模式。
Semaphore内部基于AQS的共享模式,因此实现都委托给了Sync类。
这里就看一下NonfairSync的构造方法:
NonfairSync(int permits) { super(permits); }
能够看到直接调用了父类的构造方法,Sync的构造方法以下:
Sync(int permits) { setState(permits); }
能够看到调用了setState方法,也就是说AQS中的资源就是许可证的数量。
先从获取一个许可看起,而且先看非公平模式下的实现。首先看acquire方法,acquire方法有几个重载,但主要是下面这个方法
public void acquire(int permits) throws InterruptedException { if (permits < 0) throw new IllegalArgumentException(); sync.acquireSharedInterruptibly(permits); }
从上面能够看到,调用了Sync的acquireSharedInterruptibly方法,该方法在父类AQS中,以下:
public final void acquireSharedInterruptibly(int arg) throws InterruptedException { //若是线程被中断了,抛出异常 if (Thread.interrupted()) throw new InterruptedException(); //获取许可失败,将线程加入到等待队列中 if (tryAcquireShared(arg) < 0) doAcquireSharedInterruptibly(arg); }
AQS子类若是要使用共享模式的话,须要实现tryAcquireShared方法,下面看NonfairSync的该方法实现:
protected int tryAcquireShared(int acquires) { return nonfairTryAcquireShared(acquires); }
该方法调用了父类中的nonfairTyAcquireShared方法,以下:
final int nonfairTryAcquireShared(int acquires) { for (;;) { //获取剩余许可数量 int available = getState(); //计算给完此次许可数量后的个数 int remaining = available - acquires; //若是许可不够或者能够将许可数量重置的话,返回 if (remaining < 0 || compareAndSetState(available, remaining)) return remaining; } }
从上面能够看到,只有在许可不够时返回值才会小于0,其他返回的都是剩余许可数量,这也就解释了,一旦许可不够,后面的线程将会阻塞。看完了非公平的获取,再看下公平的获取,代码以下:
protected int tryAcquireShared(int acquires) { for (;;) { //若是前面有线程再等待,直接返回-1 if (hasQueuedPredecessors()) return -1; //后面与非公平同样 int available = getState(); int remaining = available - acquires; if (remaining < 0 || compareAndSetState(available, remaining)) return remaining; } }
从上面能够看到,FairSync与NonFairSync的区别就在于会首先判断当前队列中有没有线程在等待,若是有,就老老实实进入到等待队列;而不像NonfairSync同样首先试一把,说不定就刚好得到了一个许可,这样就能够插队了。
看完了获取许可后,再看一下释放许可。
释放许可也有几个重载方法,但都会调用下面这个带参数的方法,
public void release(int permits) { if (permits < 0) throw new IllegalArgumentException(); sync.releaseShared(permits); }
releaseShared方法在AQS中,以下:
public final boolean releaseShared(int arg) { //若是改变许可数量成功 if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; }
AQS子类实现共享模式的类须要实现tryReleaseShared类来判断是否释放成功,实现以下:
protected final boolean tryReleaseShared(int releases) { for (;;) { //获取当前许可数量 int current = getState(); //计算回收后的数量 int next = current + releases; if (next < current) // overflow throw new Error("Maximum permit count exceeded"); //CAS改变许可数量成功,返回true if (compareAndSetState(current, next)) return true; } }
从上面能够看到,一旦CAS改变许可数量成功,那么就会调用doReleaseShared()方法释放阻塞的线程。
Semaphore还有减少许可数量的方法,该方法能够用于用于当资源用完不能再用时,这时就能够减少许可证。代码以下:
protected void reducePermits(int reduction) { if (reduction < 0) throw new IllegalArgumentException(); sync.reducePermits(reduction); }
能够看到,委托给了Sync,Sync的reducePermits方法以下:
final void reducePermits(int reductions) { for (;;) { //获得当前剩余许可数量 int current = getState(); //获得减完以后的许可数量 int next = current - reductions; if (next > current) // underflow throw new Error("Permit count underflow"); //若是CAS改变成功 if (compareAndSetState(current, next)) return; } }
从上面能够看到,就是CAS改变AQS中的state变量,由于该变量表明许可证的数量。
Semaphore还能够一次将剩余的许可数量所有取走,该方法是drain方法,以下:
public int drainPermits() { return sync.drainPermits(); }
Sync的实现以下:
final int drainPermits() { for (;;) { int current = getState(); if (current == 0 || compareAndSetState(current, 0)) return current; } }
能够看到,就是CAS将许可数量置为0。
Semaphore是信号量,用于管理一组资源。其内部是基于AQS的共享模式,AQS的状态表示许可证的数量,在许可证数量不够时,线程将会被挂起;而一旦有一个线程释放一个资源,那么就有可能从新唤醒等待队列中的线程继续执行。