信号量，用来限制能通知访问共享资源的线程上限
public static void main(String[] args) {
Semaphore semaphore = new Semaphore(3);
for (int i = 0; i < 10; i++) {
new Thread(() -> {
try {
semaphore.acquire();
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
log.debug("running");
Thread.sleep(1000);
log.debug("end");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
semaphore.release();
}
}).start();
}
}
输出
2022/05/15-14:34:02.327 [Thread-0] c.Test - running 2022/05/15-14:34:02.327 [Thread-3] c.Test - running 2022/05/15-14:34:02.327 [Thread-1] c.Test - running 2022/05/15-14:34:03.331 [Thread-0] c.Test - end 2022/05/15-14:34:03.331 [Thread-1] c.Test - end 2022/05/15-14:34:03.331 [Thread-3] c.Test - end 2022/05/15-14:34:03.331 [Thread-4] c.Test - running 2022/05/15-14:34:03.331 [Thread-2] c.Test - running 2022/05/15-14:34:03.331 [Thread-5] c.Test - running 2022/05/15-14:34:04.343 [Thread-4] c.Test - end 2022/05/15-14:34:04.343 [Thread-5] c.Test - end 2022/05/15-14:34:04.343 [Thread-8] c.Test - running 2022/05/15-14:34:04.343 [Thread-2] c.Test - end 2022/05/15-14:34:04.343 [Thread-9] c.Test - running 2022/05/15-14:34:04.343 [Thread-6] c.Test - running 2022/05/15-14:34:05.355 [Thread-9] c.Test - end 2022/05/15-14:34:05.355 [Thread-6] c.Test - end 2022/05/15-14:34:05.355 [Thread-8] c.Test - end 2022/05/15-14:34:05.355 [Thread-7] c.Test - running 2022/05/15-14:34:06.367 [Thread-7] c.Test - end应用
- 使用Semaphore限流，在访问高峰期时，让请求线程阻塞，高峰期过去在释放许可，当然她只适合限制单机线程数量，而不是限制资源数（例如连接数，请对比 Tomcat LimitLatch实现）
- 用Semaphore实现简单连接池，对比【享元模式】下的实现（用wait notify），性能和可读性显然更好，注意下面的实现中线程数和数据库连接数相等。
@Slf4j(topic = "c.Pool")
class Pool {
// 1. 连接池大小
private final int poolSize;
// 2. 连接对象数组
private Connection[] connections;
// 3. 连接状态数组 0 表示空闲， 1 表示繁忙
private AtomicIntegerArray states;
private Semaphore semaphore;
// 4. 构造方法初始化
public Pool(int poolSize) {
this.poolSize = poolSize;
// 让许可数与资源数一致
this.semaphore = new Semaphore(poolSize);
this.connections = new Connection[poolSize];
this.states = new AtomicIntegerArray(new int[poolSize]);
for (int i = 0; i < poolSize; i++) {
connections[i] = new MockConnection("连接" + (i + 1));
}
}
// 5. 借连接
public Connection borrow() {// t1, t2, t3
// 获取许可
try {
semaphore.acquire(); // 没有许可的线程，在此等待
} catch (InterruptedException e) {
e.printStackTrace();
}
for (int i = 0; i < poolSize; i++) {
// 获取空闲连接
if (states.get(i) == 0) {
if (states.compareAndSet(i, 0, 1)) {
log.debug("borrow {}", connections[i]);
return connections[i];
}
}
}
// 不会执行到这里
return null;
}
// 6. 归还连接
public void free(Connection conn) {
for (int i = 0; i < poolSize; i++) {
if (connections[i] == conn) {
states.set(i, 0);
log.debug("free {}", conn);
semaphore.release();
break;
}
}
}
}
原理
1、加锁解锁流程
Semaphore有点像一个停车场，permits就好像停车位数量，当线程获得了permits就像是获得了停车位，然后停车场显示空余车位数量减一
刚开始，permits(state)为3，这时5个线程来获取资源
假设其中Thread-1，Thread-2，Thread-4 cas竞争成功，而Thread-0和Thread-3竞争失败，进入AQS队列park阻塞
这时Thread-4释放了permits，状态如下
接下来Thread-0竞争成功，permits再次设置为0，设置自己为head结点，断开原来的head结点，unpark接下来的Thread-3，但由于permits是0，因此Thread-3在尝试不成功后再次进入park状态
static final class NonfairSync extends Sync {
private static final long serialVersionUID = -2694183684443567898L;
NonfairSync(int permits) {
// permits 即 state
super(permits);
}
// Semaphore 方法, 方便阅读, 放在此处
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
// AQS 继承过来的方法, 方便阅读, 放在此处
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) {
return nonfairTryAcquireShared(acquires);
}
// Sync 继承过来的方法, 方便阅读, 放在此处
final int nonfairTryAcquireShared(int acquires) {
for (; ; ) {
int available = getState();
int remaining = available - acquires;
if (
// 如果许可已经用完, 返回负数, 表示获取失败, 进入 doAcquireSharedInterruptibly
remaining < 0 ||
// 如果 cas 重试成功, 返回正数, 表示获取成功
compareAndSetState(available, remaining)
) {
return remaining;
}
}
}
// AQS 继承过来的方法, 方便阅读, 放在此处
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) {
// 成功后本线程出队（AQS）, 所在 Node设置为 head
// 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一个节点 unpark
// 如果 head.waitStatus == 0 ==> Node.PROPAGATE
// r 表示可用资源数, 为 0 则不会继续传播
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
// 不成功, 设置上一个节点 waitStatus = Node.SIGNAL, 下轮进入 park 阻塞
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
// Semaphore 方法, 方便阅读, 放在此处
public void release() {
sync.releaseShared(1);
}
// AQS 继承过来的方法, 方便阅读, 放在此处
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
// 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");
if (compareAndSetState(current, next))
return true;
}
}
}
3、为什么要有PROPAGATE
早期有BUG
- releaseShared方法
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
- doAcquireShared
private void doAcquireShared(int arg) {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
boolean interrupted = false;
for (; ; ) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
// 这里会有空档
setHeadAndPropagate(node, r);
p.next = null; // help GC
if (interrupted)
selfInterrupt();
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
- setHeadAndPropagate
private void setHeadAndPropagate(Node node, int propagate) {
setHead(node);
// 有空闲资源
if (propagate > 0 && node.waitStatus != 0) {
Node s = node.next;
// 下一个
if (s == null || s.isShared())
unparkSuccessor(node);
}
}
- 假设存在某次循环里队队列里排队的结点状态为head(-1)->t1(-1)->t2(-1)
- 假设存在将要释放的T3和T4，释放顺序为先T3后T4
修复前版本执行流程
- T3条用releaseShared(1)，直接调用了unparkSuccessor(head)，head的等待状态从-1变成0
- T1由于T3释放信号量被唤醒，调用tryAcquireShared，假设返回值为0（获取锁成功，但没有剩余资源量）
- T4调用releaseShared(1)，此时head.waitStatus为0（此时读到的head和1为同一个head），不满足条件因此不调用unparkSuccessor(head)
- T1获取信号量成功，调用setHeadAndPropagate时，因不满足propagate>0（2的返回值也就是propagate（剩余资源量）==0），从而不会唤醒后继结点，T2线程得不到唤醒
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
// 设置自己为 head
setHead(node);
// propagate 表示有共享资源（例如共享读锁或信号量）
// 原 head waitStatus == Node.SIGNAL 或 Node.PROPAGATE
// 现在 head waitStatus == Node.SIGNAL 或 Node.PROPAGATE
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
// 如果是最后一个节点或者是等待共享读锁的节点
if (s == null || s.isShared()) {
doReleaseShared();
}
}
}
private void doReleaseShared() {
// 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一个节点 unpark
// 如果 head.waitStatus == 0 ==> Node.PROPAGATE
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;
}
}
- T3调用releaseShared()，直接调用了unparkSuccessor(head)，head的等待状态从-1变为0
- T1由于T3释放信号量被唤醒，调用tryAcquireShared，假设返回值为0（获取锁成功，单没有剩余资源量）
- T4调用releaseShared()，此时head.waitStatus为0（此时读到的head和1中为同一个head），调用doReleaseShared()将等待状态设置为PROPAGATE(-3)
- T1获取信号量成功，调用setHeadAndPropagate时，读到h.waitStatus<0，从而调用doReleaseShared()唤醒T2
