经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » Java相关 » Java » 查看文章
JUC并发—7.AQS源码分析三
来源：cnblogs　　作者：东阳马生架构　　时间：2025/2/20 10:39:25　　对本文有异议

大纲

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的例子

  1. public class CountDownLatchDemo {
  2.     public static void main(String[] args) throws Exception {
  3.         final CountDownLatch latch = new CountDownLatch(2);
  4.         new Thread() {
  5.             public void run() {
  6.                 try {
  7.                     Thread.sleep(1000);
  8.                     System.out.println("线程1开始执行，休眠2秒...");
  9.                     Thread.sleep(1000);
  11.                     System.out.println("线程1准备执行countDown操作...");
  12.                     latch.countDown();
  14.                     System.out.println("线程1完成执行countDown操作...");
  15.                 } catch (Exception e) {
  16.                     e.printStackTrace();
  17.                 }
  18.             }
  19.         }.start();
  21.         new Thread() {
  22.             public void run() {
  23.                 try {
  24.                     Thread.sleep(1000);
  25.                     System.out.println("线程2开始执行，休眠2秒...");
  26.                     Thread.sleep(1000);
  28.                     System.out.println("线程2准备执行countDown操作...");
  29.                     latch.countDown();
  31.                     System.out.println("线程2完成执行countDown操作...");
  32.                 } catch (Exception e) {
  33.                     e.printStackTrace();
  34.                 }
  35.             }
  36.         }.start();
  38.         System.out.println("main线程准备执行countDownLatch的await操作，将会同步阻塞等待...");
  39.         latch.await();
  40.         System.out.println("所有线程都完成countDown操作，结束同步阻塞等待...");
  41.     }
  42. }

 

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结点加入等待队列，然后挂起当前线程进行阻塞。

  1. //A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.
  2. public class CountDownLatch {
  3.     private final Sync sync;
  4.     
  5.     public CountDownLatch(int count) {
  6.         if (count < 0) {
  7.             throw new IllegalArgumentException("count < 0");
  8.         }
  9.         this.sync = new Sync(count);
  10.     }
  11.     
  12.     //Synchronization control For CountDownLatch.
  13.     //Uses AQS state to represent count.
  14.     private static final class Sync extends AbstractQueuedSynchronizer {
  15.         Sync(int count) {
  16.             setState(count);
  17.         }
  18.         
  19.         int getCount() {
  20.             return getState();
  21.         }
  22.         
  23.         protected int tryAcquireShared(int acquires) {
  24.             return (getState() == 0) ? 1 : -1;
  25.         }
  26.         
  27.         protected boolean tryReleaseShared(int releases) {
  28.             //Decrement count; signal when transition to zero
  29.             for (;;) {
  30.                 int c = getState();
  31.                 if (c == 0) {
  32.                     return false;
  33.                 }
  34.                 int nextc = c-1;
  35.                 if (compareAndSetState(c, nextc)) {
  36.                     return nextc == 0;
  37.                 }
  38.             }
  39.         }
  40.     }
  41.     
  42.     //Causes the current thread to wait until the latch has counted down to zero, 
  43.     //unless the thread is Thread#interrupt interrupted.
  44.     public void await() throws InterruptedException {
  45.         //执行AQS的acquireSharedInterruptibly()方法
  46.         sync.acquireSharedInterruptibly(1);
  47.     }
  48.     ...
  49. }
  51. public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {
  52.     ...
  53.     //Acquires in shared mode, aborting if interrupted.
  54.     //Implemented by first checking interrupt status, then invoking at least once #tryAcquireShared, returning on success.
  55.     //Otherwise the thread is queued, possibly repeatedly blocking and unblocking,
  56.     //invoking #tryAcquireShared until success or the thread is interrupted.
  57.     public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
  58.         if (Thread.interrupted()) {
  59.             throw new InterruptedException();
  60.         }
  61.         //执行CountDownLatch的内部类Sync实现的tryAcquireShared()方法，抢占共享锁
  62.         if (tryAcquireShared(arg) < 0) {
  63.             //执行AQS的doAcquireSharedInterruptibly()方法
  64.             doAcquireSharedInterruptibly(arg);
  65.         }
  66.     }
  67.     
  68.     //Acquires in shared interruptible mode.
  69.     private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
  70.         final Node node = addWaiter(Node.SHARED);//封装当前线程为Shared类型的Node结点
  71.         boolean failed = true;
  72.         try {
  73.             //第一次循环r = -1，所以会执行AQS的shouldParkAfterFailedAcquire()方法
  74.             //将node结点的有效前驱结点的状态设置为SIGNAL
  75.             for (;;) {
  76.                 final Node p = node.predecessor();//node结点的前驱结点
  77.                 if (p == head) {
  78.                     int r = tryAcquireShared(arg);
  79.                     if (r >= 0) {
  80.                         setHeadAndPropagate(node, r);
  81.                         p.next = null; // help GC
  82.                         failed = false;
  83.                         return;
  84.                     }
  85.                 }
  86.                 //执行shouldParkAfterFailedAcquire()方法设置node结点的前驱结点的状态为SIGNAL
  87.                 //执行parkAndCheckInterrupt()方法挂起当前线程
  88.                 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) {
  89.                     throw new InterruptedException();
  90.                 }
  91.             }
  92.         } finally {
  93.             if (failed) {
  94.                 cancelAcquire(node);
  95.             }
  96.         }
  97.     }
  98.     
  99.     //Checks and updates status for a node that failed to acquire.
  100.     //Returns true if thread should block. This is the main signal control in all acquire loops.
  101.     private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
  102.         int ws = pred.waitStatus;
  103.         if (ws == Node.SIGNAL) {
  104.             //This node has already set status asking a release to signal it, so it can safely park.
  105.             return true;
  106.         }
  107.         if (ws > 0) {
  108.             //Predecessor was cancelled. Skip over predecessors and indicate retry.
  109.             do {
  110.                 node.prev = pred = pred.prev;
  111.             } while (pred.waitStatus > 0);
  112.             pred.next = node;
  113.         } else {
  114.             //waitStatus must be 0 or PROPAGATE.  
  115.             //Indicate that we need a signal, but don't park yet.  
  116.             //Caller will need to retry to make sure it cannot acquire before parking.
  117.             compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
  118.         }
  119.         return false;
  120.     }
  121.     
  122.     //设置头结点和唤醒后续线程
  123.     //Sets head of queue, and checks if successor may be waiting in shared mode, 
  124.     //if so propagating if either propagate > 0 or PROPAGATE status was set.
  125.     private void setHeadAndPropagate(Node node, int propagate) {
  126.         Node h = head;
  127.         setHead(node);//将node结点设置为头结点
  128.         if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) {
  129.             Node s = node.next;
  130.             if (s == null || s.isShared()) {
  131.                 doReleaseShared();
  132.             }
  133.         }
  134.     }
  135.     
  136.     private void setHead(Node node) {
  137.         head = node;
  138.         node.thread = null;
  139.         node.prev = null;
  140.     }
  141.     ...
  142. }

(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()方法唤醒共享锁模式下的等待队列中的线程。

  1. //A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.
  2. public class CountDownLatch {
  3.     private final Sync sync;
  4.     
  5.     public CountDownLatch(int count) {
  6.         if (count < 0) {
  7.             throw new IllegalArgumentException("count < 0");
  8.         }
  9.         this.sync = new Sync(count);
  10.     }
  11.     
  12.     //Synchronization control For CountDownLatch.
  13.     //Uses AQS state to represent count.
  14.     private static final class Sync extends AbstractQueuedSynchronizer {
  15.         Sync(int count) {
  16.             setState(count);
  17.         }
  18.         
  19.         int getCount() {
  20.             return getState();
  21.         }
  22.         
  23.         protected int tryAcquireShared(int acquires) {
  24.             return (getState() == 0) ? 1 : -1;
  25.         }
  26.         
  27.         protected boolean tryReleaseShared(int releases) {
  28.             //Decrement count; signal when transition to zero
  29.             for (;;) {
  30.                 int c = getState();
  31.                 if (c == 0) {
  32.                     return false;
  33.                 }
  34.                 int nextc = c-1;
  35.                 if (compareAndSetState(c, nextc)) {
  36.                     return nextc == 0;
  37.                 }
  38.             }
  39.         }
  40.     }
  41.            
  42.     //Decrements the count of the latch, releasing all waiting threads if the count reaches zero.
  43.     public void countDown() {
  44.         //执行AQS的releaseShared()方法
  45.         sync.releaseShared(1);
  46.     }
  47.     ...
  48. }
  50. public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {
  51.     ...
  52.     //Releases in shared mode.  
  53.     //Implemented by unblocking one or more threads if #tryReleaseShared returns true.
  54.     public final boolean releaseShared(int arg) {
  55.         //执行CountDownLatch的内部类Sync实现的tryReleaseShared()方法，释放共享锁
  56.         if (tryReleaseShared(arg)) {
  57.             //执行AQS的doReleaseShared()方法
  58.             doReleaseShared();
  59.             return true;
  60.         }
  61.         return false;
  62.     }
  63.     
  64.     //Release action for shared mode -- signals successor and ensures propagation. 
  65.     //Note: For exclusive mode, release just amounts to calling unparkSuccessor of head if it needs signal.
  66.     private void doReleaseShared() {
  67.         for (;;) {
  68.             //每次循环时头结点都会发生变化
  69.             //因为调用unparkSuccessor()方法会唤醒doAcquireSharedInterruptibly()方法中阻塞的线程
  70.             //然后阻塞的线程会在执行setHeadAndPropagate()方法时通过setHead()修改头结点
  71.             Node h = head;//获取最新的头结点
  72.             if (h != null && h != tail) {//等待队列中存在挂起线程的结点
  73.                 int ws = h.waitStatus;
  74.                 if (ws == Node.SIGNAL) {//头结点的状态正常，表示对应的线程可以被唤醒
  75.                     if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) {
  76.                         continue;//loop to recheck cases
  77.                     }
  78.                     //唤醒头结点的后继结点
  79.                     //唤醒的线程会在doAcquireSharedInterruptibly()方法中执行setHeadAndPropagate()方法修改头结点
  80.                     unparkSuccessor(h);
  81.                 } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) {
  82.                     //如果ws = 0表示初始状态，则修改结点为PROPAGATE状态
  83.                     continue;//loop on failed CAS
  84.                 }
  85.             }
  86.             if (h == head) {//判断头结点是否有变化
  87.                 break;//loop if head changed
  88.             }
  89.         }
  90.     }
  91.     
  92.     //Wakes up node's successor, if one exists.
  93.     private void unparkSuccessor(Node node) {
  94.         int ws = node.waitStatus;
  95.         if (ws < 0) {
  96.             compareAndSetWaitStatus(node, ws, 0);
  97.         }
  99.         Node s = node.next;
  100.         if (s == null || s.waitStatus > 0) {
  101.             s = null;
  102.             for (Node t = tail; t != null && t != node; t = t.prev) {
  103.                 if (t.waitStatus <= 0) {
  104.                     s = t;
  105.                 }
  106.             }
  107.         }
  108.         if (s != null) {
  109.             LockSupport.unpark(s.thread);
  110.         }
  111.     }
  112.     ...
  113. }

(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()方法的线程就有机会获得这个释放的令牌。

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

(2)Semaphore的方法

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

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

(3)Semaphore原理分析

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

 

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

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

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

  1. public class Semaphore implements java.io.Serializable {
  2.     private final Sync sync;
  3.     
  4.     //Creates a Semaphore with the given number of permits and nonfair fairness setting.
  5.     public Semaphore(int permits) {
  6.         sync = new NonfairSync(permits);
  7.     }
  8.     
  9.     static final class NonfairSync extends Sync {
  10.         NonfairSync(int permits) {
  11.             super(permits);
  12.         }
  13.         
  14.         protected int tryAcquireShared(int acquires) {
  15.             return nonfairTryAcquireShared(acquires);
  16.         }
  17.     }
  18.     
  19.     //Acquires a permit from this semaphore, blocking until one is available, 
  20.     //or the thread is Thread#interrupt interrupted.
  21.     public void acquire() throws InterruptedException {
  22.         //执行AQS的模版方法acquireSharedInterruptibly()
  23.         sync.acquireSharedInterruptibly(1);
  24.     }
  25.     
  26.     //Releases a permit, returning it to the semaphore.
  27.     public void release() {
  28.         //执行AQS的模版方法releaseShared()
  29.         sync.releaseShared(1);
  30.     }
  31.     
  32.     //Synchronization implementation for semaphore.  
  33.     //Uses AQS state to represent permits. Subclassed into fair and nonfair versions.
  34.     abstract static class Sync extends AbstractQueuedSynchronizer {
  35.         Sync(int permits) {
  36.             //设置state的值为传入的令牌数
  37.             setState(permits);
  38.         }
  39.         
  40.         final int getPermits() {
  41.             return getState();
  42.         }
  43.         
  44.         final int nonfairTryAcquireShared(int acquires) {
  45.             for (;;) {
  46.                 int available = getState();
  47.                 int remaining = available - acquires;
  48.                 if (remaining < 0 || compareAndSetState(available, remaining)) {
  49.                     return remaining;
  50.                 }
  51.             }
  52.         }
  53.         
  54.         protected final boolean tryReleaseShared(int releases) {
  55.             for (;;) {
  56.                 int current = getState();
  57.                 int next = current + releases;
  58.                 if (next < current) {
  59.                     throw new Error("Maximum permit count exceeded");
  60.                 }
  61.                 if (compareAndSetState(current, next)) {
  62.                     return true;
  63.                 }
  64.             }
  65.         }
  66.         ...
  67.     }
  68.     ...
  69. }
  71. public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {
  72.     ...
  73.     //Acquires in shared mode, aborting if interrupted.
  74.     //Implemented by first checking interrupt status, then invoking at least once #tryAcquireShared, returning on success.
  75.     //Otherwise the thread is queued, possibly repeatedly blocking and unblocking,
  76.     //invoking #tryAcquireShared until success or the thread is interrupted.
  77.     public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
  78.         if (Thread.interrupted()) {
  79.             throw new InterruptedException();
  80.         }
  81.         
  82.         //执行Semaphore的内部类Sync的子类实现的tryAcquireShared()方法，抢占共享锁
  83.         if (tryAcquireShared(arg) < 0) {
  84.             //执行AQS的doAcquireSharedInterruptibly()方法
  85.             doAcquireSharedInterruptibly(arg);
  86.         }
  87.     }
  88.     
  89.     //Releases in shared mode.  
  90.     //Implemented by unblocking one or more threads if #tryReleaseShared returns true.
  91.     public final boolean releaseShared(int arg) {
  92.         //执行Semaphore的内部类Sync实现的tryReleaseShared()方法，释放共享锁
  93.         if (tryReleaseShared(arg)) {
  94.             //执行AQS的doReleaseShared()方法
  95.             doReleaseShared();
  96.             return true;
  97.         }
  98.         return false;
  99.     }
  100.     ...
  101. }

 

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结点并加入等待队列，然后挂起线程。

  1. public class Semaphore implements java.io.Serializable {
  2.     private final Sync sync;
  3.     
  4.     //Creates a Semaphore with the given number of permits and nonfair fairness setting.
  5.     public Semaphore(int permits) {
  6.         sync = new NonfairSync(permits);
  7.     }
  8.     
  9.     static final class NonfairSync extends Sync {
  10.         NonfairSync(int permits) {
  11.             super(permits);
  12.         }
  13.         
  14.         //以非公平锁的方式获取令牌
  15.         protected int tryAcquireShared(int acquires) {
  16.             //执行Sync的nonfairTryAcquireShared()方法
  17.             return nonfairTryAcquireShared(acquires);
  18.         }
  19.     }
  20.     
  21.     static final class FairSync extends Sync {
  22.         FairSync(int permits) {
  23.             super(permits);
  24.         }
  25.         
  26.         //以公平锁的方式获取令牌
  27.         protected int tryAcquireShared(int acquires) {
  28.             for (;;) {
  29.                 //如果已经有线程在等待队列中，那么就说明获取令牌必然失败
  30.                 if (hasQueuedPredecessors()) {
  31.                     return -1;
  32.                 }
  33.                 int available = getState();
  34.                 int remaining = available - acquires;
  35.                 if (remaining < 0 || compareAndSetState(available, remaining)) {
  36.                     return remaining;
  37.                 }
  38.             }
  39.         }
  40.     }
  41.     
  42.     //Acquires a permit from this semaphore, blocking until one is available, 
  43.     //or the thread is Thread#interrupt interrupted.
  44.     public void acquire() throws InterruptedException {
  45.         //执行AQS的模版方法acquireSharedInterruptibly()
  46.         sync.acquireSharedInterruptibly(1);
  47.     }
  48.     
  49.     //Synchronization implementation for semaphore.  
  50.     //Uses AQS state to represent permits. Subclassed into fair and nonfair versions.
  51.     abstract static class Sync extends AbstractQueuedSynchronizer {
  52.         Sync(int permits) {
  53.             //设置state的值为传入的令牌数
  54.             setState(permits);
  55.         }
  56.         
  57.         final int getPermits() {
  58.             return getState();
  59.         }
  60.         
  61.         final int nonfairTryAcquireShared(int acquires) {
  62.             for (;;) {
  63.                 int available = getState();
  64.                 int remaining = available - acquires;
  65.                 if (remaining < 0 || compareAndSetState(available, remaining)) {
  66.                     return remaining;
  67.                 }
  68.             }
  69.         }
  70.         ...
  71.     }
  72.     ...
  73. }
  76. public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {
  77.     ...
  78.     //Acquires in shared mode, aborting if interrupted.
  79.     //Implemented by first checking interrupt status, then invoking at least once #tryAcquireShared, returning on success.
  80.     //Otherwise the thread is queued, possibly repeatedly blocking and unblocking,
  81.     //invoking #tryAcquireShared until success or the thread is interrupted.
  82.     public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
  83.         if (Thread.interrupted()) {
  84.             throw new InterruptedException();
  85.         }
  86.         
  87.         //执行Semaphore的内部类Sync的子类实现的tryAcquireShared()方法，抢占共享锁
  88.         if (tryAcquireShared(arg) < 0) {
  89.             //执行AQS的doAcquireSharedInterruptibly()方法
  90.             doAcquireSharedInterruptibly(arg);
  91.         }
  92.     }
  93.     
  94.     //Queries whether any threads have been waiting to acquire longer than the current thread.
  95.     public final boolean hasQueuedPredecessors() {
  96.         Node t = tail; // Read fields in reverse initialization order
  97.         Node h = head;
  98.         Node s;
  99.         return h != t && ((s = h.next) == null || s.thread != Thread.currentThread());
  100.     }
  101.     
  102.     //Acquires in shared interruptible mode.
  103.     private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
  104.         final Node node = addWaiter(Node.SHARED);//封装当前线程为Shared类型的Node结点
  105.         boolean failed = true;
  106.         try {
  107.             //第一次循环r = -1，所以会执行AQS的shouldParkAfterFailedAcquire()方法
  108.             //将node结点的有效前驱结点的状态设置为SIGNAL
  109.             for (;;) {
  110.                 final Node p = node.predecessor();//node结点的前驱结点
  111.                 if (p == head) {
  112.                     int r = tryAcquireShared(arg);
  113.                     if (r >= 0) {
  114.                         setHeadAndPropagate(node, r);
  115.                         p.next = null; // help GC
  116.                         failed = false;
  117.                         return;
  118.                     }
  119.                 }
  120.                 //执行shouldParkAfterFailedAcquire()方法设置node结点的前驱结点的状态为SIGNAL
  121.                 //执行parkAndCheckInterrupt()方法挂起当前线程
  122.                 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) {
  123.                     throw new InterruptedException();
  124.                 }
  125.             }
  126.         } finally {
  127.             if (failed) {
  128.                 cancelAcquire(node);
  129.             }
  130.         }
  131.     }
  132.     ...
  133. }

 

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()方法来减少令牌数。

  1. public class Semaphore implements java.io.Serializable {
  2.     private final Sync sync;
  3.     
  4.     //Creates a Semaphore with the given number of permits and nonfair fairness setting.
  5.     public Semaphore(int permits) {
  6.         sync = new NonfairSync(permits);
  7.     }
  8.     
  9.     //Releases a permit, returning it to the semaphore.
  10.     public void release() {
  11.         //执行AQS的模版方法releaseShared()
  12.         sync.releaseShared(1);
  13.     }
  14.     
  15.     //Synchronization implementation for semaphore.  
  16.     //Uses AQS state to represent permits. Subclassed into fair and nonfair versions.
  17.     abstract static class Sync extends AbstractQueuedSynchronizer {
  18.         Sync(int permits) {
  19.             //设置state的值为传入的令牌数
  20.             setState(permits);
  21.         }
  22.         
  23.         //尝试释放锁，也就是对state值进行累加
  24.         protected final boolean tryReleaseShared(int releases) {
  25.             for (;;) {
  26.                 int current = getState();
  27.                 int next = current + releases;
  28.                 if (next < current) {
  29.                     throw new Error("Maximum permit count exceeded");
  30.                 }
  31.                 if (compareAndSetState(current, next)) {
  32.                     return true;
  33.                 }
  34.             }
  35.         }
  36.         ...
  37.     }
  38.     ...
  39. }
  42. public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {
  43.     ...    
  44.     //Releases in shared mode.  
  45.     //Implemented by unblocking one or more threads if #tryReleaseShared returns true.
  46.     public final boolean releaseShared(int arg) {
  47.         //执行Semaphore的内部类Sync实现的tryReleaseShared()方法，释放共享锁
  48.         if (tryReleaseShared(arg)) {
  49.             //执行AQS的doReleaseShared()方法，唤醒等待队列中的线程
  50.             doReleaseShared();
  51.             return true;
  52.         }
  53.         return false;
  54.     }
  55.     
  56.     //Release action for shared mode -- signals successor and ensures propagation. 
  57.     //Note: For exclusive mode, release just amounts to calling unparkSuccessor of head if it needs signal.
  58.     private void doReleaseShared() {
  59.         for (;;) {
  60.             //每次循环时头结点都会发生变化
  61.             //因为调用unparkSuccessor()方法会唤醒doAcquireSharedInterruptibly()方法中阻塞的线程
  62.             //然后阻塞的线程会在执行setHeadAndPropagate()方法时通过setHead()修改头结点
  63.             Node h = head;//获取最新的头结点
  64.             if (h != null && h != tail) {//等待队列中存在挂起线程的结点
  65.                 int ws = h.waitStatus;
  66.                 if (ws == Node.SIGNAL) {//头结点的状态正常，表示对应的线程可以被唤醒
  67.                     if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) {
  68.                         continue;//loop to recheck cases
  69.                     }
  70.                     //唤醒头结点的后继结点
  71.                     //唤醒的线程会在doAcquireSharedInterruptibly()方法中执行setHeadAndPropagate()方法修改头结点
  72.                     unparkSuccessor(h);
  73.                 } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) {
  74.                     //如果ws = 0表示初始状态，则修改结点为PROPAGATE状态
  75.                     continue;//loop on failed CAS
  76.                 }
  77.             }
  78.             if (h == head) {//判断头结点是否有变化
  79.                 break;//loop if head changed
  80.             }
  81.         }
  82.     }
  83.     
  84.     //Wakes up node's successor, if one exists.
  85.     private void unparkSuccessor(Node node) {
  86.         int ws = node.waitStatus;
  87.         if (ws < 0) {
  88.             compareAndSetWaitStatus(node, ws, 0);
  89.         }
  91.         Node s = node.next;
  92.         if (s == null || s.waitStatus > 0) {
  93.             s = null;
  94.             for (Node t = tail; t != null && t != node; t = t.prev) {
  95.                 if (t.waitStatus <= 0) {
  96.                     s = t;
  97.                 }
  98.             }
  99.         }
  100.         if (s != null) {
  101.             LockSupport.unpark(s.thread);
  102.         }
  103.     }
  104.     ...
  105. }

 

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的成员变量

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

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()方法唤醒所有被阻塞等待的线程。

  1. //A synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point.  
  2. //CyclicBarriers are useful in programs involving a fixed sized party of threads that must occasionally wait for each other. 
  3. //The barrier is called cyclic because it can be re-used after the waiting threads are released.
  4. public class CyclicBarrier {
  5.     ...
  6.     private static class Generation {
  7.         boolean broken = false;//用来标记屏障是否被中断
  8.     }
  9.     
  10.     private final ReentrantLock lock = new ReentrantLock();
  11.     private final Condition trip = lock.newCondition();//用于线程之间相互唤醒
  12.     private final int parties;//参与的线程数量
  13.     private int count;//初始值是parties，每调用一次await()就减1
  14.     private final Runnable barrierCommand;//回调任务
  15.     private Generation generation = new Generation();
  16.     
  17.     public CyclicBarrier(int parties, Runnable barrierAction) {
  18.         if (parties <= 0) throw new IllegalArgumentException();
  19.         this.parties = parties;
  20.         this.count = parties;
  21.         this.barrierCommand = barrierAction;
  22.     }
  24.     public CyclicBarrier(int parties) {
  25.         this(parties, null);
  26.     }
  28.     //Waits until all #getParties have invoked await on this barrier.
  29.     public int await() throws InterruptedException, BrokenBarrierException {
  30.         try {
  31.             //执行CyclicBarrier的dowait()方法
  32.             return dowait(false, 0L);
  33.         } catch (TimeoutException toe) {
  34.             throw new Error(toe);
  35.         }
  36.     }
  37.     
  38.     //Main barrier code, covering the various policies.
  39.     private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException {
  40.         final ReentrantLock lock = this.lock;
  41.         lock.lock();//使用Condition需要先获取锁
  42.         try {
  43.             //获取当前的generation
  44.             final Generation g = generation;
  45.             //确认当前generation的barrier是否有效，如果generation的broken为true，则抛出屏障中断异常
  46.             if (g.broken) {
  47.                 throw new BrokenBarrierException();
  48.             }
  49.             if (Thread.interrupted()) {
  50.                 breakBarrier();
  51.                 throw new InterruptedException();
  52.             }
  53.             //统计已经到达当前generation的线程数量
  54.             int index = --count;
  55.             //如果index为0，则表示所有线程都到达了屏障点
  56.             if (index == 0) {
  57.                 boolean ranAction = false;
  58.                 try {
  59.                     final Runnable command = barrierCommand;
  60.                     if (command != null) {
  61.                         //触发回调
  62.                         command.run();
  63.                     }
  64.                     ranAction = true;
  65.                     //执行nextGeneration()方法唤醒所有线程，同时进入下一个屏障周期
  66.                     nextGeneration();
  67.                     return 0;
  68.                 } finally {
  69.                     if (!ranAction) {
  70.                         breakBarrier();
  71.                     }
  72.                 }
  73.             }
  74.             //loop until tripped, broken, interrupted, or timed out
  75.             //如果index > 0，则阻塞当前线程
  76.             for (;;) {
  77.                 try {
  78.                     if (!timed) {
  79.                         //通过Condition的await()方法，在阻塞当前线程的同时释放锁
  80.                         //这样其他线程就能获取到锁执行上面的index = --count
  81.                         trip.await();
  82.                     } else if (nanos > 0L) {
  83.                         nanos = trip.awaitNanos(nanos);
  84.                     }
  85.                 } catch (InterruptedException ie) {
  86.                     if (g == generation && ! g.broken) {
  87.                         breakBarrier();
  88.                         throw ie;
  89.                     } else {
  90.                         Thread.currentThread().interrupt();
  91.                     }
  92.                 }
  93.                 if (g.broken) {
  94.                     throw new BrokenBarrierException();
  95.                 }
  96.                 if (g != generation) {
  97.                     return index;
  98.                 }
  99.                 if (timed && nanos <= 0L) {
  100.                     //中断屏障，设置generation.broken为true
  101.                     breakBarrier();
  102.                     throw new TimeoutException();
  103.                 }
  104.             }
  105.         } finally {
  106.             lock.unlock();
  107.         }
  108.     }
  109.     
  110.     //Updates state on barrier trip and wakes up everyone.
  111.     //Called only while holding lock.
  112.     private void nextGeneration() {
  113.         //通过Condition的signalAll()唤醒所有等待的线程
  114.         trip.signalAll();
  115.         //还原count
  116.         count = parties;
  117.         //进入新的generation
  118.         generation = new Generation();
  119.     }
  120.     
  121.     //Sets current barrier generation as broken and wakes up everyone.
  122.     //Called only while holding lock.
  123.     private void breakBarrier() {
  124.         generation.broken = true;
  125.         count = parties;
  126.         //通过Condition的signalAll()唤醒所有等待的线程
  127.         trip.signalAll();
  128.     }
  129.     ...
  130. }

(3)CountDownLatch和CyclicBarrier对比

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

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

 

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

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

  1. //DataNode启动类
  2. public class DataNode {
  3.     //是否还在运行
  4.     private volatile Boolean shouldRun;
  5.   
  6.     //负责和一组NameNode(主NameNode + 备NameNode)通信的组件
  7.     private NameNodeGroupOfferService offerService;
  8.     
  9.     //初始化DataNode
  10.     private void initialize() {
  11.         this.shouldRun = true;
  12.         this.offerService = new NameNodeGroupOfferService();
  13.         this.offerService.start();  
  14.     }
  15.     
  16.     //运行DataNode
  17.     private void run() {
  18.         try {
  19.             while(shouldRun) {
  20.                 Thread.sleep(10000);  
  21.             }   
  22.         } catch (Exception e) {
  23.             e.printStackTrace();
  24.         }
  25.     }
  26.   
  27.     public static void main(String[] args) {
  28.         DataNode datanode = new DataNode();
  29.         datanode.initialize();
  30.         datanode.run(); 
  31.     }
  32. }
  35. //负责某个NameNode进行通信的线程组件
  36. public class NameNodeServiceActor {
  37.     //向某个NameNode进行注册
  38.     public void register(CountDownLatch latch) {
  39.         Thread registerThread = new RegisterThread(latch);
  40.         registerThread.start(); 
  41.     }
  42.     
  43.     //负责注册的线程，传入一个CountDownLatch
  44.     class RegisterThread extends Thread {
  45.         CountDownLatch latch;
  46.       
  47.         public RegisterThread(CountDownLatch latch) {
  48.             this.latch = latch;
  49.         }
  50.       
  51.         @Override
  52.         public void run() {
  53.             try {
  54.                 //发送rpc接口调用请求到NameNode去进行注册
  55.                 System.out.println("发送请求到NameNode进行注册...");
  56.                 Thread.sleep(1000);  
  57.                 latch.countDown();  
  58.             } catch (Exception e) {
  59.                 e.printStackTrace();
  60.             }
  61.         }
  62.     }
  63. }
  65. //负责跟一组NameNode(主NameNode + 备NameNode)进行通信的线程组件
  66. public class NameNodeGroupOfferService {
  67.     //负责跟NameNode主节点通信的ServiceActor组件
  68.     private NameNodeServiceActor activeServiceActor;
  69.   
  70.     //负责跟NameNode备节点通信的ServiceActor组件
  71.     private NameNodeServiceActor standbyServiceActor;
  72.     
  73.     //构造函数
  74.     public NameNodeGroupOfferService() {
  75.         this.activeServiceActor = new NameNodeServiceActor();
  76.         this.standbyServiceActor = new NameNodeServiceActor();
  77.     }
  78.     
  79.     //启动OfferService组件
  80.     public void start() {
  81.         //直接使用两个ServiceActor组件分别向主备两个NameNode节点进行注册
  82.         register();
  83.     }
  84.     
  85.     //向主备两个NameNode节点进行注册
  86.     private void register() {
  87.         try {
  88.             CountDownLatch latch = new CountDownLatch(2);  
  89.             this.activeServiceActor.register(latch); 
  90.             this.standbyServiceActor.register(latch); 
  91.             latch.await();//阻塞等待主备都完成注册
  92.             System.out.println("主备NameNode全部注册完毕...");
  93.         } catch (Exception e) {
  94.             e.printStackTrace();  
  95.         }
  96.     }
  97. }

 

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

  1. //输出结果：
  2. //线程1执行自己的一部分工作...
  3. //线程2执行自己的一部分工作...
  4. //线程3执行自己的一部分工作...
  5. //所有线程都完成自己的任务，可以合并结果了...
  6. //最终结果合并完成，线程3可以退出...
  7. //最终结果合并完成，线程1可以退出...
  8. //最终结果合并完成，线程2可以退出...
  9. public class CyclicBarrierDemo {
  10.     public static void main(String[] args) {
  11.         final CyclicBarrier barrier = new CyclicBarrier(3, new Runnable() {
  12.             public void run() {
  13.                 System.out.println("所有线程都完成自己的任务，可以合并结果了...");
  14.             }
  15.         });
  16.         new Thread() {
  17.             public void run() {
  18.                 try {
  19.                     System.out.println("线程1执行自己的一部分工作...");
  20.                     barrier.await();
  21.                     System.out.println("最终结果合并完成，线程1可以退出...");
  22.                 } catch (Exception e) {
  23.                     e.printStackTrace();
  24.                 }
  25.             }
  26.         }.start();
  27.         new Thread() {
  28.             public void run() {
  29.                 try {
  30.                     System.out.println("线程2执行自己的一部分工作...");
  31.                     barrier.await();
  32.                     System.out.println("最终结果合并完成，线程2可以退出...");
  33.                 } catch (Exception e) {
  34.                     e.printStackTrace();
  35.                 }
  36.             }
  37.         }.start();
  38.         new Thread() {
  39.             public void run() {
  40.                 try {
  41.                     System.out.println("线程3执行自己的一部分工作...");
  42.                     barrier.await();
  43.                     System.out.println("最终结果合并完成，线程3可以退出...");
  44.                 } catch (Exception e) {
  45.                     e.printStackTrace();
  46.                 }
  47.             }
  48.         }.start();
  49.     }
  50. }

 

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

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

  1. public class ApiServiceDemo {
  2.     public Map<String, Object> queryOrders() throws Exception {
  3.         final List<Object> results = new ArrayList<Object>();
  4.         final Map<String, Object> map = new ConcurrentHashMap<String, Object>();
  5.         CyclicBarrier barrier = new CyclicBarrier(3, new Runnable() {
  6.             @Override
  7.             public void run() {
  8.                 map.put("price", results.get(0));   
  9.                 map.put("order", results.get(1)); 
  10.                 map.put("stats", results.get(2));  
  11.             }
  12.         });
  13.         //请求价格接口
  14.         new Thread() {
  15.             public void run() {
  16.                 try {
  17.                     System.out.println("请求价格服务..."); 
  18.                     Thread.sleep(1000);  
  19.                     results.add(new Object());    
  20.                     barrier.await();
  21.                 } catch (Exception e) {
  22.                     e.printStackTrace();  
  23.                 } 
  24.             };
  25.         }.start();
  26.         //请求订单接口
  27.         new Thread() {
  28.             public void run() {
  29.                 try {
  30.                     System.out.println("请求订单服务..."); 
  31.                     Thread.sleep(1000);  
  32.                     results.add(new Object());    
  33.                     barrier.await();
  34.                 } catch (Exception e) {
  35.                     e.printStackTrace();  
  36.                 } 
  37.             };
  38.         }.start();
  39.         //请求统计接口
  40.         new Thread() {
  41.             public void run() {
  42.                 try {
  43.                     System.out.println("请求订单统计服务..."); 
  44.                     Thread.sleep(1000);  
  45.                     results.add(new Object());    
  46.                     barrier.await();
  47.                 } catch (Exception e) {
  48.                     e.printStackTrace();  
  49.                 } 
  50.             };
  51.         }.start();
  52.         while(map.size() < 3) {
  53.             Thread.sleep(100);  
  54.         }
  55.         return map;
  56.     }
  57. }

 

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

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

  1. //输出结果如下：
  2. //线程2执行一个计算任务
  3. //等待1个线程完成任务即可...
  4. //线程1执行一个计算任务
  5. public class SemaphoreDemo {
  6.     public static void main(String[] args) throws Exception {
  7.         final Semaphore semaphore = new Semaphore(0);
  8.         new Thread() {
  9.             public void run() {
  10.                 try {
  11.                     Thread.sleep(2000);
  12.                     System.out.println("线程1执行一个计算任务");
  13.                     semaphore.release();
  14.                 } catch (Exception e) {
  15.                     e.printStackTrace();
  16.                 }
  17.             }
  18.         }.start();
  20.         new Thread() {
  21.             public void run() {
  22.                 try {
  23.                     Thread.sleep(1000);
  24.                     System.out.println("线程2执行一个计算任务");
  25.                     semaphore.release();
  26.                 } catch (Exception e) {
  27.                     e.printStackTrace();
  28.                 }
  29.             }
  30.         }.start();
  32.         semaphore.acquire(1);
  33.         System.out.println("等待1个线程完成任务即可...");
  34.     }
  35. }

 

原文链接：https://www.cnblogs.com/mjunz/p/18725070

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Java热门文章

  • 【IDEA】Application context not configured for this file
  • [IDEA] chapter_reader - idea看小说插件 idea阅读插件 idea摸鱼插件
  • thrift常见异常及原因分析
  • Java 调用 PaddleDetection 模型
  • JVM 详解，大白话带你认识 JVM
  • spring-boot-2.0.3不一样系列之番外篇 - @Configuration、Condition与@Conditional
  • Java Comparator.comparing比较导致空指针异常的解决
  • Java使用EasyExcel进行单元格合并的问题详解
  • 死磕 java集合之ArrayBlockingQueue源码分析
  • Java基础之集合Set详解

    • Java推荐文章

  • 《刚刚问世》系列初窥篇-Java+Playwright自动化测试-12- iframe操作-上篇（详细教程） - 北京-宏哥
  • JUC并发—7.AQS源码分析三
  • 面试场景题：一次关于线程池使用场景的讨论。
  • 前端黑科技：使用 JavaScript 实现网页扫码功能
  • 聊一聊 Netty 数据搬运工 ByteBuf 体系的设计与实现
  • 简单设计一个JAVA并行处理工具类
  • 【杂谈】JPA乐观锁改悲观锁遇到的一些问题与思考
  • 认识netty的基本组件
  • Jenkins如何使用CrumbIssuer防御CSRF攻击
  • 如何平稳地从nacos迁移到r-nacos？

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号