【浅谈JUC工具Phaser】花门楼前见秋草,岂能贫贱相看老。这篇文章主要讲述浅谈JUC工具Phaser相关的知识,希望能为你提供帮助。
@TOC
JUC工具类: Phaser详解Phaser是JDK 7新增的一个同步辅助类,它可以实现CyclicBarrier和CountDownLatch类似的功能,而且它支持对任务的动态调整,并支持分层结构来达到更高的吞吐量。
Phaser运行机制
文章图片
- Registration(注册)
- Synchronization(同步机制)
方法arriveAndAwaitAdvance的效果类似CyclicBarrier.await。phaser的每一代都有一个相关的phase number,初始值为0,当所有注册的任务都到达phaser时phase+1,到达最大值(Integer.MAX_VALUE)之后清零。使用phase number可以独立控制 到达phaser 和 等待其他线程 的动作,通过下面两种类型的方法:
- Termination(终止机制) :
- Tiering(分层结构) :
- Monitoring(状态监控) :
Phaser源码详解 核心参数
private volatile long state;
/**
* The parent of this phaser, or null if none
*/
private final Phaser parent;
/**
* The root of phaser tree. Equals this if not in a tree.
*/
private final Phaser root;
//等待线程的栈顶元素,根据phase取模定义为一个奇数header和一个偶数header
private final AtomicReference<
QNode>
evenQ;
private final AtomicReference<
QNode>
oddQ;
state状态说明:
Phaser使用一个long型state值来标识内部状态:
- 低0-15位表示未到达parties数;
- 中16-31位表示等待的parties数;
- 中32-62位表示phase当前代;
- 高63位表示当前phaser的终止状态。
函数列表
//构造方法
public Phaser()
this(null, 0);
public Phaser(int parties)
this(null, parties);
public Phaser(Phaser parent)
this(parent, 0);
public Phaser(Phaser parent, int parties)
//注册一个新的party
public int register()
//批量注册
public int bulkRegister(int parties)
//使当前线程到达phaser,不等待其他任务到达。返回arrival phase number
public int arrive()
//使当前线程到达phaser并撤销注册,返回arrival phase number
public int arriveAndDeregister()
/*
* 使当前线程到达phaser并等待其他任务到达,等价于awaitAdvance(arrive())。
* 如果需要等待中断或超时,可以使用awaitAdvance方法完成一个类似的构造。
* 如果需要在到达后取消注册,可以使用awaitAdvance(arriveAndDeregister())。
*/
public int arriveAndAwaitAdvance()
//等待给定phase数,返回下一个 arrival phase number
public int awaitAdvance(int phase)
//阻塞等待,直到phase前进到下一代,返回下一代的phase number
public int awaitAdvance(int phase)
//响应中断版awaitAdvance
public int awaitAdvanceInterruptibly(int phase) throws InterruptedException
public int awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit)
throws InterruptedException, TimeoutException
//使当前phaser进入终止状态,已注册的parties不受影响,如果是分层结构,则终止所有phaser
public void forceTermination()
方法 - register()
//注册一个新的party
public int register()
return doRegister(1);
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();
int counts = (int)s;
int parties = counts >
>
>
PARTIES_SHIFT;
//获取已注册parties数
int unarrived = counts &
UNARRIVED_MASK;
//未到达数
if (registrations >
MAX_PARTIES - parties)
throw new IllegalStateException(badRegister(s));
phase = (int)(s >
>
>
PHASE_SHIFT);
//获取当前代
if (phase <
0)
break;
if (counts != EMPTY)// not 1st registration
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))//更新phase
break;
else
//分层结构,子phaser首次注册用父节点管理
synchronized (this)// 1st sub registration
if (state == s)// recheck under lock
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)) //更新phase
s = state;
phase = (int)(root.state >
>
>
PHASE_SHIFT);
// assert (int)s == EMPTY;
break;
return phase;
说明:register方法为phaser添加一个新的party,如果onAdvance正在运行,那么这个方法会等待它运行结束再返回结果。如果当前phaser有父节点,并且当前phaser上没有已注册的party,那么就会交给父节点注册。
register和bulkRegister都由doRegister实现,大概流程如下:
- 如果当前操作不是首次注册,那么直接在当前phaser上更新注册parties数
- 如果是首次注册,并且当前phaser没有父节点,说明是root节点注册,直接更新phase
- 如果当前操作是首次注册,并且当前phaser由父节点,则注册操作交由父节点,并更新当前phaser的phase
- 上面说过,子Phaser的phase在没有被真正使用之前,允许滞后于它的root节点。非首次注册时,如果Phaser有父节点,则调用reconcileState()方法解决root节点的phase延迟传递问题, 源码如下:
private long reconcileState()
final Phaser root = this.root;
long s = state;
if (root != this)
int phase, p;
// CAS to root phase with current parties, tripping unarrived
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;
当root节点的phase已经advance到下一代,但是子节点phaser还没有,这种情况下它们必须通过更新未到达parties数 完成它们自己的advance操作(如果parties为0,重置为EMPTY状态)。
回到register方法的第一步,如果当前未到达数为0,说明上一代phase正在进行到达操作,此时调用internalAwaitAdvance()方法等待其他任务完成到达操作,源码如下:
//阻塞等待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记录中断状态
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);
//阻塞给定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 abortreleaseWaiters(phase);
return p;
简单介绍下第二个参数node,如果不为空,则说明等待线程需要追踪中断状态或超时状态。以doRegister中的调用为例,不考虑线程争用,internalAwaitAdvance大概流程如下:
- 首先调用releaseWaiters唤醒上一代所有等待线程,确保旧队列中没有遗留的等待线程。
- 循环SPINS_PER_ARRIVAL指定的次数或者当前线程被中断,创建node记录等待线程及相关信息。
- 继续循环调用ForkJoinPool.managedBlock运行被阻塞的任务
- 继续循环,阻塞任务运行成功被释放,跳出循环
- 最后唤醒当前phase的线程
//使当前线程到达phaser,不等待其他任务到达。返回arrival phase number
public int arrive()
return doArrive(ONE_ARRIVAL);
private int doArrive(int adjust)
final Phaser root = this.root;
for (;
;
)
long s = (root == this) ? state : reconcileState();
int phase = (int)(s >
>
>
PHASE_SHIFT);
if (phase <
0)
return phase;
int counts = (int)s;
//获取未到达数
int unarrived = (counts == EMPTY) ? 0 : (counts &
UNARRIVED_MASK);
if (unarrived <
= 0)
throw new IllegalStateException(badArrive(s));
if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adjust)) //更新state
if (unarrived == 1) //当前为最后一个未到达的任务
long n = s &
PARTIES_MASK;
// base of next state
int nextUnarrived = (int)n >
>
>
PARTIES_SHIFT;
if (root == this)
if (onAdvance(phase, nextUnarrived))//检查是否需要终止phaser
n |= TERMINATION_BIT;
else if (nextUnarrived == 0)
n |= EMPTY;
else
n |= nextUnarrived;
int nextPhase = (phase + 1) &
MAX_PHASE;
n |= (long)nextPhase <
<
PHASE_SHIFT;
UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
releaseWaiters(phase);
//释放等待phase的线程//分层结构,使用父节点管理arrive
else if (nextUnarrived == 0)//propagate deregistration
phase = parent.doArrive(ONE_DEREGISTER);
UNSAFE.compareAndSwapLong(this, stateOffset,
s, s | EMPTY);
else
phase = parent.doArrive(ONE_ARRIVAL);
return phase;
说明:arrive方法手动调整到达数,使当前线程到达phaser。arrive和arriveAndDeregister都调用了doArrive实现,大概流程如下:
- 首先更新state(state - adjust);
- 如果当前不是最后一个未到达的任务,直接返回phase
- 如果当前是最后一个未到达的任务:
- 如果当前是root节点,判断是否需要终止phaser,CAS更新phase,最后释放等待的线程;
- 如果是分层结构,并且已经没有下一代未到达的parties,则交由父节点处理doArrive逻辑,然后更新state为EMPTY。
public int arriveAndAwaitAdvance()
// Specialization of doArrive+awaitAdvance eliminating some reads/paths
final Phaser root = this.root;
for (;
;
)
long s = (root == this) ? state : reconcileState();
int phase = (int)(s >
>
>
PHASE_SHIFT);
if (phase <
0)
return phase;
int counts = (int)s;
int unarrived = (counts == EMPTY) ? 0 : (counts &
UNARRIVED_MASK);
//获取未到达数
if (unarrived <
= 0)
throw new IllegalStateException(badArrive(s));
if (UNSAFE.compareAndSwapLong(this, stateOffset, s,
s -= ONE_ARRIVAL)) //更新state
if (unarrived >
1)
return root.internalAwaitAdvance(phase, null);
//阻塞等待其他任务
if (root != this)
return parent.arriveAndAwaitAdvance();
//子Phaser交给父节点处理
long n = s &
PARTIES_MASK;
// base of next state
int nextUnarrived = (int)n >
>
>
PARTIES_SHIFT;
if (onAdvance(phase, nextUnarrived))//全部到达,检查是否可销毁
n |= TERMINATION_BIT;
else if (nextUnarrived == 0)
n |= EMPTY;
else
n |= nextUnarrived;
int nextPhase = (phase + 1) &
MAX_PHASE;
//计算下一代phase
n |= (long)nextPhase <
<
PHASE_SHIFT;
if (!UNSAFE.compareAndSwapLong(this, stateOffset, s, n))//更新state
return (int)(state >
>
>
PHASE_SHIFT);
// terminated
releaseWaiters(phase);
//释放等待phase的线程
return nextPhase;
说明: 使当前线程到达phaser并等待其他任务到达,等价于awaitAdvance(arrive())。如果需要等待中断或超时,可以使用awaitAdvance方法完成一个类似的构造。如果需要在到达后取消注册,可以使用awaitAdvance(arriveAndDeregister())。效果类似于CyclicBarrier.await。大概流程如下:
- 更新state(state - 1);
- 如果未到达数大于1,调用internalAwaitAdvance阻塞等待其他任务到达,返回当前phase
- 如果为分层结构,则交由父节点处理arriveAndAwaitAdvance逻辑
- 如果未到达数< =1,判断phaser终止状态,CAS更新phase到下一代,最后释放等待当前phase的线程,并返回下一代phase。
public int awaitAdvance(int phase)
final Phaser root = this.root;
long s = (root == this) ? state : reconcileState();
int p = (int)(s >
>
>
PHASE_SHIFT);
if (phase <
0)
return phase;
if (p == phase)
return root.internalAwaitAdvance(phase, null);
return p;
//响应中断版awaitAdvance
public int awaitAdvanceInterruptibly(int phase)
throws InterruptedException
final Phaser root = this.root;
long s = (root == this) ? state : reconcileState();
int p = (int)(s >
>
>
PHASE_SHIFT);
if (phase <
0)
return phase;
if (p == phase)
QNode node = new QNode(this, phase, true, false, 0L);
p = root.internalAwaitAdvance(phase, node);
if (node.wasInterrupted)
throw new InterruptedException();
return p;
说明:awaitAdvance用于阻塞等待线程到达,直到phase前进到下一代,返回下一代的phase number。方法很简单,不多赘述。awaitAdvanceInterruptibly方法是响应中断版的awaitAdvance,不同之处在于,调用阻塞时会记录线程的中断状态。
如在文中有所收获,请点赞
推荐阅读
- Linux系列(环境变量配置文件目录)
- Linux系列(帮助命令manhelpinfo的区别)
- 某信息公司面经java
- Linux系列(花括号{}的使用)
- Lenet 神经网络
- Linux系列(多个并列表达式(是否为空))
- Linux 系列(awk 数组的用法)
- Android中的设计模式之观察者模式
- Android中在Bitmap上画网格线