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; } } 

参考文章