并发编程之Phaser原理与应用

 前言

成都创新互联公司专注于企业营销型网站建设、网站重做改版、久治网站定制设计、自适应品牌网站建设、H5页面制作、商城网站开发、集团公司官网建设、外贸营销网站建设、高端网站制作、响应式网页设计等建站业务，价格优惠性价比高，为久治等各大城市提供网站开发制作服务。

JDK5中引入了CyclicBarrier和CountDownLatch这两个并发控制类，而JDK7中引入的Phaser按照官方的说法是提供了一个功能类似但是更加灵活的实现。接下来我们带着几个问题来研究一下Phaser与(CountDownLath、CyclicBarrier)到底有哪些类似，同时带来了哪些灵活性?

  1. Phaser 是什么?
  2. Phaser 具有哪些特性?
  3. Phaser相对于 CyclicBarrier 和 CountDownLatch的优势?

CyclicBarrier和CountDownLatch

CyclicBarrier介绍

在使用CyclicBarrier时，需要创建一个CyclicBarrier对象，构造函数需要一个整数作为参数，这个参数是一个“目标”，在CyclicBarrier对象创建后，内部会有一个计数器，初始值为0，CyclicBarrier对象的await方法每被调用一次，这个计数器就会加1，一旦这个计数器的值达到设定的“目标”，所有被CyclicBarrier.await阻塞住的线程都会继续执行。这个目标是固定的，一旦设定便不能修改。

举一个例子，假设有5个人爬香山，他们要爬到山顶，等到5个人到齐了再同时出发下山，那么我们要在山顶设定一个“目标”，同时还有一个计数器，这个目标就是5，每一个人到山顶后，这个人就要等待，同时计数器加1，等到5个人到齐了，也就是计数器达到了这个“目标”，所有等待的人就开始下山了。 更多内容请阅读《并发编程之CyclicBarrier原理与使用》

CountDownLathch介绍

使用CountDownLatch时，需要创建一个CountDownLatch对象，构造函数也需要一个整数作为参数，可以把这个参数想象成一个倒计时器，CountDownLatch对象本身是一个发令枪，所有调用CountDownLatch.await方法的线程都会等待发令枪的指令，一旦倒计时器为0，这些线程同时开始执行，而CountDownLatch.countDown方法就是为倒计时器减1。

更多内容请阅读《并发编程之CountDownLatch原理与使用》

对比分析

CyclicBarrier和CountDownLatch的共同点都是有一个目标和一个计数器，等到计数器达到目标后，所有阻塞的线程都将继续执行。它们的不同点是CyclicBarrier.await在等待的同时还修改计数器，而CountDownLatch.await只负责等待，CountDownLatch.countDown才修改计数器。

CountDownLatch和CyclicBarrier都能够实现线程之间的等待，只不过它们侧重点不同：

  • CountDownLatch一般用于一个或多个线程，等待其他线程执行完任务后，再才执行;
  • CyclicBarrier一般用于一组线程互相等待至某个状态，然后这一组线程再同时执行;
  • CountDownLatch 是一次性的，CyclicBarrier 是可循环利用的;
  • CountDownLathch是一个计数器，线程完成一个记录一个，计数器递减，只能用一次。如下图：

CyclicBarrier的计数器更像一个阀门，需要所有线程都到达，然后继续执行，计数器递减，提供reset功能，可以多次使用。如下图：

Phaser是什么?

Phaser，翻译为移相器(阶段)，它适用于这样一种场景，一个大任务可以分为多个阶段完成，且每个阶段的任务可以多个线程并发执行，但是必须上一个阶段的任务都完成了才可以执行下一个阶段的任务。

这种场景虽然使用CyclicBarrier 或者 CountDownLatch 也可以实现，但是要复杂的多，首先，具体需要多少个阶段是可能变的，其次，每个阶段的任务数也可能会变的。相比于CyclicBarrier 和 CountDownLath ，Phaser更加灵活更加方便。

Phaser使用方法

Phaser同时包含CyclicBarrier和CountDownLatch两个类的功能。

  • Phaser的arrive方法将将计数器加1，awaitAdvance将线程阻塞，直到计数器达到目标，这两个方法与CountDownLatch的countDown和await方法相对应;
  • Phaser的arriveAndAwaitAdvance方法将计数器加1的同时将线程阻塞，直到计数器达到目标后继续执行，这个方法对应CyclicBarrier的await方法。

除了包含以上两个类的功能外，Phaser还提供了更大的灵活性。CyclicBarrier和CountdownLatch在构造函数指定目标后就无法修改，而Phaser提供了register和deregister方法可以对目标进行动态修改。

下面看一个最简单的使用案例：

 
 
 
 
    
  
  
  
  
  1. package com.niuh.tools;
    2. 
  
  
  
  
  2. 
  
  
  
  
  3. import java.util.concurrent.Phaser;
    4. 
  
  
  
  
  4. 
  
  
  
  
  5. /**
    6. 
  
  
  
  
  6.  * 
    7. 
  
  
  
  
  7.  * Phaser示例
    8. 
  
  
  
  
  8.  * 
    9. 
  
  
  
  
  9.  */
    10. 
  
  
  
  
  10. public class PhaserRunner {
    11. 
  
  
  
  
  11.     // 定义每个阶段需要执行3个小任务
    12. 
  
  
  
  
  12.     public static final int PARTIES = 3;
    13. 
  
  
  
  
  13.     // 定义需要4个阶段完成的大任务
    14. 
  
  
  
  
  14.     public static final int PHASES = 4;
    15. 
  
  
  
  
  15. 
  
  
  
  
  16.     public static void main(String[] args) {
    17. 
  
  
  
  
  17. 
  
  
  
  
  18.         Phaser phaser = new Phaser(PARTIES) {
    19. 
  
  
  
  
  19.             @Override
    20. 
  
  
  
  
  20.             protected boolean onAdvance(int phase, int registeredParties) {
    21. 
  
  
  
  
  21.                 System.out.println("==phase: " + phase + " finished==");
    22. 
  
  
  
  
  22.                 return super.onAdvance(phase, registeredParties);
    23. 
  
  
  
  
  23.             }
    24. 
  
  
  
  
  24.         };
    25. 
  
  
  
  
  25. 
  
  
  
  
  26.         for (int i = 0; i < PARTIES; i++) {
    27. 
  
  
  
  
  27.             new Thread(() -> {
    28. 
  
  
  
  
  28.                 for (int j = 0; j < PHASES; j++) {
    29. 
  
  
  
  
  29.                     System.out.println(String.format("%s: phase: %d", Thread.currentThread().getName(), j));
    30. 
  
  
  
  
  30.                     phaser.arriveAndAwaitAdvance();
    31. 
  
  
  
  
  31.                 }
    32. 
  
  
  
  
  32.             }, "Thread " + i).start();
    33. 
  
  
  
  
  33.         }
    34. 
  
  
  
  
  34.     }
    35. 
  
  
  
  
  35. }
    36.

 这里我们定义个需要4个阶段完成的大任务，每个阶段需要3个小任务，针对这些小任务，我们分别起3个线程来执行这些小任务，查看输出结果为：

  • Thread 2: phase: 0
  • Thread 0: phase: 0
  • Thread 1: phase: 0
  • ==phase: 0 finished==
  • Thread 2: phase: 1
  • Thread 1: phase: 1
  • Thread 0: phase: 1
  • ==phase: 1 finished==
  • Thread 1: phase: 2
  • Thread 2: phase: 2
  • Thread 0: phase: 2
  • ==phase: 2 finished==
  • Thread 1: phase: 3
  • Thread 0: phase: 3
  • Thread 2: phase: 3
  • ==phase: 3 finished==

可以看到，每个阶段都是三个线程都完成来才进入下一个阶段。这是怎么实现的呢?

Phaser原理猜测

结合AQS的原理，大概猜测一下Phaser的实现原理：

  • 首先，需要存储当前阶段phase、当前阶段的任务数(参与者)parties、未完成参与者的数量，这三个变量我们可以放在一个变量state中存储。
  • 其次，需要一个队列存储先完成的参与者，当最后一个参与者完成任务时，需要唤醒队列中的参与者。

结合上面的案例带入：初始时当前阶段为0，参与者为3个，未完成参与者数为3;

  • 第一个线程执行到 phaser.arriveAndAwaitAdvance(); 时进入队列;
  • 第二个线程执行到 phaser.arriveAndAwaitadvance(); 时进入队列;
  • 第三个线程执行到 phaser.arriveAndAwaitadvance(); 时先执行这个阶段的总结 onAdvance(), 再唤醒签名两个线程继续执行下一个阶段的任务。

基于这样的一个思路，整体能说的通，至于是不是这样?让我们一起来看源码吧。

Phaser源码分析

主要API

  1. register()，增加一个参与者，需要同时增加parties和unarrived两个数值，也就是state中的16位和低16位
  2. onAdvance(int phase, int registeredParties)，当前阶段所有线程完成时，会调用OnAdvance()
  3. bulkRegister(int parties)，指定参与者数目注册到Phaser中，同时增加parties和unarrived两个数值
  4. arrive()，作用使parties值加1，并且不在屏障处等待，直接运行下面的代码
  5. awaitAdvance(int phase)，如果传入的参数与当前阶段一致，这个方法会将当前线程置于休眠，直到这个阶段的参与者都完成运行。如果传入的阶段参数与当前阶段不一致，立即返回
  6. arriveAndAwaitAdvance()，当前线程当前阶段执行完毕，等待其它线程完成当前阶段
  7. arriveAndDeregister()，当一个线程调用来此方法时，parties将减1，并且通知这个线程已经完成来当前预警，不会参加到下一个阶段中，因此Phaser对象在开始下一个阶段时不会等待这个线程。
  8. awaitAdvanceInterruptibly(int phase)，这个方法跟awaitAdvance(int phase)一样，不同之处是，如果这个方法中休眠的线程被中断，它将抛出InterruptedException异常。
  9. getPhase()，当前阶段
  10. getRegisteredParties()，总数
  11. getArrivedParties()，到达总数
  12. getUnarrivedParties()，未到达总数

内部类QNode

QNode用来跟踪当前线程的信息的。QNode被组织成单向链表的形式。用来管理是否阻塞或者被中断。

QNode继承自ForkJoinPool.ManagedBlocker。ForkJoinPool来管理是否阻塞和中断状态。这里只需要重写isReleasable和block。

  • isReleaseable用于判断是否释放当前节点。
  • block用于阻塞。
 
 
 
 
    
  
  
  
  
  1. static final class QNode implements ForkJoinPool.ManagedBlocker {
    2. 
  
  
  
  
  2.         final Phaser phaser;
    3. 
  
  
  
  
  3.         final int phase;
    4. 
  
  
  
  
  4.         final boolean interruptible;
    5. 
  
  
  
  
  5.         final boolean timed;
    6. 
  
  
  
  
  6.         boolean wasInterrupted;
    7. 
  
  
  
  
  7.         long nanos;
    8. 
  
  
  
  
  8.         final long deadline;
    9. 
  
  
  
  
  9.         volatile Thread thread; // nulled to cancel wait
    10. 
  
  
  
  
  10.         QNode next;
    11. 
  
  
  
  
  11. 
  
  
  
  
  12.         QNode(Phaser phaser, int phase, boolean interruptible,
    13. 
  
  
  
  
  13.               boolean timed, long nanos) {
    14. 
  
  
  
  
  14.             this.phaser = phaser;
    15. 
  
  
  
  
  15.             this.phase = phase;
    16. 
  
  
  
  
  16.             this.interruptible = interruptible;
    17. 
  
  
  
  
  17.             this.nanos = nanos;
    18. 
  
  
  
  
  18.             this.timed = timed;
    19. 
  
  
  
  
  19.             this.deadline = timed ? System.nanoTime() + nanos : 0L;
    20. 
  
  
  
  
  20.             thread = Thread.currentThread();
    21. 
  
  
  
  
  21.         }
    22. 
  
  
  
  
  22. 
  
  
  
  
  23.         public boolean isReleasable() {
    24. 
  
  
  
  
  24.             if (thread == null)
    25. 
  
  
  
  
  25.                 return true;
    26. 
  
  
  
  
  26.             if (phaser.getPhase() != phase) {
    27. 
  
  
  
  
  27.                 thread = null;
    28. 
  
  
  
  
  28.                 return true;
    29. 
  
  
  
  
  29.             }
    30. 
  
  
  
  
  30.             if (Thread.interrupted())
    31. 
  
  
  
  
  31.                 wasInterrupted = true;
    32. 
  
  
  
  
  32.             if (wasInterrupted && interruptible) {
    33. 
  
  
  
  
  33.                 thread = null;
    34. 
  
  
  
  
  34.                 return true;
    35. 
  
  
  
  
  35.             }
    36. 
  
  
  
  
  36.             if (timed) {
    37. 
  
  
  
  
  37.                 if (nanos > 0L) {
    38. 
  
  
  
  
  38.                     nanos = deadline - System.nanoTime();
    39. 
  
  
  
  
  39.                 }
    40. 
  
  
  
  
  40.                 if (nanos <= 0L) {
    41. 
  
  
  
  
  41.                     thread = null;
    42. 
  
  
  
  
  42.                     return true;
    43. 
  
  
  
  
  43.                 }
    44. 
  
  
  
  
  44.             }
    45. 
  
  
  
  
  45.             return false;
    46. 
  
  
  
  
  46.         }
    47. 
  
  
  
  
  47. 
  
  
  
  
  48.         public boolean block() {
    49. 
  
  
  
  
  49.             if (isReleasable())
    50. 
  
  
  
  
  50.                 return true;
    51. 
  
  
  
  
  51.             else if (!timed)
    52. 
  
  
  
  
  52.                 LockSupport.park(this);
    53. 
  
  
  
  
  53.             else if (nanos > 0L)
    54. 
  
  
  
  
  54.                 LockSupport.parkNanos(this, nanos);
    55. 
  
  
  
  
  55.             return isReleasable();
    56. 
  
  
  
  
  56.         }
    57. 
  
  
  
  
  57.     }
    58.

 整体代码比较简单。要注意的是在isReleasable中使用了thread=null来使得避免解锁任务。使用方法类似于internalAwaitAdvance中的用法。先完成的参与者放入队列中的节点，这里我们只需要关注 thread 和 next两个属性即可，很明显这是一个单链表，存储这入队的线程。

主要属性

 
 
 
 
    
  
  
  
  
  1. /*
    2. 
  
  
  
  
  2.  * unarrived  -- 还没有抵达屏障的参与者的个数 (bits 0-15)
    3. 
  
  
  
  
  3.  * parties    -- 需要等待的参与者的个数      (bits 16-31)
    4. 
  
  
  
  
  4.  * phase      -- 屏障所处的阶段             (bits 32-62)
    5. 
  
  
  
  
  5.  * terminated -- 屏障的结束标记             (bit 63 / sign)
    6. 
  
  
  
  
  6.  */
    7. 
  
  
  
  
  7. // 状态变量，用于存储当前阶段phase、参与者数parties、未完成的参与者数unarrived_count 
    8. 
  
  
  
  
  8. private volatile long state;
    9. 
  
  
  
  
  9. // 最多可以有多少个参与者，即每个阶段最多有多少个任务
    10. 
  
  
  
  
  10. private static final int  MAX_PARTIES     = 0xffff;
    11. 
  
  
  
  
  11. // 最多可以有多少阶段
    12. 
  
  
  
  
  12. private static final int  MAX_PHASE       = Integer.MAX_VALUE;
    13. 
  
  
  
  
  13. // 参与者数量的偏移量
    14. 
  
  
  
  
  14. private static final int  PARTIES_SHIFT   = 16;
    15. 
  
  
  
  
  15. // 当前阶段的偏移量
    16. 
  
  
  
  
  16. private static final int  PHASE_SHIFT     = 32;
    17. 
  
  
  
  
  17. // 未完成的参与者数的掩码，低16位
    18. 
  
  
  
  
  18. private static final int  UNARRIVED_MASK  = 0xffff;      // to mask ints
    19. 
  
  
  
  
  19. // 参与者数，中间16位
    20. 
  
  
  
  
  20. private static final long PARTIES_MASK    = 0xffff0000L; // to mask longs
    21. 
  
  
  
  
  21. // counts的掩码，counts等于参与者数和未完成的参与者数的 '｜' 操作
    22. 
  
  
  
  
  22. private static final long COUNTS_MASK     = 0xffffffffL;
    23. 
  
  
  
  
  23. private static final long TERMINATION_BIT = 1L << 63;
    24. 
  
  
  
  
  24. 
  
  
  
  
  25. // 一次一个参与者完成
    26. 
  
  
  
  
  26. private static final int  ONE_ARRIVAL     = 1;
    27. 
  
  
  
  
  27. // 增加减少参与者时使用
    28. 
  
  
  
  
  28. private static final int  ONE_PARTY       = 1 << PARTIES_SHIFT;
    29. 
  
  
  
  
  29. // 减少参与者时使用
    30. 
  
  
  
  
  30. private static final int  ONE_DEREGISTER  = ONE_ARRIVAL|ONE_PARTY;
    31. 
  
  
  
  
  31. // 没有参与者使用
    32. 
  
  
  
  
  32. private static final int  EMPTY           = 1;
    33. 
  
  
  
  
  33. // 用于求未完成参与者数量
    34. 
  
  
  
  
  34. private static int unarrivedOf(long s) {
    35. 
  
  
  
  
  35.  int counts = (int)s;
    36. 
  
  
  
  
  36.     return (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
    37. 
  
  
  
  
  37. }
    38. 
  
  
  
  
  38. // 用于求参与者数量（中间16位），注意int的为止
    39. 
  
  
  
  
  39. private static int partiesOf(long s) {
    40. 
  
  
  
  
  40.  return (int)s >>> PARTIES_SHIFT;
    41. 
  
  
  
  
  41. }
    42. 
  
  
  
  
  42. // 用于求阶段数（高32位），注意int的位置
    43. 
  
  
  
  
  43. private static int phaseOf(long s) {
    44. 
  
  
  
  
  44.  return (int)(s >>> PHASE_SHIFT);
    45. 
  
  
  
  
  45. }
    46. 
  
  
  
  
  46. // 已完成参与者数量
    47. 
  
  
  
  
  47. private static int arrivedOf(long s) {
    48. 
  
  
  
  
  48.  int counts = (int)s;
    49. 
  
  
  
  
  49.  return (counts == EMPTY) ? 0 :
    50. 
  
  
  
  
  50.   (counts >>> PARTIES_SHIFT) - (counts & UNARRIVED_MASK);
    51. 
  
  
  
  
  51. }
    52. 
  
  
  
  
  52. 
  
  
  
  
  53. /**
    54. 
  
  
  
  
  54. * The parent of this phaser, or null if none
    55. 
  
  
  
  
  55. */
    56. 
  
  
  
  
  56. private final Phaser parent;
    57. 
  
  
  
  
  57. 
  
  
  
  
  58. /**
    59. 
  
  
  
  
  59. * The root of phaser tree. Equals this if not in a tree.
    60. 
  
  
  
  
  60. */
    61. 
  
  
  
  
  61. private final Phaser root;
    62. 
  
  
  
  
  62. 
  
  
  
  
  63. // 用于存储已经=完成参与者所在的线程，根据当前阶段的奇偶性选择不同的队列
    64. 
  
  
  
  
  64. private final AtomicReference evenQ;
    65. 
  
  
  
  
  65. private final AtomicReference oddQ;
    66.

主要属性位 state 和 evenQ 及 oddQ

  • state，volatile的long来表示状态变量，高32位存储当前阶段phase，中间16位存储参与者的数量，低16位存储未完成参与者的数量。
  • unarrived -- 还没有抵达屏障的参与者的个数 (bits 0-15)
  • parties -- 需要等待的参与者的个数 (bits 16-31)
  • phase -- 屏障所处的阶段 (bits 32-62)
  • terminated -- 屏障的结束标记 (bit 63 / sign)

如果是空状态，也就是没有子阶段注册的初始阶段。这里用一个EMPTY状态表示，也就是0个子阶段和一个未到达阶段。

所有的状态变化都是通过CAS操作执行的，唯一例外是注册一个子相移器(sub-Phaser)，用于构成树的，也就是Phaser的父Phaser非空。这个子相移器的分阶段是通过一个内置锁来设置。

  • evenQ 和 oddQ，是根据phaser的奇偶状态来设置的，用来存储等待的线程。为了避免竞争，这里使用了Phaser的数值奇偶来存储，此外对于子相移器，它与其根相移器使用同一个evenQ或者oddQ，以加速释放。

构造方法

 
 
 
 
    
  
  
  
  
  1. public Phaser() {
    2. 
  
  
  
  
  2.  this(null, 0);
    3. 
  
  
  
  
  3. }
    4. 
  
  
  
  
  4. 
  
  
  
  
  5. public Phaser(int parties) {
    6. 
  
  
  
  
  6.  this(null, parties);
    7. 
  
  
  
  
  7. }
    8. 
  
  
  
  
  8. 
  
  
  
  
  9. public Phaser(Phaser parent) {
    10. 
  
  
  
  
  10.  this(parent, 0);
    11. 
  
  
  
  
  11. }
    12. 
  
  
  
  
  12. 
  
  
  
  
  13. public Phaser(Phaser parent, int parties) {
    14. 
  
  
  
  
  14.  if (parties >>> PARTIES_SHIFT != 0)
    15. 
  
  
  
  
  15.   throw new IllegalArgumentException("Illegal number of parties");
    16. 
  
  
  
  
  16.     int phase = 0;
    17. 
  
  
  
  
  17.     this.parent = parent;
    18. 
  
  
  
  
  18.     if (parent != null) { // 父phaser不为空
    19. 
  
  
  
  
  19.      final Phaser root = parent.root;
    20. 
  
  
  
  
  20.         this.root = root; // 指向root phaser
    21. 
  
  
  
  
  21.         this.evenQ = root.evenQ; // 两个栈，整个phaser链只有一份
    22. 
  
  
  
  
  22.         this.oddQ = root.oddQ;
    23. 
  
  
  
  
  23.         if (parties != 0)
    24. 
  
  
  
  
  24.          phase = parent.doRegister(1); // 向父phaser注册当前线程
    25. 
  
  
  
  
  25.     }
    26. 
  
  
  
  
  26.     else {
    27. 
  
  
  
  
  27.      this.root = this; // 否则，自己是root phaser
    28. 
  
  
  
  
  28.         this.evenQ = new AtomicReference(); // 负责创建两个栈（QNode链）
    29. 
  
  
  
  
  29.         this.oddQ = new AtomicReference();
    30. 
  
  
  
  
  30.     }
    31. 
  
  
  
  
  31.     // 状态变量state的存储分为三段
    32. 
  
  
  
  
  32.     this.state = (parties == 0) ? (long)EMPTY :
    33. 
  
  
  
  
  33.             ((long)phase << PHASE_SHIFT) |
    34. 
  
  
  
  
  34.             ((long)parties << PARTIES_SHIFT) |
    35. 
  
  
  
  
  35.             ((long)parties);
    36. 
  
  
  
  
  36. }
    37.

 构造函数中还有一个parent和root，这是用来构造多层级阶段的，用于构成树的。

重点还是还是看state的赋值方式，高32位存储当前阶段phase，中间16位存储参与者的数量，低16位存储未完成参与者的数量。

主要方法

下面我们一起来看看几个主要方法的源码，重点是三个private的核心方法：doArrive、doRegister、reconcileState

register方法

增加一个参与者，需要同时增加parties和unarrived两个数值，也就是state中的16位和低16位(中间16位存储参与者的数量，低16位存储未完成参与者的数量)

 
 
 
 
    
  
  
  
  
  1. public int register() {
    2. 
  
  
  
  
  2.  return doRegister(1);
    3. 
  
  
  
  
  3. }
    4.

 这里主要调用的是doRegister方法，我们往下看。

doRegister方法

 
 
 
 
    
  
  
  
  
  1. private int doRegister(int registrations) {
    2. 
  
  
  
  
  2.     // adjustment to state
    3. 
  
  
  
  
  3.     // state应该加的值，注意这里是相当于同时增加parties和unarrived
    4. 
  
  
  
  
  4.     long adjust = ((long)registrations << PARTIES_SHIFT) | registrations; //计算出需要调整的量
    5. 
  
  
  
  
  5.     final Phaser parent = this.parent; //查看可能存在的相移器
    6. 
  
  
  
  
  6.     int phase;
    7. 
  
  
  
  
  7.     for (;;) {
    8. 
  
  
  
  
  8.         // state的值
    9. 
  
  
  
  
  9.         long s = (parent == null) ? state : reconcileState(); // reconcileState()方法是调整当前phaser的状态与root的一致
    10. 
  
  
  
  
  10.         // state的低32未，也就是parties和unarrived的值
    11. 
  
  
  
  
  11.         int counts = (int)s;
    12. 
  
  
  
  
  12.         // parties的值
    13. 
  
  
  
  
  13.         int parties = counts >>> PARTIES_SHIFT;
    14. 
  
  
  
  
  14.         // unarrived的值
    15. 
  
  
  
  
  15.         int unarrived = counts & UNARRIVED_MASK;
    16. 
  
  
  
  
  16.         // 检查是否溢出
    17. 
  
  
  
  
  17.         if (registrations > MAX_PARTIES - parties) //如果需要注册的数量超过运行注册的最大值，则抛出异常状态异常
    18. 
  
  
  
  
  18.             throw new IllegalStateException(badRegister(s));
    19. 
  
  
  
  
  19.   // 当前阶段phase
    20. 
  
  
  
  
  20.         phase = (int)(s >>> PHASE_SHIFT);
    21. 
  
  
  
  
  21.         if (phase < 0) //如果当前状态为终止状态则跳出循环直接退出
    22. 
  
  
  
  
  22.             break;
    23. 
  
  
  
  
  23.         // 不是第一个参与者
    24. 
  
  
  
  
  24.         if (counts != EMPTY) {          // not 1st registration //如果当前状态不是第一次注册线程
    25. 
  
  
  
  
  25.             if (parent == null || reconcileState() == s) { //如果当相移器的父相移器为空，则直接信息CAS，如果当前相移器部位空则调用reconcileState处理，这个稍后再看。reconcileState这里主要为了防止出现同步性错误。
    26. 
  
  
  
  
  26.                 // unarrived等于0说明当前阶段正在执行onAdvance()方法，等待其执行完毕
    27. 
  
  
  
  
  27.                 if (unarrived == 0)             // wait out advance
    28. 
  
  
  
  
  28.                     root.internalAwaitAdvance(phase, null);
    29. 
  
  
  
  
  29.                 // 否则就修改state的值，增加adjust，如果成功就跳出循环
    30. 
  
  
  
  
  30.                 else if (UNSAFE.compareAndSwapLong(this, stateOffset,
    31. 
  
  
  
  
  31.                                                    s, s + adjust))
    32. 
  
  
  
  
  32.                     break;
    33. 
  
  
  
  
  33.             }
    34. 
  
  
  
  
  34.         }
    35. 
  
  
  
  
  35.         // 是第一个参与者，当前状态是第一次注册。如果如果当前相移器没有父相移器。则直接进行CAS
    36. 
  
  
  
  
  36.         else if (parent == null) {          // 1st root registration
    37. 
  
  
  
  
  37.             // 计算state的值
    38. 
  
  
  
  
  38.             long next = ((long)phase << PHASE_SHIFT) | adjust;
    39. 
  
  
  
  
  39.             // 修改state的值，如果成功就跳出循环
    40. 
  
  
  
  
  40.             if (UNSAFE.compareAndSwapLong(this, stateOffset, s, next))
    41. 
  
  
  
  
  41.              break;
    42. 
  
  
  
  
  42.             }
    43. 
  
  
  
  
  43.         else { // 如果当前是第一次设置，并且该相移器被组织在一个树中则需要考虑一下，则需要使用内置锁来进如
    44. 
  
  
  
  
  44.             // 多层级阶段的处理方式
    45. 
  
  
  
  
  45.             synchronized (this) {               // 1st sub registration
    46. 
  
  
  
  
  46.                 if (state == s) {               // recheck under lock 这里有可能发生竞争。所以这里还需要检查一下，如果失败则需退出同步区重新尝试进入。
    47. 
  
  
  
  
  47.                     phase = parent.doRegister(1); // 调用其父相移器的注册方法
    48. 
  
  
  
  
  48.                     if (phase < 0)
    49. 
  
  
  
  
  49.                         break;
    50. 
  
  
  
  
  50.                     // finish registration whenever parent registration
    51. 
  
  
  
  
  51.                     // succeeded, even when racing with termination,
    52. 
  
  
  
  
  52.                     // since these are part of the same "transaction".
    53. 
  
  
  
  
  53.                     while (!UNSAFE.compareAndSwapLong
    54. 
  
  
  
  
  54.                            (this, stateOffset, s,
    55. 
  
  
  
  
  55.                             ((long)phase << PHASE_SHIFT) | adjust)) {
    56. 
  
  
  
  
  56.                         s = state;
    57. 
  
  
  
  
  57.                         phase = (int)(root.state >>> PHASE_SHIFT);
    58. 
  
  
  
  
  58.                         // assert (int)s == EMPTY;
    59. 
  
  
  
  
  59.                     }
    60. 
  
  
  
  
  60.                     break;
    61. 
  
  
  
  
  61.                 }
    62. 
  
  
  
  
  62.             }
    63. 
  
  
  
  
  63.         }
    64. 
  
  
  
  
  64.     }
    65. 
  
  
  
  
  65.     return phase;
    66. 
  
  
  
  
  66. }
    67.

 增加一个参与者的总体的逻辑为：

  • 增加一个参与者，需要同时增加parties和unarrived两个数值，也就是state中的16位和低16位;
  • 如果是第一个参与者，则尝试原子更新state的值，如果成功了就退出;
  • 如果不是第一个参与者，则检查是不是在执行onAdvance() , 如果是等待onAdvance() 执行完成，如果否则尝试原子更新state的值，直到成功退出;
  • 等待onAdvance() 完成是采用先自旋后进入队列排队的方式等待，减少线程上下文切换;

arriveAndAwaitAdvance()方法

当前线程当前阶段执行完毕，等待其他线程完成当前阶段。 如果当前线程是该阶段最后一个到达的，则当前线程会执行onAdvance()方法，并唤醒其它线程进入下一个阶段。

 
 
 
 
    
  
  
  
  
  1. public int arriveAndAwaitAdvance() {
    2. 
  
  
  
  
  2.      // Specialization of doArrive+awaitAdvance eliminating some reads/paths
    3. 
  
  
  
  
  3.      final Phaser root = this.root;
    4. 
  
  
  
  
  4.      for (;;) {
    5. 
  
  
  
  
  5.          // state的值
    6. 
  
  
  
  
  6.          long s = (root == this) ? state : reconcileState();
    7. 
  
  
  
  
  7.          // 当前阶段
    8. 
  
  
  
  
  8.          int phase = (int)(s >>> PHASE_SHIFT);
    9. 
  
  
  
  
  9.          if (phase < 0)
    10. 
  
  
  
  
  10.              return phase;
    11. 
  
  
  
  
  11.          // parties 和 unarrived的值
    12. 
  
  
  
  
  12.          int counts = (int)s;
    13. 
  
  
  
  
  13.          // unarrived的值（state的低16位）
    14. 
  
  
  
  
  14.          int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
    15. 
  
  
  
  
  15.          if (unarrived <= 0)
    16. 
  
  
  
  
  16.              throw new IllegalStateException(badArrive(s));
    17. 
  
  
  
  
  17.          // 修改state的值
    18. 
  
  
  
  
  18.          if (UNSAFE.compareAndSwapLong(this, stateOffset, s,
    19. 
  
  
  
  
  19.                                        s -= ONE_ARRIVAL)) {
    20. 
  
  
  
  
  20.              // 如果不是最后一个到达的，则调用internalAwaitAdvance()方法自旋或进入队列等待
    21. 
  
  
  
  
  21.              if (unarrived > 1)
    22. 
  
  
  
  
  22.                  // 这里是直接返回了，internalAwaitAdvance()方法的源码见register()方法解析
    23. 
  
  
  
  
  23.                  return root.internalAwaitAdvance(phase, null);
    24. 
  
  
  
  
  24.              // 到这里说明是最后一个到达的参与者
    25. 
  
  
  
  
  25.              if (root != this)
    26. 
  
  
  
  
  26.                  return parent.arriveAndAwaitAdvance();
    27. 
  
  
  
  
  27.              // n 只保留了state中parties的部分，也就是中16位
    28. 
  
  
  
  
  28.              long n = s & PARTIES_MASK;  // base of next state
    29. 
  
  
  
  
  29.              // parties的值，即下一次需要到达的参与者数量
    30. 
  
  
  
  
  30.              int nextUnarrived = (int)n >>> PARTIES_SHIFT;
    31. 
  
  
  
  
  31.              // 执行onAdvance()方法，返回true表示下一阶段参与者数量为0了，也就是结束了
    32. 
  
  
  
  
  32.              if (onAdvance(phase, nextUnarrived))
    33. 
  
  
  
  
  33.                  n |= TERMINATION_BIT;
    34. 
  
  
  
  
  34.              else if (nextUnarrived == 0)
    35. 
  
  
  
  
  35.                  n |= EMPTY;
    36. 
  
  
  
  
  36.              else
    37. 
  
  
  
  
  37.                  n |= nextUnarrived; // n加上unarrived的值
    38. 
  
  
  
  
  38.              // 下阶段等待当前阶段加1
    39. 
  
  
  
  
  39.              int nextPhase = (phase + 1) & MAX_PHASE;
    40. 
  
  
  
  
  40.              // n 加上下一个阶段的值
    41. 
  
  
  
  
  41.              n |= (long)nextPhase << PHASE_SHIFT;
    42. 
  
  
  
  
  42.              // 修改state的值为n
    43. 
  
  
  
  
  43.              if (!UNSAFE.compareAndSwapLong(this, stateOffset, s, n))
    44. 
  
  
  
  
  44.                  return (int)(state >>> PHASE_SHIFT); // terminated
    45. 
  
  
  
  
  45.              // 唤醒其它参与者并进入下一个阶段
    46. 
  
  
  
  
  46.              releaseWaiters(phase);
    47. 
  
  
  
  
  47.              // 返回下一阶段的值
    48. 
  
  
  
  
  48.              return nextPhase;
    49. 
  
  
  
  
  49.          }
    50. 
  
  
  
  
  50.      }
    51. 
  
  
  
  
  51.  }
    52.

 arriveAndAwaitAdvance的大致逻辑为：

  • 修改state中unarrived部分的值减1;
  • 如果不是最后一个到达，则调用internalAwaitAdvance() 方法自旋或排队等待;
  • 如果是最后一个到达的，则调用onAdvance() 方法，然后修改state的值为下一阶段对应的值，并唤醒其它等待的线程;
  • 返回下一阶段俄值。

internalAwaitAdvance方法

internalAwaitAdvance方法。实际上Phaser中阻塞都是通过这个语句实现的。这个语句必须通过根相移器调用。换句话说所有的阻塞都是在根相移器阻塞的。

输入参数中phase是需要阻塞的阶段。node是用来跟踪可能中断的阻塞节点。

 
 
 
 
    
  
  
  
  
  1. // 等待onAdvance()方法执行完毕
    2. 
  
  
  
  
  2. // 原理是先自旋一定次数，如果进入下一个阶段，这个方法直接返回了，
    3. 
  
  
  
  
  3. // 如果自旋一定次数还没有进入下一个阶段，则当前线程入队列，等待onAdvance()执行完成唤醒
    4. 
  
  
  
  
  4. private int internalAwaitAdvance(int phase, QNode node) {
    5. 
  
  
  
  
  5.     // assert root == this;
    6. 
  
  
  
  
  6.     // 保证队列为空
    7. 
  
  
  
  
  7.     releaseWaiters(phase-1);      // ensure old queue clean
    8. 
  
  
  
  
  8.     boolean queued = false;       // true when node is enqueued
    9. 
  
  
  
  
  9.     int lastUnarrived = 0;        // to increase spins upon change
    10. 
  
  
  
  
  10.     // 自旋的次数
    11. 
  
  
  
  
  11.     int spins = SPINS_PER_ARRIVAL;
    12. 
  
  
  
  
  12.     long s;
    13. 
  
  
  
  
  13.     int p;
    14. 
  
  
  
  
  14.     // 检查当前阶段是否变化，如果变化了说明进入下一个阶段了，这时候就没有必要自旋了
    15. 
  
  
  
  
  15.     while ((p = (int)((s = state) >>> PHASE_SHIFT)) == phase) {
    16. 
  
  
  
  
  16.         // 如果node为空，注册的时候传入的为空
    17. 
  
  
  
  
  17.         if (node == null) {           // spinning in noninterruptible mode
    18. 
  
  
  
  
  18.             // 未完成的参与者数量
    19. 
  
  
  
  
  19.             int unarrived = (int)s & UNARRIVED_MASK;
    20. 
  
  
  
  
  20.             // unarrived 有变化，增加自旋次数
    21. 
  
  
  
  
  21.             if (unarrived != lastUnarrived &&
    22. 
  
  
  
  
  22.                 (lastUnarrived = unarrived) < NCPU)
    23. 
  
  
  
  
  23.                 spins += SPINS_PER_ARRIVAL;
    24. 
  
  
  
  
  24.             boolean interrupted = Thread.interrupted();
    25. 
  
  
  
  
  25.             // 自旋次数万了，则新建一个节点
    26. 
  
  
  
  
  26.             if (interrupted || --spins < 0) { // need node to record intr
    27. 
  
  
  
  
  27.                 node = new QNode(this, phase, false, false, 0L);
    28. 
  
  
  
  
  28.                 node.wasInterrupted = interrupted;             
    
                分享名称：并发编程之Phaser原理与应用
                
    
                当前URL：http://www.shufengxianlan.com/qtweb/news1/273201.html
            
    
            
    
                网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等
                                
    
            
    
                广告
                
            
    
            
    
                声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源：
                创新互联

猜你还喜欢下面的内容

ChatGPT知识

各行业网站

棕树数据中心　　 　广汉锦华橡塑　　 　普特斯腐植酸　　 　网站排名　　 　成都网站制作　　 　保温橡塑管　　 　香港云虚拟主机　　 　四川乐山服务器托管　　 　四川艾名斯　　 　吉安网站建设　　 　新都四新网站建设　　 　郫都区消防器材　　 　成都网站设计　　 　网站ssl证书　　 　免备案虚拟主机　　 　彭州网站建设　　 　h5页面制作　　 　成都企业网站建设公司　　 　网站制作　　 　成都服务器托管