当前位置：首页 > news >正文

JUC并发—7.AQS源码分析三

news 来源：原创 2025/9/20 11:59:33

大纲

1.等待多线程完成的CountDownLatch介绍

2.CountDownLatch.await()方法源码

3.CountDownLatch.coutDown()方法源码

4.CountDownLatch总结

5.控制并发线程数的Semaphore介绍

6.Semaphore的令牌获取过程

7.Semaphore的令牌释放过程

8.同步屏障CyclicBarrier介绍

9.CyclicBarrier的await()方法源码

10.使用CountDownLatch等待注册的完成

11.使用CyclicBarrier将工作任务多线程分而治之

12.使用CyclicBarrier聚合服务接口的返回结果

13.使用Semaphore等待指定数量线程完成任务

volatile、synchronized、CAS、AQS、读写锁、锁优化和锁故障、并发集合、线程池、同步组件

1.等待多线程完成的CountDownLatch

(1)CountDownLatch的简介

(2)CountDownLatch的应用

(3)CountDownLatch的例子

(1)CountDownLatch的简介

CountDownLatch允许一个或多个线程等待其他线程完成操作。CountDownLatch提供了两个核心方法，分别是await()方法和countDown()方法。CountDownLatch.await()方法让调用线程进行阻塞进入等待状态，CountDownLatch.countDown()方法用于对计数器进行递减。

CountDownLatch在构造时需要传入一个正整数作为计数器初始值。线程每调用一次countDown()方法，都会对该计数器减一。当计数器为0时，会唤醒所有执行await()方法时被阻塞的线程。

(2)CountDownLatch的应用

应用一：

使用多线程去解析一个Excel里多个sheet的数据，每个线程解析一个sheet里的数据，等所有sheet解析完再提示处理完成。此时便可以使用CountDownLatch来实现，当然可以使用Thread.join()方法。

注意：Thread.join()方法是基于wait()和notify()来实现的。在main线程里开启一个线程A，main线程如果执行了线程A的join()方法，那么就会导致main线程被阻塞，main线程会等待线程A执行完毕才会继续往下执行。

应用二：

微服务注册中心的register-client，为了在注册线程执行成功后，才发送心跳。可以使用CountDownLatch，当然也可以使用Thread.join()方法。

应用三：

可以通过CountDownLatch实现类似并发的效果。把CountDownLatch的计数器设置为1，然后让1000个线程调用await()方法。当1000个线程初始化完成后，在main线程调用countDown()让计数器归零。这样这1000个线程就会在一个for()循环中，依次被唤醒。

(3)CountDownLatch的例子

public class CountDownLatchDemo {public static void main(String[] args) throws Exception {final CountDownLatch latch = new CountDownLatch(2);new Thread() {public void run() {try {Thread.sleep(1000);System.out.println("线程1开始执行，休眠2秒...");Thread.sleep(1000);System.out.println("线程1准备执行countDown操作...");latch.countDown();System.out.println("线程1完成执行countDown操作...");} catch (Exception e) {e.printStackTrace();}}}.start();new Thread() {public void run() {try {Thread.sleep(1000);System.out.println("线程2开始执行，休眠2秒...");Thread.sleep(1000);System.out.println("线程2准备执行countDown操作...");latch.countDown();System.out.println("线程2完成执行countDown操作...");} catch (Exception e) {e.printStackTrace();}}}.start();System.out.println("main线程准备执行countDownLatch的await操作，将会同步阻塞等待...");latch.await();System.out.println("所有线程都完成countDown操作，结束同步阻塞等待...");}
}

2.CountDownLatch.await()方法源码

(1)CountDownLatch.await()方法的阻塞流程

(2)CountDownLatch.await()方法的唤醒流程

(3)CountDownLatch.await()方法的阻塞总结

(1)CountDownLatch.await()方法的阻塞流程

CountDownLatch是基于AQS中的共享锁来实现的。从CountDownLatch的构造方法可知，CountDownLatch的count就是AQS的state。

调用CountDownLatch的await()方法时，会先调用AQS的acquireSharedInterruptibly()模版方法，然后会调用CountDownLatch的内部类Sync实现的tryAcquireShared()方法。tryAcquireShared()方法会判断state的值是否为0，如果为0，才返回1，否则返回-1。

当调用CountDownLatch内部类Sync的tryAcquireShared()方法获得的返回值是-1时，才会调用AQS的doAcquireSharedInterruptibly()方法，将当前线程封装成Node结点加入等待队列，然后挂起当前线程进行阻塞。

//A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.
public class CountDownLatch {private final Sync sync;public CountDownLatch(int count) {if (count < 0) {throw new IllegalArgumentException("count < 0");}this.sync = new Sync(count);}//Synchronization control For CountDownLatch.//Uses AQS state to represent count.private static final class Sync extends AbstractQueuedSynchronizer {Sync(int count) {setState(count);}int getCount() {return getState();}protected int tryAcquireShared(int acquires) {return (getState() == 0) ? 1 : -1;}protected boolean tryReleaseShared(int releases) {//Decrement count; signal when transition to zerofor (;;) {int c = getState();if (c == 0) {return false;}int nextc = c-1;if (compareAndSetState(c, nextc)) {return nextc == 0;}}}}//Causes the current thread to wait until the latch has counted down to zero, //unless the thread is Thread#interrupt interrupted.public void await() throws InterruptedException {//执行AQS的acquireSharedInterruptibly()方法sync.acquireSharedInterruptibly(1);}...
}public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {...//Acquires in shared mode, aborting if interrupted.//Implemented by first checking interrupt status, then invoking at least once #tryAcquireShared, returning on success.//Otherwise the thread is queued, possibly repeatedly blocking and unblocking,//invoking #tryAcquireShared until success or the thread is interrupted.public final void acquireSharedInterruptibly(int arg) throws InterruptedException {if (Thread.interrupted()) {throw new InterruptedException();}//执行CountDownLatch的内部类Sync实现的tryAcquireShared()方法，抢占共享锁if (tryAcquireShared(arg) < 0) {//执行AQS的doAcquireSharedInterruptibly()方法doAcquireSharedInterruptibly(arg);}}//Acquires in shared interruptible mode.private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {final Node node = addWaiter(Node.SHARED);//封装当前线程为Shared类型的Node结点boolean failed = true;try {//第一次循环r = -1，所以会执行AQS的shouldParkAfterFailedAcquire()方法//将node结点的有效前驱结点的状态设置为SIGNALfor (;;) {final Node p = node.predecessor();//node结点的前驱结点if (p == head) {int r = tryAcquireShared(arg);if (r >= 0) {setHeadAndPropagate(node, r);p.next = null; // help GCfailed = false;return;}}//执行shouldParkAfterFailedAcquire()方法设置node结点的前驱结点的状态为SIGNAL//执行parkAndCheckInterrupt()方法挂起当前线程if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) {throw new InterruptedException();}}} finally {if (failed) {cancelAcquire(node);}}}//Checks and updates status for a node that failed to acquire.//Returns true if thread should block. This is the main signal control in all acquire loops.private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {int ws = pred.waitStatus;if (ws == Node.SIGNAL) {//This node has already set status asking a release to signal it, so it can safely park.return true;}if (ws > 0) {//Predecessor was cancelled. Skip over predecessors and indicate retry.do {node.prev = pred = pred.prev;} while (pred.waitStatus > 0);pred.next = node;} else {//waitStatus must be 0 or PROPAGATE.  //Indicate that we need a signal, but don't park yet.  //Caller will need to retry to make sure it cannot acquire before parking.compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false;}//设置头结点和唤醒后续线程//Sets head of queue, and checks if successor may be waiting in shared mode, //if so propagating if either propagate > 0 or PROPAGATE status was set.private void setHeadAndPropagate(Node node, int propagate) {Node h = head;setHead(node);//将node结点设置为头结点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 setHead(Node node) {head = node;node.thread = null;node.prev = null;}...
}

(2)CountDownLatch.await()方法的唤醒流程

调用await()方法时，首先会将当前线程封装成Node结点并添加到等待队列中，然后在执行第一次for循环时会设置该Node结点的前驱结点状态为SIGNAL，接着在执行第二次for循环时才会将当前线程进行挂起阻塞。

当该线程后续被唤醒时，该线程又会进入下一次for循环。如果该线程对应的node结点的前驱结点是等待队列的头结点且state值已为0，那么就执行AQS的setHeadAndPropagate()方法设置头结点 + 唤醒后续线程。

其中setHeadAndPropagate()方法有两个工作(设置头结点 + 唤醒传递)：

工作一：设置当前被唤醒线程对应的结点为头结点

工作二：当满足如下这两个条件的时候需要调用doReleaseShared()方法唤醒后续的线程

条件一：propagate > 0，表示当前是共享锁，需要进行唤醒传递

条件二：s.isShared()判断当前结点为共享模式

CountDownLatch的实现中会在以下两个场景调用doReleaseShared()方法：

场景一：state为1时调用的countDown()方法会调用doReleaseShared()方法

场景二：当阻塞的线程被唤醒时，会调用setHeadAndPropagate()方法，进而调用doReleaseShared()方法，这样可以提升唤醒共享结点的速度

(3)CountDownLatch.await()方法的阻塞总结

只要state != 0，就会进行如下处理：

一.将当前线程封装成一个Node结点，然后添加到AQS的等待队列中

二.调用LockSupport.park()方法，挂起当前线程

3.CountDownLatch.coutDown()方法源码

(1)CountDownLatch.coutDown()的唤醒流程

(2)CountDownLatch.tryReleaseShared()

(3)AQS的doReleaseShared()方法

(1)CountDownLatch.coutDown()的唤醒流程

调用CountDownLatch的countDown()方法时，会先调用AQS的releaseShared()模版方法，然后会执行CountDownLatch的内部类Sync实现的tryReleaseShared()方法。

如果tryReleaseShared()方法返回true，则执行AQS的doReleaseShared()方法，通过AQS的doReleaseShared()方法唤醒共享锁模式下的等待队列中的线程。

//A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.
public class CountDownLatch {private final Sync sync;public CountDownLatch(int count) {if (count < 0) {throw new IllegalArgumentException("count < 0");}this.sync = new Sync(count);}//Synchronization control For CountDownLatch.//Uses AQS state to represent count.private static final class Sync extends AbstractQueuedSynchronizer {Sync(int count) {setState(count);}int getCount() {return getState();}protected int tryAcquireShared(int acquires) {return (getState() == 0) ? 1 : -1;}protected boolean tryReleaseShared(int releases) {//Decrement count; signal when transition to zerofor (;;) {int c = getState();if (c == 0) {return false;}int nextc = c-1;if (compareAndSetState(c, nextc)) {return nextc == 0;}}}}//Decrements the count of the latch, releasing all waiting threads if the count reaches zero.public void countDown() {//执行AQS的releaseShared()方法sync.releaseShared(1);}...
}public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {...//Releases in shared mode.  //Implemented by unblocking one or more threads if #tryReleaseShared returns true.public final boolean releaseShared(int arg) {//执行CountDownLatch的内部类Sync实现的tryReleaseShared()方法，释放共享锁if (tryReleaseShared(arg)) {//执行AQS的doReleaseShared()方法doReleaseShared();return true;}return false;}//Release action for shared mode -- signals successor and ensures propagation. //Note: For exclusive mode, release just amounts to calling unparkSuccessor of head if it needs signal.private void doReleaseShared() {for (;;) {//每次循环时头结点都会发生变化//因为调用unparkSuccessor()方法会唤醒doAcquireSharedInterruptibly()方法中阻塞的线程//然后阻塞的线程会在执行setHeadAndPropagate()方法时通过setHead()修改头结点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}//唤醒头结点的后继结点//唤醒的线程会在doAcquireSharedInterruptibly()方法中执行setHeadAndPropagate()方法修改头结点unparkSuccessor(h);} else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) {//如果ws = 0表示初始状态，则修改结点为PROPAGATE状态continue;//loop on failed CAS}}if (h == head) {//判断头结点是否有变化break;//loop if head changed}}}//Wakes up node's successor, if one exists.private void unparkSuccessor(Node node) {int ws = node.waitStatus;if (ws < 0) {compareAndSetWaitStatus(node, ws, 0);}Node s = node.next;if (s == null || s.waitStatus > 0) {s = null;for (Node t = tail; t != null && t != node; t = t.prev) {if (t.waitStatus <= 0) {s = t;}}}if (s != null) {LockSupport.unpark(s.thread);}}...
}

(2)CountDownLatch.tryReleaseShared()

从tryReleaseShared()方法可知：每次countDown()其实就是把AQS的state值减1，然后通过CAS更新state值。如果CAS设置成功，那么就判断当前state值是否为0。如果是0那么就返回true，如果不是0那么就返回false。返回true的时候会调用AQS的doReleaseShared()方法，唤醒等待队列中的线程。

(3)AQS的doReleaseShared()方法

该方法要从AQS的等待队列中唤醒头结点的后继结点，需要满足：

条件一：等待队列中要存在挂起线程的结点(h != null && h != tail)

条件二：等待队列的头结点的状态正常(h.waitStatus = Node.SIGNAL)

在共享锁模式下，state为0时需要通过唤醒传递把所有挂起的线程都唤醒。首先doReleaseShared()方法会通过for(;;)进行自旋操作，每次循环都会通过Node h = head来获取等待队列中最新的头结点，然后通过if (h == head)来判断等待队列中的头结点是否发生变化。如果没有变化，则退出自旋。

注意：在共享锁模式下，被unparkSuccessor()唤醒的等待队列中的线程，会继续在在doAcquireSharedInterruptibly()方法中，执行setHeadAndPropagate()方法修改头结点，从而实现唤醒传递。

4.CountDownLatch总结

假设有两个线程A和B，分别调用了CountDownLatch的await()方法，此时state所表示的计数器不为0。所以线程A和B会被封装成SHARED类型的结点，并添加到AQS的等待队列中。

当线程C调用CountDownLatch的coutDown()方法后，如果state被递减到0，那么就会调用doReleaseShared()方法唤醒等待队列中的线程。然后被唤醒的线程会继续调用setHeadAndPropagate()方法实现唤醒传递，从而继续在doReleaseShared()方法中唤醒所有在等待队列中的被阻塞的线程。

5.控制并发线程数的Semaphore介绍

(1)Semaphore的作用

(2)Semaphore的方法

(3)Semaphore原理分析

(1)Semaphore的作用

Semaphore信号量用来控制同时访问特定资源的线程数量，有两核心方法。

方法一：acquire()方法，获取一个令牌

方法二：release()方法，释放一个令牌

多个线程访问某限制访问流量的资源时，可先调用acquire()获取访问令牌。如果能够正常获得，则表示允许访问。如果令牌不够，则会阻塞当前线程。当某个获得令牌的线程通过release()方法释放一个令牌后，被阻塞在acquire()方法的线程就有机会获得这个释放的令牌。

public class SemaphoreDemo {public static void main(String[] args) throws InterruptedException {Semaphore semaphore = new Semaphore(10, true);//初始化10个资源，使用公平锁 semaphore.acquire();//每次获取一个资源，如果获取不到，线程就会阻塞semaphore.release();//释放一个资源}
}

(2)Semaphore的方法

Semaphore实际上并没有一个真实的令牌发给线程，Semaphore只是对一个可分配数量进行计数维护，或者说进行许可证管理。Semaphore可以在公共资源有限的场景下实现流量控制，如数据库连接。

一.Semaphore(permits, fair)：permits表示令牌数，fair表示公平性
二.acquire(permits)：获取指定数量的令牌，如果数量不足则阻塞当前线程
三.tryAcquire(permits)：尝试获取指定数量的令牌，此过程是非阻塞的，成功返回true，失败返回false 
四.release(permits)：释放指定数量的令牌
五.drainPermits()：当前线程获得剩下的所有令牌
六.hasQueuedThread()：判断当前Semaphore实例上是否存在等待令牌的线程

(3)Semaphore原理分析

Semaphore也是基于AQS中的共享锁来实现的。在创建Semaphore实例时传递的参数permits，其实就是AQS中的state属性。每次调用Semaphore的acquire()方法，都会对state值进行递减。

所以从根本上说，Semaphore是通过重写AQS的两个方法来实现的：

方法一：tryAcquireShared()，抢占共享锁

方法二：tryReleaseShared()，释放共享锁

public class Semaphore implements java.io.Serializable {private final Sync sync;//Creates a Semaphore with the given number of permits and nonfair fairness setting.public Semaphore(int permits) {sync = new NonfairSync(permits);}static final class NonfairSync extends Sync {NonfairSync(int permits) {super(permits);}protected int tryAcquireShared(int acquires) {return nonfairTryAcquireShared(acquires);}}//Acquires a permit from this semaphore, blocking until one is available, //or the thread is Thread#interrupt interrupted.public void acquire() throws InterruptedException {//执行AQS的模版方法acquireSharedInterruptibly()sync.acquireSharedInterruptibly(1);}//Releases a permit, returning it to the semaphore.public void release() {//执行AQS的模版方法releaseShared()sync.releaseShared(1);}//Synchronization implementation for semaphore.  //Uses AQS state to represent permits. Subclassed into fair and nonfair versions.abstract static class Sync extends AbstractQueuedSynchronizer {Sync(int permits) {//设置state的值为传入的令牌数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) {throw new Error("Maximum permit count exceeded");}if (compareAndSetState(current, next)) {return true;}}}...}...
}public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {...//Acquires in shared mode, aborting if interrupted.//Implemented by first checking interrupt status, then invoking at least once #tryAcquireShared, returning on success.//Otherwise the thread is queued, possibly repeatedly blocking and unblocking,//invoking #tryAcquireShared until success or the thread is interrupted.public final void acquireSharedInterruptibly(int arg) throws InterruptedException {if (Thread.interrupted()) {throw new InterruptedException();}//执行Semaphore的内部类Sync的子类实现的tryAcquireShared()方法，抢占共享锁if (tryAcquireShared(arg) < 0) {//执行AQS的doAcquireSharedInterruptibly()方法doAcquireSharedInterruptibly(arg);}}//Releases in shared mode.  //Implemented by unblocking one or more threads if #tryReleaseShared returns true.public final boolean releaseShared(int arg) {//执行Semaphore的内部类Sync实现的tryReleaseShared()方法，释放共享锁if (tryReleaseShared(arg)) {//执行AQS的doReleaseShared()方法doReleaseShared();return true;}return false;}...
}

6.Semaphore的令牌获取过程

(1)Semaphore的令牌获取过程

(2)Semaphore的公平策略

(3)Semaphore的非公平策略

(4)tryAcquireShared()后的处理

(1)Semaphore的令牌获取过程

在调用Semaphore的acquire()方法获取令牌时：首先会执行AQS的模版方法acquireSharedInterruptibly()，然后执行Sync子类实现的tryAcquireShared()方法来抢占锁。如果抢占锁失败，则执行AQS的doAcquireSharedInterruptibly()方法。该方法会将当前线程封装成Node结点并加入等待队列，然后挂起线程。

(2)Semaphore的公平策略

在执行Sync子类FairSync的tryAcquireShared()方法尝试获取令牌时，先通过AQS的hasQueuedPredecessors()判断是否已有线程在等待队列中。如果已经有线程在等待队列中，那么当前线程获取令牌就必然失败。否则，就递减state的值 + 判断state是否小于0 + CAS设置state的值。

(3)Semaphore的非公平策略

在执行Sync子类NonfairSync的tryAcquireShared()方法尝试获取令牌时，则会直接执行Sync的nonfairTryAcquireShared()方法来获取令牌，也就是递减state的值 + 判断state是否小于0 + CAS设置state的值。

(4)tryAcquireShared()后的处理

不管公平策略还是非公平策略，对应的tryAcquireShared()方法都是通过自旋来抢占令牌(CAS设置state)，直到令牌数不够时才会让tryAcquireShared()方法返回小于0的数值。然后触发执行AQS的doAcquireSharedInterruptibly()方法，该方法会将当前线程封装成Node结点并加入等待队列，然后挂起线程。

public class Semaphore implements java.io.Serializable {private final Sync sync;//Creates a Semaphore with the given number of permits and nonfair fairness setting.public Semaphore(int permits) {sync = new NonfairSync(permits);}static final class NonfairSync extends Sync {NonfairSync(int permits) {super(permits);}//以非公平锁的方式获取令牌protected int tryAcquireShared(int acquires) {//执行Sync的nonfairTryAcquireShared()方法return nonfairTryAcquireShared(acquires);}}static final class FairSync extends Sync {FairSync(int permits) {super(permits);}//以公平锁的方式获取令牌protected int tryAcquireShared(int acquires) {for (;;) {//如果已经有线程在等待队列中，那么就说明获取令牌必然失败if (hasQueuedPredecessors()) {return -1;}int available = getState();int remaining = available - acquires;if (remaining < 0 || compareAndSetState(available, remaining)) {return remaining;}}}}//Acquires a permit from this semaphore, blocking until one is available, //or the thread is Thread#interrupt interrupted.public void acquire() throws InterruptedException {//执行AQS的模版方法acquireSharedInterruptibly()sync.acquireSharedInterruptibly(1);}//Synchronization implementation for semaphore.  //Uses AQS state to represent permits. Subclassed into fair and nonfair versions.abstract static class Sync extends AbstractQueuedSynchronizer {Sync(int permits) {//设置state的值为传入的令牌数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;}}}...}...
}public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {...//Acquires in shared mode, aborting if interrupted.//Implemented by first checking interrupt status, then invoking at least once #tryAcquireShared, returning on success.//Otherwise the thread is queued, possibly repeatedly blocking and unblocking,//invoking #tryAcquireShared until success or the thread is interrupted.public final void acquireSharedInterruptibly(int arg) throws InterruptedException {if (Thread.interrupted()) {throw new InterruptedException();}//执行Semaphore的内部类Sync的子类实现的tryAcquireShared()方法，抢占共享锁if (tryAcquireShared(arg) < 0) {//执行AQS的doAcquireSharedInterruptibly()方法doAcquireSharedInterruptibly(arg);}}//Queries whether any threads have been waiting to acquire longer than the current thread.public final boolean hasQueuedPredecessors() {Node t = tail; // Read fields in reverse initialization orderNode h = head;Node s;return h != t && ((s = h.next) == null || s.thread != Thread.currentThread());}//Acquires in shared interruptible mode.private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {final Node node = addWaiter(Node.SHARED);//封装当前线程为Shared类型的Node结点boolean failed = true;try {//第一次循环r = -1，所以会执行AQS的shouldParkAfterFailedAcquire()方法//将node结点的有效前驱结点的状态设置为SIGNALfor (;;) {final Node p = node.predecessor();//node结点的前驱结点if (p == head) {int r = tryAcquireShared(arg);if (r >= 0) {setHeadAndPropagate(node, r);p.next = null; // help GCfailed = false;return;}}//执行shouldParkAfterFailedAcquire()方法设置node结点的前驱结点的状态为SIGNAL//执行parkAndCheckInterrupt()方法挂起当前线程if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) {throw new InterruptedException();}}} finally {if (failed) {cancelAcquire(node);}}}...
}

7.Semaphore的令牌释放过程

(1)Semaphore的令牌释放过程

(2)Semaphore的令牌释放本质

(1)Semaphore的令牌释放过程

在调用Semaphore的release()方法去释放令牌时：首先会执行AQS的模版方法releaseShared()，然后执行Sync实现的tryReleaseShared()方法来释放锁(累加state值)。如果释放锁成功，则执行AQS的doReleaseShared()方法去唤醒线程。

(2)Semaphore的令牌释放本质

Semaphore的release()方法释放令牌的本质就是对state字段进行累加，然后唤醒等待队列头结点的后继结点 + 唤醒传递来唤醒等待的线程。

注意：并非一定要执行acquire()方法的线程才能调用release()方法，任意一个线程都可以调用release()方法，也可以通过reducePermits()方法来减少令牌数。

public class Semaphore implements java.io.Serializable {private final Sync sync;//Creates a Semaphore with the given number of permits and nonfair fairness setting.public Semaphore(int permits) {sync = new NonfairSync(permits);}//Releases a permit, returning it to the semaphore.public void release() {//执行AQS的模版方法releaseShared()sync.releaseShared(1);}//Synchronization implementation for semaphore.  //Uses AQS state to represent permits. Subclassed into fair and nonfair versions.abstract static class Sync extends AbstractQueuedSynchronizer {Sync(int permits) {//设置state的值为传入的令牌数setState(permits);}//尝试释放锁，也就是对state值进行累加protected final boolean tryReleaseShared(int releases) {for (;;) {int current = getState();int next = current + releases;if (next < current) {throw new Error("Maximum permit count exceeded");}if (compareAndSetState(current, next)) {return true;}}}...}...
}public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {...    //Releases in shared mode.  //Implemented by unblocking one or more threads if #tryReleaseShared returns true.public final boolean releaseShared(int arg) {//执行Semaphore的内部类Sync实现的tryReleaseShared()方法，释放共享锁if (tryReleaseShared(arg)) {//执行AQS的doReleaseShared()方法，唤醒等待队列中的线程doReleaseShared();return true;}return false;}//Release action for shared mode -- signals successor and ensures propagation. //Note: For exclusive mode, release just amounts to calling unparkSuccessor of head if it needs signal.private void doReleaseShared() {for (;;) {//每次循环时头结点都会发生变化//因为调用unparkSuccessor()方法会唤醒doAcquireSharedInterruptibly()方法中阻塞的线程//然后阻塞的线程会在执行setHeadAndPropagate()方法时通过setHead()修改头结点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}//唤醒头结点的后继结点//唤醒的线程会在doAcquireSharedInterruptibly()方法中执行setHeadAndPropagate()方法修改头结点unparkSuccessor(h);} else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) {//如果ws = 0表示初始状态，则修改结点为PROPAGATE状态continue;//loop on failed CAS}}if (h == head) {//判断头结点是否有变化break;//loop if head changed}}}//Wakes up node's successor, if one exists.private void unparkSuccessor(Node node) {int ws = node.waitStatus;if (ws < 0) {compareAndSetWaitStatus(node, ws, 0);}Node s = node.next;if (s == null || s.waitStatus > 0) {s = null;for (Node t = tail; t != null && t != node; t = t.prev) {if (t.waitStatus <= 0) {s = t;}}}if (s != null) {LockSupport.unpark(s.thread);}}...
}

8.同步屏障CyclicBarrier介绍

(1)CyclicBarrier的作用

(2)CyclicBarrier的基本原理

(1)CyclicBarrier的作用

CyclicBarrier的字面意思就是可循环使用的屏障。CyclicBarrier的主要作用就是让一组线程到达一个屏障时被阻塞，直到最后一个线程到达屏障时屏障才会打开，接着才让所有被屏障拦截的线程一起继续往下执行。线程进入屏障是通过CyclicBarrier的await()方法来实现的。

(2)CyclicBarrier的基本原理

假设有3个线程在运行中都会调用CyclicBarrier的await()方法，而每个线程从开始执行到执行await()方法所用时间可能不一样，最终当执行时间最长的线程到达屏障时，会唤醒其他较早到达屏障的线程继续往下执行。

CyclicBarrier包含两个层面的意思：

一是Barrier屏障点，线程调用await()方法都会阻塞在屏障点，直到所有线程都到达屏障点后再放行。

二是Cyclic循环，当所有线程通过当前屏障点后，又可以进入下一轮的屏障点进行等待，可以不断循环。

9.CyclicBarrier的await()方法源码

(1)CyclicBarrier的成员变量

(2)CyclicBarrier的await()方法源码

(3)CountDownLatch和CyclicBarrier对比

(1)CyclicBarrier的成员变量

//A synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point.  
//CyclicBarriers are useful in programs involving a fixed sized party of threads that must occasionally wait for each other. 
//The barrier is called cyclic because it can be re-used after the waiting threads are released.
public class CyclicBarrier {...private static class Generation {boolean broken = false;}private final ReentrantLock lock = new ReentrantLock();private final Condition trip = lock.newCondition();//用于线程之间相互唤醒private final int parties;//参与的线程数量private int count;//初始值是parties，每调用一次await()就减1private final Runnable barrierCommand;//回调任务private Generation generation = new Generation();...
}

CyclicBarrier是基于ReentrantLock + Condition来实现的。

parties表示每次要求到达屏障点的线程数，只有到达屏障点的线程数满足指定的parties数量，所有线程才会被唤醒。

count是一个初始值为parties的计数器，每个线程调用await()方法会对count减1，当count为0时会唤醒所有线程，并且结束当前的屏障周期generation，然后所有线程进入下一个屏障周期，而且count会恢复成parties。

(2)CyclicBarrier的await()方法源码

线程调用CyclicBarrier的await()方法时，会触发调用CyclicBarrier的dowait()方法。

CyclicBarrier的dowait()方法会对count计数器进行递减。如果count递减到0，则会调用CyclicBarrier的nextGeneration()唤醒所有线程，同时如果异步回调任务barrierCommand不为空，则会执行该任务。如果count还没递减到0，则调用Condition的await()方法阻塞当前线程。

被阻塞的线程，除了会被CyclicBarrier的nextGeneration()方法唤醒外，还会被Thread的interrupt()方法唤醒、被中断异常唤醒，而这些唤醒会调用CyclicBarrier的breakBarrier()方法。

在CyclicBarrier的nextGeneration()方法和CyclicBarrier的breakBarrier()方法中，都会通过Condition的signalAll()方法唤醒所有被阻塞等待的线程。

//A synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point.  
//CyclicBarriers are useful in programs involving a fixed sized party of threads that must occasionally wait for each other. 
//The barrier is called cyclic because it can be re-used after the waiting threads are released.
public class CyclicBarrier {...private static class Generation {boolean broken = false;//用来标记屏障是否被中断}private final ReentrantLock lock = new ReentrantLock();private final Condition trip = lock.newCondition();//用于线程之间相互唤醒private final int parties;//参与的线程数量private int count;//初始值是parties，每调用一次await()就减1private final Runnable barrierCommand;//回调任务private Generation generation = new Generation();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);}//Waits until all #getParties have invoked await on this barrier.public int await() throws InterruptedException, BrokenBarrierException {try {//执行CyclicBarrier的dowait()方法return dowait(false, 0L);} catch (TimeoutException toe) {throw new Error(toe);}}//Main barrier code, covering the various policies.private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException {final ReentrantLock lock = this.lock;lock.lock();//使用Condition需要先获取锁try {//获取当前的generationfinal Generation g = generation;//确认当前generation的barrier是否有效，如果generation的broken为true，则抛出屏障中断异常if (g.broken) {throw new BrokenBarrierException();}if (Thread.interrupted()) {breakBarrier();throw new InterruptedException();}//统计已经到达当前generation的线程数量int index = --count;//如果index为0，则表示所有线程都到达了屏障点if (index == 0) {boolean ranAction = false;try {final Runnable command = barrierCommand;if (command != null) {//触发回调command.run();}ranAction = true;//执行nextGeneration()方法唤醒所有线程，同时进入下一个屏障周期nextGeneration();return 0;} finally {if (!ranAction) {breakBarrier();}}}//loop until tripped, broken, interrupted, or timed out//如果index > 0，则阻塞当前线程for (;;) {try {if (!timed) {//通过Condition的await()方法，在阻塞当前线程的同时释放锁//这样其他线程就能获取到锁执行上面的index = --counttrip.await();} else if (nanos > 0L) {nanos = trip.awaitNanos(nanos);}} catch (InterruptedException ie) {if (g == generation && ! g.broken) {breakBarrier();throw ie;} else {Thread.currentThread().interrupt();}}if (g.broken) {throw new BrokenBarrierException();}if (g != generation) {return index;}if (timed && nanos <= 0L) {//中断屏障，设置generation.broken为truebreakBarrier();throw new TimeoutException();}}} finally {lock.unlock();}}//Updates state on barrier trip and wakes up everyone.//Called only while holding lock.private void nextGeneration() {//通过Condition的signalAll()唤醒所有等待的线程trip.signalAll();//还原countcount = parties;//进入新的generationgeneration = new Generation();}//Sets current barrier generation as broken and wakes up everyone.//Called only while holding lock.private void breakBarrier() {generation.broken = true;count = parties;//通过Condition的signalAll()唤醒所有等待的线程trip.signalAll();}...
}

(3)CountDownLatch和CyclicBarrier对比

一.CyclicBarrier可以被重用、可以响应中断

二.CountDownLatch的计数器只能使用一次，但可以通过reset()方法重置

10.使用CountDownLatch等待注册的完成

Hadoop HDFS(分布式存储系统)的NameNode分为主备两个节点，各个DataNode在启动时都会向两个NameNode进行注册，此时就可以使用CountDownLatch等待向主备节点注册的完成。

//DataNode启动类
public class DataNode {//是否还在运行private volatile Boolean shouldRun;//负责和一组NameNode(主NameNode + 备NameNode)通信的组件private NameNodeGroupOfferService offerService;//初始化DataNodeprivate void initialize() {this.shouldRun = true;this.offerService = new NameNodeGroupOfferService();this.offerService.start();  }//运行DataNodeprivate void run() {try {while(shouldRun) {Thread.sleep(10000);  }   } catch (Exception e) {e.printStackTrace();}}public static void main(String[] args) {DataNode datanode = new DataNode();datanode.initialize();datanode.run(); }
}//负责某个NameNode进行通信的线程组件
public class NameNodeServiceActor {//向某个NameNode进行注册public void register(CountDownLatch latch) {Thread registerThread = new RegisterThread(latch);registerThread.start(); }//负责注册的线程，传入一个CountDownLatchclass RegisterThread extends Thread {CountDownLatch latch;public RegisterThread(CountDownLatch latch) {this.latch = latch;}@Overridepublic void run() {try {//发送rpc接口调用请求到NameNode去进行注册System.out.println("发送请求到NameNode进行注册...");Thread.sleep(1000);  latch.countDown();  } catch (Exception e) {e.printStackTrace();}}}
}//负责跟一组NameNode(主NameNode + 备NameNode)进行通信的线程组件
public class NameNodeGroupOfferService {//负责跟NameNode主节点通信的ServiceActor组件private NameNodeServiceActor activeServiceActor;//负责跟NameNode备节点通信的ServiceActor组件private NameNodeServiceActor standbyServiceActor;//构造函数public NameNodeGroupOfferService() {this.activeServiceActor = new NameNodeServiceActor();this.standbyServiceActor = new NameNodeServiceActor();}//启动OfferService组件public void start() {//直接使用两个ServiceActor组件分别向主备两个NameNode节点进行注册register();}//向主备两个NameNode节点进行注册private void register() {try {CountDownLatch latch = new CountDownLatch(2);  this.activeServiceActor.register(latch); this.standbyServiceActor.register(latch); latch.await();//阻塞等待主备都完成注册System.out.println("主备NameNode全部注册完毕...");} catch (Exception e) {e.printStackTrace();  }}
}

11.使用CyclicBarrier将工作任务多线程分而治之

//输出结果：
//线程1执行自己的一部分工作...
//线程2执行自己的一部分工作...
//线程3执行自己的一部分工作...
//所有线程都完成自己的任务，可以合并结果了...
//最终结果合并完成，线程3可以退出...
//最终结果合并完成，线程1可以退出...
//最终结果合并完成，线程2可以退出...
public class CyclicBarrierDemo {public static void main(String[] args) {final CyclicBarrier barrier = new CyclicBarrier(3, new Runnable() {public void run() {System.out.println("所有线程都完成自己的任务，可以合并结果了...");}});new Thread() {public void run() {try {System.out.println("线程1执行自己的一部分工作...");barrier.await();System.out.println("最终结果合并完成，线程1可以退出...");} catch (Exception e) {e.printStackTrace();}}}.start();new Thread() {public void run() {try {System.out.println("线程2执行自己的一部分工作...");barrier.await();System.out.println("最终结果合并完成，线程2可以退出...");} catch (Exception e) {e.printStackTrace();}}}.start();new Thread() {public void run() {try {System.out.println("线程3执行自己的一部分工作...");barrier.await();System.out.println("最终结果合并完成，线程3可以退出...");} catch (Exception e) {e.printStackTrace();}}}.start();}
}

12.使用CyclicBarrier聚合服务接口的返回结果

当然也可以使用CountDownLatch来实现聚合服务接口的返回结果；

public class ApiServiceDemo {public Map<String, Object> queryOrders() throws Exception {final List<Object> results = new ArrayList<Object>();final Map<String, Object> map = new ConcurrentHashMap<String, Object>();CyclicBarrier barrier = new CyclicBarrier(3, new Runnable() {@Overridepublic void run() {map.put("price", results.get(0));   map.put("order", results.get(1)); map.put("stats", results.get(2));  }});//请求价格接口new Thread() {public void run() {try {System.out.println("请求价格服务..."); Thread.sleep(1000);  results.add(new Object());    barrier.await();} catch (Exception e) {e.printStackTrace();  } };}.start();//请求订单接口new Thread() {public void run() {try {System.out.println("请求订单服务..."); Thread.sleep(1000);  results.add(new Object());    barrier.await();} catch (Exception e) {e.printStackTrace();  } };}.start();//请求统计接口new Thread() {public void run() {try {System.out.println("请求订单统计服务..."); Thread.sleep(1000);  results.add(new Object());    barrier.await();} catch (Exception e) {e.printStackTrace();  } };}.start();while(map.size() < 3) {Thread.sleep(100);  }return map;}
}

13.使用Semaphore等待指定数量线程完成任务

可以通过Semaphore实现等待指定数量的线程完成任务才往下执行。

//输出结果如下：
//线程2执行一个计算任务
//等待1个线程完成任务即可...
//线程1执行一个计算任务
public class SemaphoreDemo {public static void main(String[] args) throws Exception {final Semaphore semaphore = new Semaphore(0);new Thread() {public void run() {try {Thread.sleep(2000);System.out.println("线程1执行一个计算任务");semaphore.release();} catch (Exception e) {e.printStackTrace();}}}.start();new Thread() {public void run() {try {Thread.sleep(1000);System.out.println("线程2执行一个计算任务");semaphore.release();} catch (Exception e) {e.printStackTrace();}}}.start();semaphore.acquire(1);System.out.println("等待1个线程完成任务即可...");}
}

相关文章：