1.控制并发线程数的Semaphore
Semaphore(信号量)是用来控制同时访问特定资源的线程数量,它通过协调各个线程,保证合理的使用公共资源。
线程可以通过acquire()方法来获取信号量的许可,当信号量中没有可用的许可的时候,线程阻塞,直到有可用的许可为止。线程可以通过release()方法释放它持有
的信号量的许可。
2.Semaphore的方法列表:
// 创建具有给定的许可数和非公平的公平设置的 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)
3.Semaphore的内部结构
4.Semaphore的源码:
"公平信号量"和"非公平信号量"的区别
"公平信号量"和"非公平信号量"的释放信号量的机制是一样的!不同的是它们获取信号量的机制:线程在尝试获取信号量许可时,对于公平信号量而言,如果当前线程不在CLH队列的头部,则排队等候;而对于非公平信号量而言,无论当前线程是不是在CLH队列的头部,它都会直接获取信号量。该差异具体的体现在,它们的tryAcquireShared()函数的实现不同。
公平信号量tryAcquireShared源码如下:
/** * Fair version */ 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; if (remaining < 0 || compareAndSetState(available, remaining)) return remaining; } } }
非公平信号量tryAcquireShared源码如下:
/** * NonFair version */ 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); } }
实例:
public class SemaphoreTest { private static final int THREAD_COUNT = 10; private static ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT); // 创建5个许可,允许5个并发执行 private static Semaphore s = new Semaphore(5); public static void main(String[] args) { //创建10个线程执行任务 for (int i = 0; i < THREAD_COUNT; i++) { executorService.execute(new Runnable() { @Override public void run() { try { //同时只能有5个线程并发执行保存数据的任务 s.acquire(); System.out.println("线程" + Thread.currentThread().getName() + " 保存数据"); Thread.sleep(2000); //5个线程保存完数据,释放1个许可,其他的线程才能获取许可,继续执行保存数据的任务 s.release(); System.out.println("线程" + Thread.currentThread().getName() + " 释放许可"); } catch (InterruptedException e) { e.printStackTrace(); } } }); } executorService.shutdown(); } }
结果:10个线程保存数据,但是只允许5个线程并发的执行,当5个线程都保存完数据以后,释放许可,其他线程才能拿到许可继续保存数据,直到10个线程都保存完数据释放许可为止。
线程pool-1-thread-2 保存数据 线程pool-1-thread-3 保存数据 线程pool-1-thread-1 保存数据 线程pool-1-thread-4 保存数据 线程pool-1-thread-5 保存数据 线程pool-1-thread-2 释放许可 线程pool-1-thread-6 保存数据 线程pool-1-thread-1 释放许可 线程pool-1-thread-3 释放许可 线程pool-1-thread-4 释放许可 线程pool-1-thread-9 保存数据 线程pool-1-thread-10 保存数据 线程pool-1-thread-5 释放许可 线程pool-1-thread-8 保存数据 线程pool-1-thread-7 保存数据 线程pool-1-thread-10 释放许可 线程pool-1-thread-9 释放许可 线程pool-1-thread-6 释放许可 线程pool-1-thread-8 释放许可 线程pool-1-thread-7 释放许可
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);
}
这两个构造方法,都必须提供许可的数量,第二个构造方法可以指定是公平模式还是非公平模式,默认非公平模式。
在ReentrantLock中公平锁和非公平锁获取锁机制的差别:对于公平锁而言,如果当前线程不在CLH队列的头部,则需要排队等候,而非公平锁则不同,它无论当前线程处于CLH队列的何处都会直接获取锁。所以公平信号量和非公平信号量的区别也一样。
获取许可证
Semaphore类提供了4种获取许可证的方法,分别如下:
//获取一个许可证(响应中断),在没有可用的许可证时当前线程被阻塞。
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
//获取一个许可证(不响应中断)
public void acquireUninterruptibly() {
sync.acquireShared(1);
}
//尝试获取许可证(非公平获取),立即返回结果(非阻塞)。
public boolean tryAcquire() {
return sync.nonfairTryAcquireShared(1) >= 0;
}
//尝试获取许可证(定时获取)
public boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}
下面我们一起看看,acquire()是如何获取许可证的? 其源码如下:
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
acquireSharedInterruptibly方法首先就是去调用tryAcquireShared方法去尝试获取,tryAcquireShared在AQS里面是抽象方法,FairSync和NonfairSync这两个派生类实现了该方法的逻辑。FairSync实现的是公平获取的逻辑,而NonfairSync实现的非公平获取的逻辑。
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;
// 1.如果remaining小于0则直接返回remaining
// 2.如果remaining大于0则先更新同步状态再返回remaining
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
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");
if (compareAndSetState(current, next))
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");
if (compareAndSetState(current, next))
return;
}
}
final int drainPermits() {
for (;;) {
int current = getState();
if (current == 0 || compareAndSetState(current, 0))
return current;
}
}
}
// 非公平同步器
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;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
}
非公平获取锁的逻辑是先取出当前同步状态(同步状态表示许可证个数),将当前同步状态减去参入的参数,如果结果不小于0的话证明还有可用的许可证,那么就直接使用CAS操作更新同步状态的值,最后不管结果是否小于0都会返回该结果值。
公平获取锁的逻辑仅仅是在此之前先去调用hasQueuedPredecessors方法判断同步队列是否存在排队,如果有的话就直接返回-1表示获取失败,否则继续执行和非公平获取一样的步骤。
释放许可证
Semaphore类提供了2种释放许可证方法,分别如下:
public void release() {
sync.releaseShared(1);
}
public void release(int permits) {
if (permits < 0) throw new IllegalArgumentException();
sync.releaseShared(permits);
}
调用release方法是释放一个许可证,它的操作很简单,就调用了AQS的releaseShared方法。
//释放锁的操作(共享模式)
public final boolean releaseShared(int arg) {
//1.尝试去释放锁
if (tryReleaseShared(arg)) {
//2.如果释放成功就唤醒其他线程
doReleaseShared();
return true;
}
return false;
}
AQS的releaseShared方法首先调用tryReleaseShared方法尝试释放锁,其实现逻辑在子类Sync里面。
// 尝试释放操作
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;
}
}
tryReleaseShared方法里面采用for循环进行自旋,首先获取同步状态,将同步状态加上传入的参数,然后以CAS方式更新同步状态,更新成功就返回true并跳出方法,否则就继续循环直到成功为止。
动手写个连接池
下面我们就来利用Semaphore实现一个简单的数据库连接池,通过这个例子希望读者们能更加深入的掌握Semaphore的运用。
public class ConnectPool {
//连接池大小
private int size;
//数据库连接集合
private Connect[] connects;
//连接状态标志
private boolean[] connectFlag;
//剩余可用连接数
private volatile int available;
//信号量
private Semaphore semaphore;
//构造器
public ConnectPool(int size) {
this.size = size;
this.available = size;
semaphore = new Semaphore(size, true);
connects = new Connect[size];
connectFlag = new boolean[size];
initConnects();
}
//初始化连接
private void initConnects() {
//生成指定数量的数据库连接
for(int i = 0; i < this.size; i++) {
connects[i] = new Connect();
}
}
//获取数据库连接
private synchronized Connect getConnect(){
for(int i = 0; i < connectFlag.length; i++) {
//遍历集合找到未使用的连接
if(!connectFlag[i]) {
//将连接设置为使用中
connectFlag[i] = true;
//可用连接数减1
available--;
System.out.println("【"+Thread.currentThread().getName()+"】以获取连接 剩余连接数:" + available);
//返回连接引用
return connects[i];
}
}
return null;
}
//获取一个连接
public Connect openConnect() throws InterruptedException {
//获取许可证
semaphore.acquire();
//获取数据库连接
return getConnect();
}
//释放一个连接
public synchronized void release(Connect connect) {
for(int i = 0; i < this.size; i++) {
if(connect == connects[i]){
//将连接设置为未使用
connectFlag[i] = false;
//可用连接数加1
available++;
System.out.println("【"+Thread.currentThread().getName()+"】以释放连接 剩余连接数:" + available);
//释放许可证
semaphore.release();
}
}
}
//剩余可用连接数
public int available() {
return available;
}
}
测试代码:
public class TestThread extends Thread {
private static ConnectPool pool = new ConnectPool(3);
@Override
public void run() {
try {
Connect connect = pool.openConnect();
Thread.sleep(100); //休息一下
pool.release(connect);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
for(int i = 0; i < 10; i++) {
new TestThread().start();
}
}
}
使用一个数组来存放数据库连接的引用,在初始化连接池的时候会调用initConnects方法创建指定数量的数据库连接,并将它们的引用存放到数组中,此外还有一个相同大小的数组来记录连接是否可用。
每当外部线程请求获取一个连接时,首先调用semaphore.acquire()方法获取一个许可证,然后将连接状态设置为使用中,最后返回该连接的引用。许可证的数量由构造时传入的参数决定,每调用一次semaphore.acquire()方法许可证数量减1,当数量减为0时说明已经没有连接可以使用了,这时如果其他线程再来获取就会被阻塞。每当线程释放一个连接的时候会调用semaphore.release()将许可证释放,此时许可证的总量又会增加,代表可用的连接数增加了,那么之前被阻塞的线程将会醒来继续获取连接,这时再次获取就能够成功获取连接了。
Semaphore是计数信号量。Semaphore管理一系列许可证。每个acquire方法阻塞,直到有一个许可证可以获得然后拿走一个许可证;每个release方法增加一个许可证,这可能会释放一个阻塞的acquire方法。然而,其实并没有实际的许可证这个对象,Semaphore只是维持了一个可获得许可证的数量。
Semaphore经常用于限制获取某种资源的线程数量。下面举个例子,比如说操场上有5个跑道,一个跑道一次只能有一个学生在上面跑步,一旦所有跑道在使用,那么后面的学生就需要等待,直到有一个学生不跑了,下面是这个例子:
/**
* 操场,有5个跑道
* Created by Xingfeng on 2016-12-09.
*/
public class Playground {
/**
* 跑道类
*/
static class Track {
private int num;
public Track(int num) {
this.num = num;
}
@Override
public String toString() {
return "Track{" +
"num=" + num +
'}';
}
}
private Track[] tracks = {
new Track(1), new Track(2), new Track(3), new Track(4), new Track(5)};
private volatile boolean[] used = new boolean[5];
private Semaphore semaphore = new Semaphore(5, true);
/**
* 获取一个跑道
*/
public Track getTrack() throws InterruptedException {
semaphore.acquire(1);
return getNextAvailableTrack();
}
/**
* 返回一个跑道
*
* @param track
*/
public void releaseTrack(Track track) {
if (makeAsUsed(track))
semaphore.release(1);
}
/**
* 遍历,找到一个没人用的跑道
*
* @return
*/
private Track getNextAvailableTrack() {
for (int i = 0; i < used.length; i++) {
if (!used[i]) {
used[i] = true;
return tracks[i];
}
}
return null;
}
/**
* 返回一个跑道
*
* @param track
*/
private boolean makeAsUsed(Track track) {
for (int i = 0; i < used.length; i++) {
if (tracks[i] == track) {
if (used[i]) {
used[i] = false;
return true;
} else {
return false;
}
}
}
return false;
}
}
从上面可以看到,创建了5个跑道对象,并使用一个boolean类型的数组记录每个跑道是否被使用了,初始化了5个许可证的Semaphore,在获取跑道时首先调用acquire(1)获取一个许可证,在归还一个跑道是调用release(1)释放一个许可证。接下来再看启动程序,如下:
public class SemaphoreDemo {
static class Student implements Runnable {
private int num;
private Playground playground;
public Student(int num, Playground playground) {
this.num = num;
this.playground = playground;
}
@Override
public void run() {
try {
//获取跑道
Playground.Track track = playground.getTrack();
if (track != null) {
System.out.println("学生" + num + "在" + track.toString() + "上跑步");
TimeUnit.SECONDS.sleep(2);
System.out.println("学生" + num + "释放" + track.toString());
//释放跑道
playground.releaseTrack(track);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
Executor executor = Executors.newCachedThreadPool();
Playground playground = new Playground();
for (int i = 0; i < 100; i++) {
executor.execute(new Student(i+1,playground));
}
}
}
上面的代码中,Student类代表学生,首先获取跑道,一旦获取到就打印一句话,然后睡眠2s,然后再打印释放,最后归还跑道。
源码解析
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的状态表示许可证的数量,在许可证数量不够时,线程将会被挂起;而一旦有一个线程释放一个资源,那么就有可能重新唤醒等待队列中的线程继续执行。