基于AQS构建CountDownLatch、CyclicBarrier和Semaphore
前言
本文先用 CountDownLatch 将共享模式说清楚，而后顺着把其余 AQS 相关的类 CyclicBarrier、Semaphore 的源码一块儿过一下。java
CountDownLatch
CountDownLatch 这个类是比较典型的 AQS 的共享模式的使用，这是一个高频使用的类。latch 的中文意思是门栓、栅栏，具体怎么解释我就不废话了，你们随意，看两个例子就知道在哪里用、怎么用了。node
- 使用例子
咱们看下 Doug Lea 在 java doc 中给出的例子，这个例子很是实用，咱们常常会写这个代码。架构
假设咱们有 N ( N > 0 ) 个任务，那么咱们会用 N 来初始化一个 CountDownLatch，而后将这个 latch 的引用传递到各个线程中，在每一个线程完成了任务后，调用 latch.countDown() 表明完成了一个任务。并发
调用 latch.await() 的方法的线程会阻塞，直到全部的任务完成。app
class Driver2 { // ...
void main() throws InterruptedException {
CountDownLatch doneSignal = new CountDownLatch(N);
Executor e = Executors.newFixedThreadPool(8);
// 建立 N 个任务，提交给线程池来执行
for (int i = 0; i < N; ++i) // create and start threads
e.execute(new WorkerRunnable(doneSignal, i));
// 等待全部的任务完成，这个方法才会返回
doneSignal.await(); // wait for all to finish
}
}
class WorkerRunnable implements Runnable {
private final CountDownLatch doneSignal;
private final int i;
WorkerRunnable(CountDownLatch doneSignal, int i) {
this.doneSignal = doneSignal;
this.i = i;
}
public void run() {
try {
doWork(i);
// 这个线程的任务完成了，调用 countDown 方法
doneSignal.countDown();
} catch (InterruptedException ex) {
} // return;
}
void doWork() { ...}
}
因此说 CountDownLatch 很是实用，咱们经常会将一个比较大的任务进行拆分，而后开启多个线程来执行，等全部线程都执行完了之后，再往下执行其余操做。这里例子中，只有 main 线程调用了 await 方法。ide
咱们再来看另外一个例子，这个例子很典型，用了两个 CountDownLatch：oop
class Driver { // ...
void main() throws InterruptedException {
CountDownLatch startSignal = new CountDownLatch(1);
CountDownLatch doneSignal = new CountDownLatch(N);
for (int i = 0; i < N; ++i) // create and start threads
new Thread(new Worker(startSignal, doneSignal)).start();
// 这边插入一些代码，确保上面的每一个线程先启动起来，才执行下面的代码。
doSomethingElse(); // don't let run yet
// 由于这里 N == 1，因此，只要调用一次，那么全部的 await 方法均可以经过
startSignal.countDown(); // let all threads proceed
doSomethingElse();
// 等待全部任务结束
doneSignal.await(); // wait for all to finish
}
}
class Worker implements Runnable {
private final CountDownLatch startSignal;
private final CountDownLatch doneSignal;
Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
this.startSignal = startSignal;
this.doneSignal = doneSignal;
}
public void run() {
try {
// 为了让全部线程同时开始任务，咱们让全部线程先阻塞在这里
// 等你们都准备好了，再打开这个门栓
startSignal.await();
doWork();
doneSignal.countDown();
} catch (InterruptedException ex) {
} // return;
}
void doWork() { ...}
}
这个例子中，doneSignal 同第一个例子的使用，咱们说说这里的 startSignal。N 个新开启的线程都调用了startSignal.await() 进行阻塞等待，它们阻塞在栅栏上，只有当条件知足的时候（startSignal.countDown()），它们才能同时经过这个栅栏。若是始终只有一个线程调用 await 方法等待任务完成，那么 CountDownLatch 就会简单不少，因此以后的源码分析读者必定要在脑海中构建出这么一个场景：有 m 个线程是作任务的，有 n 个线程在某个栅栏上等待这 m 个线程作完任务，直到全部 m 个任务完成后，n 个线程同时经过栅栏。源码分析
- 源码分析
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
// 老套路了，内部封装一个 Sync 类继承自 AQS
private static final class Sync extends AbstractQueuedSynchronizer {
Sync(int count) {
// 这样就 state == count 了
setState(count);
}
...
}
代码都是套路，先分析套路：AQS 里面的 state 是一个整数值，这边用一个 int count 参数其实初始化就是设置了这个值，全部调用了 await 方法的等待线程会挂起，而后有其余一些线程会作 state = state - 1 操做，当 state 减到 0 的同时，那个线程会负责唤醒调用了 await 方法的全部线程。都是套路啊，只是 Doug Lea 的套路很深，代码很巧妙，否则咱们也没有要分析源码的必要。ui
对于 CountDownLatch，咱们仅仅须要关心两个方法，一个是 countDown() 方法，另外一个是 await() 方法。countDown() 方法每次调用都会将 state 减 1，直到 state 的值为 0；而 await 是一个阻塞方法，当 state 减为 0 的时候，await 方法才会返回。await 能够被多个线程调用，读者这个时候脑子里要有个图：全部调用了 await 方法的线程阻塞在 AQS 的阻塞队列中，等待条件知足（state == 0），将线程从队列中一个个唤醒过来。this
咱们用如下程序来分析源码，t1 和 t2 负责调用 countDown() 方法，t3 和 t4 调用 await 方法阻塞：
public class CountDownLatchDemo {
public static void main(String[] args) {
CountDownLatch latch = new CountDownLatch(2);
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(5000);
} catch (InterruptedException ignore) {
}
// 休息 5 秒后(模拟线程工做了 5 秒)，调用 countDown()
latch.countDown();
}
}, "t1");
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(10000);
} catch (InterruptedException ignore) {
}
// 休息 10 秒后(模拟线程工做了 10 秒)，调用 countDown()
latch.countDown();
}
}, "t2");
t1.start();
t2.start();
Thread t3 = new Thread(new Runnable() {
@Override
public void run() {
try {
// 阻塞，等待 state 减为 0
latch.await();
System.out.println("线程 t3 从 await 中返回了");
} catch (InterruptedException e) {
System.out.println("线程 t3 await 被中断");
Thread.currentThread().interrupt();
}
}
}, "t3");
Thread t4 = new Thread(new Runnable() {
@Override
public void run() {
try {
// 阻塞，等待 state 减为 0
latch.await();
System.out.println("线程 t4 从 await 中返回了");
} catch (InterruptedException e) {
System.out.println("线程 t4 await 被中断");
Thread.currentThread().interrupt();
}
}
}, "t4");
t3.start();
t4.start();
}
}
上述程序，大概在过了 10 秒左右的时候，会输出：
线程 t3 从 await 中返回了 线程 t4 从 await 中返回了 // 这两条输出，顺序不是绝对的 // 后面的分析，咱们假设 t3 先进入阻塞队列
接下来，咱们按照流程一步一步走：先 await 等待，而后被唤醒，await 方法返回。
首先，咱们来看 await() 方法，它表明线程阻塞，等待 state 的值减为 0。
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
// 这也是老套路了，我在第二篇的中断那一节说过了
if (Thread.interrupted())
throw new InterruptedException();
// t3 和 t4 调用 await 的时候，state 都大于 0。
// 也就是说，这个 if 返回 true，而后往里看
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
// 只有当 state == 0 的时候，这个方法才会返回 1
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
从方法名咱们就能够看出，这个方法是获取共享锁，而且此方法是可中断的（中断的时候抛出 InterruptedException 退出这个方法）。
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
// 1. 入队
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
// 同上，只要 state 不等于 0，那么这个方法返回 -1
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
// 2
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
咱们再一步步看具体的流程。首先，咱们看 countDown() 方法:
public void countDown() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
// 只有当 state 减为 0 的时候，tryReleaseShared 才返回 true
// 不然只是简单的 state = state - 1 那么 countDown 方法就结束了
if (tryReleaseShared(arg)) {
// 唤醒 await 的线程
doReleaseShared();
return true;
}
return false;
}
// 这个方法很简单，用自旋的方法实现 state 减 1
protected boolean tryReleaseShared(int releases) {
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
countDown 方法就是每次调用都将 state 值减 1，若是 state 减到 0 了，那么就调用下面的方法进行唤醒阻塞队列中的线程：
// 调用这个方法的时候，state == 0
// 这个方法先不要看全部的代码，按照思路往下到我写注释的地方，其余的以后还会仔细分析
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
// t3 入队的时候，已经将头节点的 waitStatus 设置为 Node.SIGNAL（-1） 了
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
// 就是这里，唤醒 head 的后继节点，也就是阻塞队列中的第一个节点
// 在这里，也就是唤醒 t3
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) // todo
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
一旦 t3 被唤醒后，咱们继续回到 await 的这段代码，parkAndCheckInterrupt 返回，咱们先不考虑中断的状况：
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); // 2. 这里是下一步
p.next = null; // help GC
failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) &&
// 1. 唤醒后这个方法返回
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
接下来，t3 会进到 setHeadAndPropagate(node, r) 这个方法，先把 head 给占了，而后唤醒队列中其余的线程：
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
setHead(node);
// 下面说的是，唤醒当前 node 以后的节点，即 t3 已经醒了，立刻唤醒 t4
// 相似的，若是 t4 后面还有 t5，那么 t4 醒了之后，立刻将 t5 给唤醒了
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
if (s == null || s.isShared())
// 又是这个方法，只是如今的 head 已经不是原来的空节点了，是 t3 的节点了
doReleaseShared();
}
}
又回到这个方法了，那么接下来，咱们好好分析 doReleaseShared 这个方法，咱们根据流程，头节点 head 此时是 t3 节点了：
// 调用这个方法的时候，state == 0
private void doReleaseShared() {
for (;;) {
Node h = head;
// 1. h == null: 说明阻塞队列为空
// 2. h == tail: 说明头结点多是刚刚初始化的头节点，
// 或者是普通线程节点，可是此节点既然是头节点了，那么表明已经被唤醒了，阻塞队列没有其余节点了
// 因此这两种状况不须要进行唤醒后继节点
if (h != null && h != tail) {
int ws = h.waitStatus;
// t4 将头节点(此时是 t3)的 waitStatus 设置为 Node.SIGNAL（-1） 了
if (ws == Node.SIGNAL) {
// 这里 CAS 失败的场景请看下面的解读
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
// 就是这里，唤醒 head 的后继节点，也就是阻塞队列中的第一个节点
// 在这里，也就是唤醒 t4
unparkSuccessor(h);
}
else if (ws == 0 &&
// 这个 CAS 失败的场景是：执行到这里的时候，恰好有一个节点入队，入队会将这个 ws 设置为 -1
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
// 若是到这里的时候，前面唤醒的线程已经占领了 head，那么再循环
// 不然，就是 head 没变，那么退出循环，
// 退出循环是否是意味着阻塞队列中的其余节点就不唤醒了？固然不是，唤醒的线程以后仍是会调用这个方法的
if (h == head) // loop if head changed
break;
}
}
咱们分析下最后一个 if 语句，而后才能解释第一个 CAS 为何可能会失败：
h == head：说明头节点尚未被刚刚用 unparkSuccessor 唤醒的线程（这里能够理解为 t4）占有，此时 break 退出循环。
h != head：头节点被刚刚唤醒的线程（这里能够理解为 t4）占有，那么这里从新进入下一轮循环，唤醒下一个节点（这里是 t4 ）。咱们知道，等到 t4 被唤醒后，实际上是会主动唤醒 t五、t六、t7...，那为何这里要进行下一个循环来唤醒 t5 呢？我以为是出于吞吐量的考虑。
知足上面的 2 的场景，那么咱们就能知道为何上面的 CAS 操做 compareAndSetWaitStatus(h, Node.SIGNAL, 0) 会失败了？
由于当前进行 for 循环的线程到这里的时候，可能刚刚唤醒的线程 t4 也刚恰好到这里了，那么就有可能 CAS 失败了。
for 循环第一轮的时候会唤醒 t4，t4 醒后会将本身设置为头节点，若是在 t4 设置头节点后，for 循环才跑到 if (h == head)，那么此时会返回 false，for 循环会进入下一轮。t4 唤醒后也会进入到这个方法里面，那么 for 循环第二轮和 t4 就有可能在这个 CAS 相遇，那么就只会有一个成功了。
CyclicBarrier
字面意思是“可重复使用的栅栏”，CyclicBarrier 相比 CountDownLatch 来讲，要简单不少，其源码没有什么高深的地方，它是 ReentrantLock 和 Condition 的组合使用。看以下示意图，CyclicBarrier 和 CountDownLatch 是否是很像，只是 CyclicBarrier 能够有不止一个栅栏，由于它的栅栏（Barrier）能够重复使用（Cyclic）。
首先，CyclicBarrier 的源码实现和 CountDownLatch 截然不同，CountDownLatch 基于 AQS 的共享模式的使用，而 CyclicBarrier 基于 Condition 来实现。
由于 CyclicBarrier 的源码相对来讲简单许多，读者只要熟悉了前面关于 Condition 的分析，那么这里的源码是毫无压力的，就是几个特殊概念罢了。
废话结束，先上基本属性和构造方法：
public class CyclicBarrier {
// 咱们说了，CyclicBarrier 是能够重复使用的，咱们把每次从开始使用到穿过栅栏当作"一代"
private static class Generation {
boolean broken = false;
}
/** The lock for guarding barrier entry */
private final ReentrantLock lock = new ReentrantLock();
// CyclicBarrier 是基于 Condition 的
// Condition 是“条件”的意思，CyclicBarrier 的等待线程经过 barrier 的“条件”是你们都到了栅栏上
private final Condition trip = lock.newCondition();
// 参与的线程数
private final int parties;
// 若是设置了这个，表明越过栅栏以前，要执行相应的操做
private final Runnable barrierCommand;
// 当前所处的“代”
private Generation generation = new Generation();
// 尚未到栅栏的线程数，这个值初始为 parties，而后递减
// 尚未到栅栏的线程数 = parties - 已经到栅栏的数量
private int count;
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
public CyclicBarrier(int parties) {
this(parties, null);
}
我用一图来描绘下 CyclicBarrier 里面的一些概念：
看图咱们也知道了，CyclicBarrier 的源码最重要的就是 await() 方法了。
首先，先看怎么开启新的一代：
// 开启新的一代，当最后一个线程到达栅栏上的时候，调用这个方法来唤醒其余线程，同时初始化“下一代”
private void nextGeneration() {
// 首先，须要唤醒全部的在栅栏上等待的线程
trip.signalAll();
// 更新 count 的值
count = parties;
// 从新生成“新一代”
generation = new Generation();
}
看看怎么打破一个栅栏：
private void breakBarrier() {
// 设置状态 broken 为 true
generation.broken = true;
// 重置 count 为初始值 parties
count = parties;
// 唤醒全部已经在等待的线程
trip.signalAll();
}
这两个方法以后用获得，如今开始分析最重要的等待经过栅栏方法 await 方法：
// 不带超时机制
public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
// 带超时机制，若是超时抛出 TimeoutException 异常
public int await(long timeout, TimeUnit unit)
throws InterruptedException,
BrokenBarrierException,
TimeoutException {
return dowait(true, unit.toNanos(timeout));
}
继续往里看：
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
final ReentrantLock lock = this.lock;
// 先要获取到锁，而后在 finally 中要记得释放锁
// 若是记得 Condition 部分的话，咱们知道 condition 的 await 会释放锁，signal 的时候须要从新获取锁
lock.lock();
try {
final Generation g = generation;
// 检查栅栏是否被打破，若是被打破，抛出 BrokenBarrierException 异常
if (g.broken)
throw new BrokenBarrierException();
// 检查中断状态，若是中断了，抛出 InterruptedException 异常
if (Thread.interrupted()) {
breakBarrier();
throw new InterruptedException();
}
// index 是这个 await 方法的返回值
// 注意到这里，这个是从 count 递减后获得的值
int index = --count;
// 若是等于 0，说明全部的线程都到栅栏上了，准备经过
if (index == 0) { // tripped
boolean ranAction = false;
try {
// 若是在初始化的时候，指定了经过栅栏前须要执行的操做，在这里会获得执行
final Runnable command = barrierCommand;
if (command != null)
command.run();
// 若是 ranAction 为 true，说明执行 command.run() 的时候，没有发生异常退出的状况
ranAction = true;
// 唤醒等待的线程，而后开启新的一代
nextGeneration();
return 0;
} finally {
if (!ranAction)
// 进到这里，说明执行指定操做的时候，发生了异常，那么须要打破栅栏
// 以前咱们说了，打破栅栏意味着唤醒全部等待的线程，设置 broken 为 true，重置 count 为 parties
breakBarrier();
}
}
// loop until tripped, broken, interrupted, or timed out
// 若是是最后一个线程调用 await，那么上面就返回了
// 下面的操做是给那些不是最后一个到达栅栏的线程执行的
for (;;) {
try {
// 若是带有超时机制，调用带超时的 Condition 的 await 方法等待，直到最后一个线程调用 await
if (!timed)
trip.await();
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
// 若是到这里，说明等待的线程在 await（是 Condition 的 await）的时候被中断
if (g == generation && ! g.broken) {
// 打破栅栏
breakBarrier();
// 打破栅栏后，从新抛出这个 InterruptedException 异常给外层调用的方法
throw ie;
} else {
// 到这里，说明 g != generation, 说明新的一代已经产生，即最后一个线程 await 执行完成，
// 那么此时没有必要再抛出 InterruptedException 异常，记录下来这个中断信息便可
// 或者是栅栏已经被打破了，那么也不该该抛出 InterruptedException 异常，
// 而是以后抛出 BrokenBarrierException 异常
Thread.currentThread().interrupt();
}
}
// 唤醒后，检查栅栏是不是“破的”
if (g.broken)
throw new BrokenBarrierException();
// 这个 for 循环除了异常，就是要从这里退出了
// 咱们要清楚，最后一个线程在执行完指定任务(若是有的话)，会调用 nextGeneration 来开启一个新的代
// 而后释放掉锁，其余线程从 Condition 的 await 方法中获得锁并返回，而后到这里的时候，其实就会知足 g != generation 的
// 那何时不知足呢？barrierCommand 执行过程当中抛出了异常，那么会执行打破栅栏操做，
// 设置 broken 为true，而后唤醒这些线程。这些线程会从上面的 if (g.broken) 这个分支抛 BrokenBarrierException 异常返回
// 固然，还有最后一种可能，那就是 await 超时，此种状况不会从上面的 if 分支异常返回，也不会从这里返回，会执行后面的代码
if (g != generation)
return index;
// 若是醒来发现超时了，打破栅栏，抛出异常
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
lock.unlock();
}
}
好了，我想我应该讲清楚了吧，我好像几乎没有漏掉任何一行代码吧？
下面开始收尾工做。
首先，咱们看看怎么获得有多少个线程到了栅栏上，处于等待状态：
public int getNumberWaiting() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return parties - count;
} finally {
lock.unlock();
}
}
判断一个栅栏是否被打破了，这个很简单，直接看 broken 的值便可：
public boolean isBroken() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return generation.broken;
} finally {
lock.unlock();
}
}
前面咱们在说 await 的时候也几乎说清楚了，何时栅栏会被打破，总结以下：
1.中断，咱们说了，若是某个等待的线程发生了中断，那么会打破栅栏，同时抛出 InterruptedException 异常；
2.超时，打破栅栏，同时抛出 TimeoutException 异常；
3.指定执行的操做抛出了异常，这个咱们前面也说过。
最后，咱们来看看怎么重置一个栅栏：
public void reset() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
breakBarrier(); // break the current generation
nextGeneration(); // start a new generation
} finally {
lock.unlock();
}
}
咱们设想一下，若是初始化时，指定了线程 parties = 4，前面有 3 个线程调用了 await 等待，在第 4 个线程调用 await 以前，咱们调用 reset 方法，那么会发生什么？
首先，打破栅栏，那意味着全部等待的线程（3个等待的线程）会唤醒，await 方法会经过抛出 BrokenBarrierException 异常返回。而后开启新的一代，重置了 count 和 generation，至关于一切归零了。
怎么样，CyclicBarrier 源码很简单吧。
Semaphore
有了 CountDownLatch 的基础后，分析 Semaphore 会简单不少。Semaphore 是什么呢？它相似一个资源池（读者能够类比线程池），每一个线程须要调用 acquire() 方法获取资源，而后才能执行，执行完后，须要 release 资源，让给其余的线程用。
大概你们也能够猜到，Semaphore 其实也是 AQS 中共享锁的使用，由于每一个线程共享一个池嘛。
套路解读：建立 Semaphore 实例的时候，须要一个参数 permits，这个基本上能够肯定是设置给 AQS 的 state 的，而后每一个线程调用 acquire 的时候，执行 state = state - 1，release 的时候执行 state = state + 1，固然，acquire 的时候，若是 state = 0，说明没有资源了，须要等待其余线程 release。
构造方法：
public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
这里和 ReentrantLock 相似，用了公平策略和非公平策略。
看 acquire 方法：
public void acquire() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public void acquireUninterruptibly() {
sync.acquireShared(1);
}
public void acquire(int permits) throws InterruptedException {
if (permits < 0) throw new IllegalArgumentException();
sync.acquireSharedInterruptibly(permits);
}
public void acquireUninterruptibly(int permits) {
if (permits < 0) throw new IllegalArgumentException();
sync.acquireShared(permits);
}
这几个方法也是老套路了，你们基本都懂了吧，这边多了两个能够传参的 acquire 方法，不过你们也都懂的吧，若是咱们须要一次获取超过一个的资源，会用得着这个的。
咱们接下来看不抛出 InterruptedException 异常的 acquireUninterruptibly() 方法吧：
public void acquireUninterruptibly() {
sync.acquireShared(1);
}
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
前面说了，Semaphore 分公平策略和非公平策略，咱们对比一下两个 tryAcquireShared 方法：
// 公平策略：
protected int tryAcquireShared(int acquires) {
for (;;) {
// 区别就在因而不是会先判断是否有线程在排队，而后才进行 CAS 减操做
if (hasQueuedPredecessors())
return -1;
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
// 非公平策略：
protected int tryAcquireShared(int acquires) {
return nonfairTryAcquireShared(acquires);
}
final int nonfairTryAcquireShared(int acquires) {
for (;;) {
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
也是老套路了，因此从源码分析角度的话，咱们其实不太须要关心是否是公平策略仍是非公平策略，它们的区别每每就那么一两行。
咱们再回到 acquireShared 方法，
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
因为 tryAcquireShared(arg) 返回小于 0 的时候，说明 state 已经小于 0 了（没资源了），此时 acquire 不能立马拿到资源，须要进入到阻塞队列等待，虽然贴了不少代码，不在意多这点了：
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);
}
}
这个方法我就不介绍了，线程挂起后等待有资源被 release 出来。接下来，咱们就要看 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;
// 溢出，固然，咱们通常也不会用这么大的数
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
if (compareAndSetState(current, next))
return true;
}
}
tryReleaseShared 方法老是会返回 true，而后是 doReleaseShared，这个也是咱们熟悉的方法了，我就贴下代码，不分析了，这个方法用于唤醒全部的等待线程：
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;
}
}
Semphore 的源码确实很简单，基本上都是分析过的老代码的组合使用了。
总结
写到这里，终于把 AbstractQueuedSynchronizer 基本上说完了，对于 Java 并发，Doug Lea 真的是神同样的存在。往后咱们还会接触到不少 Doug Lea 的代码，但愿咱们你们均可以朝着大神的方向不断打磨本身的技术，少一些高大上的架构，多一些实实在在的优秀代码吧。
（全文完）
- 1. CountDownLatch、CyclicBarrier、 Semaphore、ReentrantLock和AQS
- 2. J.U.C之AQS之 CountDownLatch、Semaphore、CyclicBarrier
- 3. AQS中的 CountdownLatch、CyclicBarrier、 Semaphore
- 4. CountDownLatch、CyclicBarrier和Semaphore
- 5. CountDownLatch 、CyclicBarrier和Semaphore
- 6. java高并发7.1 AQS--CountDownLatch, Semaphore, CyclicBarrier
- 7. CountDownLatch、CyclicBarrier、Semaphore
- 8. CountDownLatch&CyclicBarrier&Semaphore
- 9. CountDownLatch、Semaphore、CyclicBarrier
- 10. CountDownLatch和CyclicBarrier和Semaphore应用
- 更多相关文章...
- • Spring使用AspectJ开发AOP：基于XML和基于Annotation - Spring教程
- • Spring基于Annotation装配Bean - Spring教程
- • 适用于PHP初学者的学习线路和建议
- • ☆基于Java Instrument的Agent实现
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. 在windows下的虚拟机中,安装华为电脑的deepin操作系统
- 2. 强烈推荐款下载不限速解析神器
- 3. 【区块链技术】孙宇晨：区块链技术带来金融服务的信任变革
- 4. 搜索引起的链接分析-计算网页的重要性
- 5. TiDB x 微众银行 | 耗时降低 58%，分布式架构助力实现普惠金融
- 6. 《数字孪生体技术白皮书》重磅发布（附完整版下载）
- 7. 双十一“避坑”指南：区块链电子合同为电商交易保驾护航！
- 8. 区块链产业，怎样“链”住未来？
- 9. OpenglRipper使用教程
- 10. springcloud请求一次好用一次不好用zuul Name or service not known
- 1. CountDownLatch、CyclicBarrier、 Semaphore、ReentrantLock和AQS
- 2. J.U.C之AQS之 CountDownLatch、Semaphore、CyclicBarrier
- 3. AQS中的 CountdownLatch、CyclicBarrier、 Semaphore
- 4. CountDownLatch、CyclicBarrier和Semaphore
- 5. CountDownLatch 、CyclicBarrier和Semaphore
- 6. java高并发7.1 AQS--CountDownLatch, Semaphore, CyclicBarrier
- 7. CountDownLatch、CyclicBarrier、Semaphore
- 8. CountDownLatch&CyclicBarrier&Semaphore
- 9. CountDownLatch、Semaphore、CyclicBarrier
- 10. CountDownLatch和CyclicBarrier和Semaphore应用