JUC系列同步工具类之Phaser
Posted 顧棟
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了JUC系列同步工具类之Phaser相关的知识，希望对你有一定的参考价值。
Phaser
文章目录
可重用的同步屏障，功能类似于 CyclicBarrier 和 CountDownLatch，但支持更灵活的使用。
主要机制
Registration（注册机制）
与其他屏障的情况不同，在Phaser上注册同步的线程数量可能会随着时间而变化。 可以随时注册任务（使用register、bulkRegister或建立初始所需线程数量的构造函数），并且可以选择在任何到达时取消注册（使用arriveAndDeregister）。 与大多数基本同步结构一样，注册和注销仅影响内部计数； 他们没有建立任何进一步的内部簿记，因此任务无法查询它们是否已注册。 （但是，您可以通过子类化此类来引入此类簿记。）
动态性-支持屏障所需线程的数量
Synchronization（同步机制）
像CyclicBarrier一样，Phaser可能会被反复等待。 方法arriveAndAwaitAdvance具有类似于CyclicBarrier.await的效果。 每一代Phaser都有一个相关的阶段编号(phase)。 阶段编号从零开始，当所有线程到达屏障时前进，在达到Integer.MAX_VALUE后重置0。 阶段编号的使用可以通过任何注册方可以调用的两种方法，在到达阶段和等待其他阶段时独立控制操作：
重入性-一个Phaser可以支持多个阶段同步操作
Arrival（到达机制）
方法arrive和到arriveAndDeregister记录到达。 这些方法不会阻塞，而是返回相关的到达阶段编号； 即，到达应用的Phaser的相位编号。 当给定阶段的最后一方到达时，将执行一个可选操作并且该阶段前进。 这些动作由触发阶段推进的一方执行，并通过覆盖方法onAdvance(int, int) 进行安排，该方法也支持控制终止。 覆盖此方法与向CyclicBarrier提供屏障操作类似，但更灵活。
Waiting（等待机制）
方法awaitAdvance需要一个指示到达阶段编号的参数，并在Phaser前进到（或已经处于）不同阶段时返回。 与使用CyclicBarrier的类似构造不同，即使等待线程被中断，方法awaitAdvance也会继续等待。 可中断和超时版本也可用，但在任务等待中断或超时时遇到的异常不会改变Phaser的状态。 如有必要，您可以在这些异常的处理程序中执行任何相关的恢复，通常是在调用forceTermination之后。 在ForkJoinPool中执行的任务也可以使用Phaser，这将确保在其他人被阻塞等待阶段推进时执行任务有足够的并行性。
Termination（终止机制）
Phaser可以进入终止状态，可以使用方法isTerminated进行检查。 终止后，所有同步方法立即返回，无需等待提前，返回值为负数。 同样，在终止时尝试注册也无效。 当onAdvance的调用返回 true 时触发终止。 如果取消注册导致注册方的数量变为零，则默认实现返回 true。 如下图所示，当Phaser控制具有固定迭代次数的动作时，通常可以方便地覆盖此方法以在当前阶段数达到阈值时导致终止。 方法forceTermination也可用于突然释放等待线程并允许它们终止。
Tiering（分层结构）
Phaser可以分层（即以树型结构构建）以减少争用。 可以改为设置具有大量参与方的Phaser，否则这些Phaser将经历沉重的同步争用成本，以便子Phaser组共享一个共同的父级。 这可能会大大增加吞吐量，即使它会产生更大的每次操作开销。
在分层Phaser树中，自动管理子移相器与其父级的注册和注销。 每当子移相器的注册方数量变为非零时（如在Phaser(Phaser, int)构造函数、register或bulkRegister中建立的那样），子Phaser将向其父移相器注册。 每当注册方的数量由于调用到达和取消注册而变为零时，子Phaser就会从其父Phaser中注销。
层次性-将多个Phaser树形结构组织起来，通过牺牲操作的开销增加吞吐量。
Monitoring（监控）
虽然同步方法只能由注册方调用，但Phaser的当前状态可以由任何调用者监视。 在任何给定时刻，总共有getRegisteredParties方，其中getArrivedParties已到达当前阶段 (getPhase)。 当剩余的 (getUnarrivedParties) 方到达时，阶段前进。 这些方法返回的值可能反映瞬态状态，因此通常对同步控制没有用处。 方法toString以一种便于非正式监控的形式返回这些状态查询的快照。
组成
内部类QNode
// 代表等待队列的 Treiber 堆栈的等待节点
static final class QNode implements ForkJoinPool.ManagedBlocker
final Phaser phaser;
final int phase;
final boolean interruptible;
final boolean timed;
boolean wasInterrupted;
long nanos;
final long deadline;
volatile Thread thread; // nulled to cancel wait
QNode next;
QNode(Phaser phaser, int phase, boolean interruptible,
boolean timed, long nanos)
this.phaser = phaser;
this.phase = phase;
this.interruptible = interruptible;
this.nanos = nanos;
this.timed = timed;
this.deadline = timed ? System.nanoTime() + nanos : 0L;
thread = Thread.currentThread();
public boolean isReleasable()
if (thread == null)
return true;
if (phaser.getPhase() != phase)
thread = null;
return true;
if (Thread.interrupted())
wasInterrupted = true;
if (wasInterrupted && interruptible)
thread = null;
return true;
if (timed)
if (nanos > 0L)
nanos = deadline - System.nanoTime();
if (nanos <= 0L)
thread = null;
return true;
return false;
public boolean block()
if (isReleasable())
return true;
else if (!timed)
LockSupport.park(this);
else if (nanos > 0L)
LockSupport.parkNanos(this, nanos);
return isReleasable();
成员变量
// 同步状态位
private volatile long state;
// 此移相器的父级，如果没有则为 null
private final Phaser parent;
// 移相器树的根。 如果不在树中，则等于此值。
private final Phaser root;
// 偶数阶段单向链表
private final AtomicReference<QNode> evenQ;
// 奇数阶段单向链表
private final AtomicReference<QNode> oddQ;
state状态位的含义，如下图
所有状态更新都通过 CAS 执行，除了子相位器的初始注册（即具有非 null 父级的子相位器）。在这种（相对罕见的）情况下，我们在首次向其父级注册时使用内置同步来锁定。子相位器的相位可以滞后于其祖先的相位，直到它被实际访问——参见方法 reconcileState。
在Phaser中使用奇偶两个单向链表来实现阻塞队列，降低操作的冲突。
构造函数
使用给定的父级和已注册的未到达方数量创建一个新的移相器。 当给定的父节点不为空并且给定的参与方数量大于零时，此子移相器将向其父节点注册。
/**
* @param parent 父移相器
* @param parties 进入下一阶段所需的参与方数量
*/
public Phaser(Phaser parent, int parties)
// 无符号右移16位--验证parties是否在合理范围内
if (parties >>> PARTIES_SHIFT != 0)
throw new IllegalArgumentException("Illegal number of parties");
// 阶段0
int phase = 0;
this.parent = parent;
if (parent != null)
// 获取父阶段的根节点
final Phaser root = parent.root;
this.root = root;
// 偶数head
this.evenQ = root.evenQ;
// 奇数head
this.oddQ = root.oddQ;
// 成员量不为0，需要注册一个阶段
if (parties != 0)
phase = parent.doRegister(1);
// 不存在实际父阶段器
else
this.root = this;
this.evenQ = new AtomicReference<QNode>();
this.oddQ = new AtomicReference<QNode>();
// 设置同步状态值--最终二进制位数含义和分布，如上图所示。
this.state = (parties == 0) ? (long)EMPTY :
((long)phase << PHASE_SHIFT) |
((long)parties << PARTIES_SHIFT) |
((long)parties);
核心方法
| 方法名 | 描述 |
|---|---|
| int register() | 向此移相器添加一个新的未到方。 如果正在进行的 onAdvance调用正在进行中，则此方法可能会在返回之前等待其完成。如果此移相器有父移相器，并且此移相器之前没有注册方，则此子移相器也向其父移相器注册。 如果此移相器终止，则注册尝试无效，并返回负值。 |
| int bulkRegister(int parties) | 向此移相器批量添加新的未到方。 如果正在进行的 onAdvance 调用正在进行中，则此方法可能会在返回之前等待其完成。 如果此移相器有父级，并且给定的参与方数量大于零，并且此移相器之前没有注册方，则此子移相器也向其父级注册。 如果此移相器终止，则注册尝试无效，并返回负值。 |
| int arrive() | 当前线程到达此移相器，无需等待其他人到达。返回arrival phase number。 未注册方调用此方法属于使用错误。 但是，此错误可能仅在此移相器上的某些后续操作（如果有）时导致 IllegalStateException。 |
| int arriveAndDeregister() | 到达此移相器并从中注销，而无需等待其他人到达。返回arrival phase number。 取消注册减少了在未来阶段推进所需的参与方数量。 如果此移相器有父级，并且注销导致此移相器的参与方为零，则此移相器也会从其父级中注销。 未注册方调用此方法属于使用错误。 但是，此错误可能仅在此移相器上的某些后续操作（如果有）时导致 IllegalStateException。 |
| int arriveAndAwaitAdvance() | 到达此移相器并等待其他移相器。 等效于awaitAdvance(arrive())。如果您需要等待中断或超时，您可以使用 awaitAdvance方法的其他形式之一通过类似的结构来安排它。 如果您需要在抵达时取消注册，请使用 awaitAdvance(arriveAndDeregister())。未注册方调用此方法属于使用错误。 但是，此错误可能仅在此移相器上的某些后续操作（如果有）时导致 IllegalStateException。 |
| int awaitAdvance(int phase) | 等待此移相器的相位从给定的相位值前进,等待给定phase数，返回下一个 arrival phase number 如果当前相位不等于给定的相位值或此移相器终止，则立即返回。 |
| int awaitAdvanceInterruptibly(int phase) throws InterruptedException | 等待此移相器的阶段从给定的阶段值前进，等待给定phase数，返回下一个 arrival phase number，响应中断 如果在等待期间被中断，则抛出 InterruptedException， 如果当前阶段不等于给定的阶段值或此移相器终止，则立即返回。 |
| int awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException | 等待此移相器的阶段从给定的阶段值或给定的超时时间过去， 如果在等待期间中断，则抛出 InterruptedException， 如果当前阶段不等于给定的阶段值或此移相器终止，则立即返回。 |
| void forceTermination() | 强制此移相器进入终止状态。 注册方的数量不受影响。 如果此移相器是分层移相器集的成员，则该集中的所有移相器都将终止。 如果此移相器已终止，则此方法无效。 此方法可用于在一个或多个任务遇到意外异常后协调恢复。 |
| int getPhase() | 返回当前阶段编号。 最大阶段数是 Integer.MAX_VALUE，之后它会从零重新开始。 终止时，阶段编号为负数，在这种情况下，可以通过 getPhase() + Integer.MIN_VALUE 获得终止之前的主要阶段。 |
| int getRegisteredParties() | 返回在此移相器上注册的参与方数量。 |
| int getArrivedParties() | 返回已到达此移相器当前阶段的已注册方的数量。 如果此移相器已终止，则返回的值是无意义的和任意的。 |
| int getUnarrivedParties() | 返回尚未到达此移相器当前阶段的已注册方的数量。 如果此移相器已终止，则返回的值是无意义的和任意的。 |
| Phaser getParent() | 返回此移相器的父级，如果没有则返回 null。 |
| Phaser getRoot() | 返回此移相器的根祖先，如果它没有父移相器，则与此移相器相同。 |
| boolean isTerminated() | 如果此移相器已终止，则返回 true。 |
| boolean onAdvance(int phase, int registeredParties) | 在即将到来的相位超前时执行操作并控制终止的可覆盖方法。 此方法在推进此移相器的一方到达时调用（当所有其他等待方都处于休眠状态时）。 如果此方法返回 true，则此移相器将在前进时设置为最终终止状态，随后对 isTerminated 的调用将返回 true。 调用此方法引发的任何（未经检查的）异常或错误都会传播到试图推进此移相器的一方，在这种情况下不会发生推进。 此方法的参数提供了当前转换占主导地位的移相器的状态。从 onAdvance 内部调用此移相器的到达、注册和等待方法的效果是未指定的，不应依赖。 如果此移相器是一组分层移相器的成员，则仅在每次前进时为其根移相器调用 onAdvance。 为了支持最常见的用例，当注册方的数量由于一方调用到达和取消注册而变为零时，此方法的默认实现返回 true。您可以通过覆盖此方法以始终返回 false 来禁用此行为，从而在未来注册时继续进行： Phaser phaser = new Phaser() protected boolean onAdvance(int phase, int parties) return false; |
方法说明
register()与bulkRegister(int parties)在验证参与线程数之后，核心都是调用int doRegister(int registrations)实现。
int doRegister(int registrations)
private int doRegister(int registrations)
// adjustment to state
// 通过位运算将新增的所需成员数增加到状态中
long adjust = ((long)registrations << PARTIES_SHIFT) | registrations;
// 找到当前父节点
final Phaser parent = this.parent;
// 阶段
int phase;
// 无限循环
for (;;)
// 如果没有父结点，那么状态不变，否则需要重新协调状态
long s = (parent == null) ? state : reconcileState();
// 截取state低32位
int counts = (int)s;
// 获取已经注册的全部线程数
int parties = counts >>> PARTIES_SHIFT;
// 未到达的线程数
int unarrived = counts & UNARRIVED_MASK;
// 验证注册数是否超过最大值
if (registrations > MAX_PARTIES - parties)
throw new IllegalStateException(badRegister(s));
// 获取当前年代数
phase = (int)(s >>> PHASE_SHIFT);
// 此时年代小于0 说明年代取消或损坏 不能再上面注册参与者线程，跳出无限循环
if (phase < 0)
break;
// 当前并非此年代第一次注册
if (counts != EMPTY) // not 1st registration
// 若父节点为null 或者 状态无变化
if (parent == null || reconcileState() == s)
if (unarrived == 0) // wait out advance
// 从根节点开始等待其他年代
root.internalAwaitAdvance(phase, null); // 等待其他任务到达
else if (UNSAFE.compareAndSwapLong(this, stateOffset,
s, s + adjust)) //更新注册的parties数
break;
// 逻辑到此，说明是第一次注册，若父节点直接更新状态
else if (parent == null) // 1st root registration
long next = ((long)phase << PHASE_SHIFT) | adjust;
if (UNSAFE.compareAndSwapLong(this, stateOffset, s, next))
break;
// 第一次注册且有父节点
else
synchronized (this) // 1st sub registration
// 通过检查同步状态值变化，判断锁定情况
if (state == s) // recheck under lock
// 父节点注册1个项？为了增加层级
phase = parent.doRegister(1);
if (phase < 0)
break;
// finish registration whenever parent registration
// succeeded, even when racing with termination,
// since these are part of the same "transaction".
// 由于在同一个事务里，即使phaser已终止，也会完成注册
while (!UNSAFE.compareAndSwapLong
(this, stateOffset, s,
((long)phase << PHASE_SHIFT) | adjust)) //更新s、phase
s = state;
phase = (int)(root.state >>> PHASE_SHIFT);
// assert (int)s == EMPTY;
break;
return phase;
reconcileState()
协调状态
private long reconcileState()
final Phaser root = this.root;
long s = state;
// 如果本实例就是根节点，不需要调整状态，否则需要进行state调整
if (root != this)
int phase, p;
// CAS to root phase with current parties, tripping unarrived
// CAS 与当前各方一起进入根阶段，跳过为到未达的
while ((phase = (int)(root.state >>> PHASE_SHIFT)) !=
(int)(s >>> PHASE_SHIFT) &&
!UNSAFE.compareAndSwapLong
(this, stateOffset, s,
s = (((long)phase << PHASE_SHIFT) |
((phase < 0) ? (s & COUNTS_MASK) :
(((p = (int)s >>> PARTIES_SHIFT) == 0) ? EMPTY :
((s & PARTIES_MASK) | p))))))
s = state;
return s;
internalAwaitAdvance(int phase, QNode node)
阻塞等待phase到下一代
private int internalAwaitAdvance(int phase, QNode node)
// assert root == this;
// 释放上一代
releaseWaiters(phase-1); // ensure old queue clean
boolean queued = false; // true when node is enqueued
int lastUnarrived = 0; // to increase spins upon change
int spins = SPINS_PER_ARRIVAL;
long s;
int p;
while ((p = (int)((s = state) >>> PHASE_SHIFT)) == phase)
if (node == null) // spinning in noninterruptible mode
int unarrived = (int)s & UNARRIVED_MASK;
if (unarrived != lastUnarrived &&
(lastUnarrived = unarrived) < NCPU)
spins += SPINS_PER_ARRIVAL;
boolean interrupted = Thread.interrupted();
if (interrupted || --spins < 0) // need node to record intr
node = new QNode(this, phase, false, false, 0L);
node.wasInterrupted = interrupted;
else if (node.isReleasable()) // done or aborted
break;
else if (!queued) // push onto queue
AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
QNode q = node.next = head.get();
if ((q == null || q.phase == phase) &&
(int)(state >>> PHASE_SHIFT) == phase) // avoid stale enq
queued = head.compareAndSet(q, node);
else
try
ForkJoinPool.managedBlock(node);
catch (InterruptedException ie)
node.wasInterrupted = true;
if (node != null)
if (node.thread != null)
node.thread = null; // avoid need for unpark()
if (node.wasInterrupted && !node.interruptible)
Thread.currentThread().interrupt();
if (p == phase && (p = (int)(state >>> PHASE_SHIFT)) == phase)
return abortWait(phase); // possibly clean up on abort
releaseWaiters(phase);
return p;
releaseWaiters(int phase)
private void releaseWaiters(int phase)
QNode q; JUC系列同步工具类之ThreadLocal