java源码 - Semaphore
开篇
Semaphore是一种基于计数的信号量。它可以设定一个阈值，基于此，多个线程竞争获取许可信号，做完自己的申请后归还，超过阈值后，线程申请许可信号将会被阻塞。
Semaphore可以用来构建一些对象池，资源池之类的，比如数据库连接池，我们也可以创建计数为1的Semaphore，将其作为一种类似互斥锁的机制，这也叫二元信号量，表示两种互斥状态。
从源码角度来看，Semaphore的实现方式和CountDownLatch非常相似，基于AQS做了一些定制。通过维持AQS的锁全局计数state字段来实现定量锁的加锁和解锁操作。
Semaphore类定义
- Semaphore类内部的Sync继承自AQS，作为Semaphore的公平锁和非公平锁的基类。
- Semaphore类内部的NonfairSync继承自Sync类，通过非公平的方法加锁和解锁。
- Semaphore类内部的FairSync继承自Sync类，通过公平的方法加锁和解锁。
- Semaphore的构造函数创建FairSync或NonfairSync对象赋值给Sync。
public class Semaphore implements java.io.Serializable {
private final Sync sync;
// 继承自AQS实现抽象类Sync，作为NonfairSync和FairSync的基类。
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;
}
}
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;
// 判断是否还能获取锁，通过递减全局计数state来实现
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
}
// 构造函数
public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
// 构造函数
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
Semaphore加锁过程
- Semaphore加锁过程基于AQS实现的。
- Semaphore加锁过程步骤是尝试获取锁，如果获锁成功线程继续执行，获锁失败就挂起当前线程。
- Semaphore加锁过程因为公平锁和非公平锁略有不同，但是大流程是一致的。
- tryAcquireShared()方法是尝试的操作，doAcquireSharedInterruptibly()是获锁失败的操作。
- tryAcquireShared()方法通过递减锁全局技术变量state来判定是否能获取锁。
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);
}
protected int tryAcquireShared(int acquires) {
for (;;) {
if (hasQueuedPredecessors())
return -1;
// 通过递减锁全局技术变量state来判定是否能获取锁
// 值小于0说明获锁失败，否则代表获锁成功
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
Semaphore解锁过程
- Semaphore解锁过程基于AQS实现的。
- Semaphore加锁过程步骤是尝试释放锁，释放成功后就唤醒其他等待线程。
- Semaphore的tryReleaseShared()方法尝试释放锁，doReleaseShared()方法唤醒休眠等待线程。
- Semaphore的tryReleaseShared()方法通过递增锁的全局计数state来实现。
public void release() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
protected final boolean tryReleaseShared(int releases) {
for (;;) {
// 释放锁，通过累加全局计数state来实现
int current = getState();
int next = current + releases;
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
if (compareAndSetState(current, next))
return true;
}
}
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
