Semaphore就是一个信号量，它的作用是限制某段代码块的并发数，首先我们来看下它的用法：
public static void main(String[] args) {
int N = 8; //工人数
Semaphore semaphore = new Semaphore(5); //机器数目
for(int i=0;i<N;i++)
new Worker(i,semaphore).start();
}
static class Worker extends Thread{
private int num;
private Semaphore semaphore;
public Worker(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();
}
}
}
不难看出Semaphore比较适合做限流，实际上一些开源框架也确实是基于它来做限流器的。
同ReentrantLock类似，Semaphore中也维护了一个Sync属性，用来做加锁解锁操作
可以看出有两个比较关键的方法acquire 以及release。
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); // 没拿到锁则入队列并阻塞
}
protected int tryAcquireShared(int acquires) {
for (;;) {
// 这里是针对公平锁 如果有前驱动节点 直接返回加锁失败
if (hasQueuedPredecessors())
return -1;
// 在初始化Semaphore的时候 会传入state的初始值，也就是总共可以被获取的量
int available = getState();
int remaining = available - acquires; // 减去需要的量 则为剩余的量
if (remaining < 0 || // 小于0 说明不足
compareAndSetState(available, remaining)) // cas成功 则抢到了需要的量
return remaining; // 返回
}
}
release
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 (;;) {
int current = getState();
int next = current + releases; // 当前state 加上 释放的量 为新的总量
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
if (compareAndSetState(current, next)) // cas 将state设置回去
return true;
}
}
AQS共享模式
上面有几个关键方法没有看， 那些都是在AQS中实现的方法可以来看下：
doAcquireSharedInterruptibly：
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted()) // Interruptibly结尾的方法表示中断会抛出异常
throw new InterruptedException();
if (tryAcquireShared(arg) < 0) // 子类实现 执行拿锁逻辑
doAcquireSharedInterruptibly(arg); // 拿锁失败 则执行该方法
}
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) { // 只有前驱节点是head的时候 才去尝试拿锁
int r = tryAcquireShared(arg); // 拿锁 并返回剩余的锁数量
if (r >= 0) {
// 设置头节点 并且传播
// 这里不仅仅是将本节点设置为头节点 同时也会唤醒后继节点继续拿锁（满足条件的话）
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) && // 判断是否应该阻塞
parkAndCheckInterrupt()) // 阻塞自己 并且如果被中断 返回true
// 由于是Interruptibly方法，所以这里会跑出一场 而不是设置中断标识
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // 将老的头节点记录下
setHead(node); // 执行普通的头节点出队操作
/*
* Try to signal next queued node if:
* Propagation was indicated by caller,
* or was recorded (as h.waitStatus either before
* or after setHead) by a previous operation
* (note: this uses sign-check of waitStatus because
* PROPAGATE status may transition to SIGNAL.)
* and
* The next node is waiting in shared mode,
* or we don't know, because it appears null
*
* The conservatism in both of these checks may cause
* unnecessary wake-ups, but only when there are multiple
* racing acquires/releases, so most need signals now or soon
* anyway.
*/
// 这里的判断条件看起来是复杂了 不过是由于之前出现过bug才这么写的 可以看后记
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared()) // 如果后继节点也是共享节点的话
doReleaseShared(); // 唤醒后继节点
}
}
doReleaseShard:
private void doReleaseShared() {
/*
* Ensure that a release propagates, even if there are other
* in-progress acquires/releases. This proceeds in the usual
* way of trying to unparkSuccessor of head if it needs
* signal. But if it does not, status is set to PROPAGATE to
* ensure that upon release, propagation continues.
* Additionally, we must loop in case a new node is added
* while we are doing this. Also, unlike other uses of
* unparkSuccessor, we need to know if CAS to reset status
* fails, if so rechecking.
*/
for (;;) {
Node h = head; // 拿到当前的头节点
if (h != null && h != tail) { // 队列必须非空
int ws = h.waitStatus;
if (ws == Node.SIGNAL) { // 如果头节点状态为可唤醒（头节点的后继节点可唤醒）
// 将状态设置为0 设置失败则下一次循环
// 这里是由于多线程操作 可能别的线程吧状态给改了 导致这边cas失败
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h); // cas成功的话 则唤醒后继节点
}
// 如果h的状态已经是0了 可能是新入队了头节点 但是还没来得及将状态改为SIGNAL
// 此时 尝试将h的状态改为PROPAGATE 具体原因见后记
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
// 如果到这里 h还是头节点 那就说明没有别的线程对head做操作 则break掉
if (h == head) // loop if head changed
break;
}
}
后记
一直搞不明白 doReleaseShared中将头节点设置为PROPAGATE 有什么作用，也没有找到对PROPAGATE在别的地方有什么引用，所以去查了下
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
参看：https://blog.csdn.net/zy353003874/article/details/110535122
