Phaser

简介

可重用的同步屏障，其功能类似于CyclicBarrier和CountDownLatch，但支持更灵活的用法。

这个工具类我们暂时就翻译为：移相器/相位器

使用入门

我们来看一个简单的例子。

假设我们有一个比赛，有多位玩家参加。

当所有玩家完成第一次比赛，我们认为上半场游戏结束；全部参加完第二次比赛，认为下半场游戏结束。

这个要如何实现呢？

自定义 Phaser 类

public class MyPhaser extends Phaser {

    @Override
    protected boolean onAdvance(int phase, int registeredParties) {
        switch (phase) {
            case 0 :
                System.out.println("上半场完成");
                return false;
            case 1:
                System.out.println("下半场完成");
                return false;
            default:
                return true;
        }
    }

}

自定义 Runnable 类

private static class GameRunnable implements Runnable {
    private final Phaser phaser;
    private GameRunnable(Phaser phaser) {
        this.phaser = phaser;
    }

    @Override
    public void run() {
        //参加上半场比赛
        System.out.println("玩家-"+Thread.currentThread().getName()+":参加上半场比赛");
        //执行这个方法的话会等所有的选手都完成了之后再继续下面的方法
        phaser.arriveAndAwaitAdvance();
        // 下半场
        //参加上半场比赛
        System.out.println("玩家-"+Thread.currentThread().getName()+":参加下半场比赛");
        //执行这个方法的话会等所有的选手都完成了之后再继续下面的方法
        phaser.arriveAndAwaitAdvance();
    }
}

测试验证

public static void main(String[] args) {
    int nums = 3;
    Phaser phaser = new MyPhaser();

    //注册一次表示 phaser 维护的线程个数
    phaser.register();
    for(int i = 0; i >> PARTIES_SHIFT != 0)
        throw new IllegalArgumentException("Illegal number of parties");
    int phase = 0;
    this.parent = parent;
    if (parent != null) {
        final Phaser root = parent.root;
        this.root = root;
        this.evenQ = root.evenQ;
        this.oddQ = root.oddQ;
        if (parties != 0)
            phase = parent.doRegister(1);
    }
    else {
        this.root = this;
        this.evenQ = new AtomicReference();
        this.oddQ = new AtomicReference();
    }
    this.state = (parties == 0) ? (long)EMPTY :
        ((long)phase  evenQ;
private final AtomicReference oddQ;

register 注册

我们主要看一下案例中用到的几个方法，首先看一下 register 方法。

public int register() {
    return doRegister(1);
}

实际上调用的是下面的方法：

/**
 * Implementation of register, bulkRegister
 *
 * @param registrations 要添加到双方和未到达字段的数量。 必须大于零。
 * @author 老马啸西风
 */
private int doRegister(int registrations) {
    // adjustment to state
    // 位移+或运算
    long adjust = ((long)registrations >> PARTIES_SHIFT;
        int unarrived = counts & UNARRIVED_MASK;
        if (registrations > MAX_PARTIES - parties)
            // 返回异常信息，见下方
            throw new IllegalStateException(badRegister(s));
        phase = (int)(s >>> PHASE_SHIFT);
        if (phase >> PHASE_SHIFT);
                        // assert (int)s == EMPTY;
                    }
                    break;
                }
            }
        }
    }
    return phase;
}

reconcileState 解析

/**
 * 如有必要，解决从根开始的滞后相位传播。
 * 协调通常在root已提前但子相位尚未执行时发生，在这种情况下，它们必须通过将未到达方设置为前进（或如果方为零，则重置为未注册的EMPTY状态）来完成自己的 * 前。
 *
 * @return reconciled state
 * @author 老马啸西风
 */
private long reconcileState() {
    // 获取 root 节点
    final Phaser root = this.root;
    long s = state;

    // 默认的 root 就是 this，不等于说明有真正的 root 节点。
    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 >> PARTIES_SHIFT) == 0) ? EMPTY :
                       ((s & PARTIES_MASK) | p))))))
            s = state;
    }
    return s;
}

badRegister

返回相关注册失败的信息。

/**
 * Returns message string for bounds exceptions on registration.
 */
private String badRegister(long s) {
    return "Attempt to register more than " +
        MAX_PARTIES + " parties for " + stateToString(s);
}

对应的状态信息为：

/**
 * Implementation of toString and string-based error messages
 */
private String stateToString(long s) {
    return super.toString() +
        "[phase = " + phaseOf(s) +
        " parties = " + partiesOf(s) +
        " arrived = " + arrivedOf(s) + "]";
}

internalAwaitAdvance 内部的等待方法

这个方法只能被 root 节点调用，用于阻塞线程，等待阶段完成使用。

/**
 * 除非中止，否则可能会阻塞并等待阶段前进。
 * 仅在根相位器上调用。
 *
 * @param phase current phase
 * @param node if non-null, the wait node to track interrupt and timeout;
 * if null, denotes noninterruptible wait
 * @return current phase
 * @author 老马啸西风
 */
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)  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 释放等待者

/**
* 从队列中删除线程并发出信号通知阶段。
* @author 老马啸西风
*/
private void releaseWaiters(int phase) {
    // 队列中的第一个元素
    QNode q;   
    // 对应的线程信息
    Thread t;  
    // 根据奇偶，选择不同的队列。
    AtomicReference head = (phase & 1) == 0 ? evenQ : oddQ;
    while ((q = head.get()) != null &&
           q.phase != (int)(root.state >>> PHASE_SHIFT)) {
        // 设置 q 为 q.next，并且 q.thread 持有线程       
        if (head.compareAndSet(q, q.next) &&
            (t = q.thread) != null) {
            // 清空对应的线程    
            q.thread = null;
            // 唤醒 t 对应的线程
            LockSupport.unpark(t);
        }
    }
}

• abortWait 中断等待

/**
 * releaseWaiters的一种变体，它另外尝试删除由于超时或中断而不再等待提前的任何节点。
 * 当前，仅当节点位于队列头时才将其删除，这足以减少大多数使用情况下的内存占用。
 *
 * @return current phase on exit
 * @author 老马啸西风
 */
private int abortWait(int phase) {
    // 选择奇偶队列
    AtomicReference head = (phase & 1) == 0 ? evenQ : oddQ;
    for (;;) {
        Thread t;
        // 头节点
        QNode q = head.get();
        int p = (int)(root.state >>> PHASE_SHIFT);
        if (q == null || ((t = q.thread) != null && q.phase == p))
            return p;

        // CAS 设置节点 q 为 q.next（删除头节点）    
        if (head.compareAndSet(q, q.next) && t != null) {
            q.thread = null;
            LockSupport.unpark(t);
        }
    }
}

arriveAndAwaitAdvance 到达并且等待

实现

/**
 * 到达此移相器并等待其他人。
 * 等效于awaitAdvance。
 * 如果您需要等待中断或超时，则可以使用 awaitAdvance 方法的其他形式之一以类似的方式进行安排。
 * 如果相反，您需要在到达时注销，请使用 awaitAdvance（arriveAndDeregister（））。
 *
 * 未注册方调用此方法是错误的用法。
 * 但是，仅在此相位器上进行一些后续操作时，此错误才可能导致IllegalStateException。
 *
 * @return the arrival phase number, or the (negative)
 * {@linkplain #getPhase() current phase} if terminated
 * @throws IllegalStateException if not terminated and the number
 * of unarrived parties would become negative
 * @author 老马啸西风
 */
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  1)
                return root.internalAwaitAdvance(phase, null);
            if (root != this)
                // 这个方法上面有解析，不过此处调用的时 parent
                return parent.arriveAndAwaitAdvance();

            long n = s & PARTIES_MASK;  // base of next state
            int nextUnarrived = (int)n >>> PARTIES_SHIFT;

            // 注意：这里调用了对应的 onAdvance 方法，就是我们前面自定义实现的方法。
            if (onAdvance(phase, nextUnarrived))
                n |= TERMINATION_BIT;
            else if (nextUnarrived == 0)
                n |= EMPTY;
            else
                n |= nextUnarrived;

            int nextPhase = (phase + 1) & MAX_PHASE;
            n |= (long)nextPhase >> PHASE_SHIFT); // terminated
            releaseWaiters(phase);
            return nextPhase;
        }
    }
}

onAdvance 重载时的核心方法

一般我们都会对这个方法进行重载。

/**
 * 一种在即将到来的相位超前执行操作并控制终止的可重写方法。
 * 在推进此移相器的一方到达时（当所有其他等待方都处于休眠状态时）调用此方法。
 * 如果此方法返回{@code true}，则此移相器将提前设置为最终终止状态，随后对{@link #isTerminated}的调用将返回true。
 *  调用此方法引发的任何（未经检查的）异常或错误都会传播到尝试推进此相位器的一方，在这种情况下，不会发生提前。
 *
 * 此方法的参数提供了当前过渡中普遍使用的移相器状态。
 * 从{@code onAdvance}内在此相位器上调用到达，注册和等待方法的效果是不确定的，因此不应依赖。
 *
 * 如果此相位器是分层相位器集合的成员，则每次前进时仅为其根相位器调用{@code onAdvance}。
 *
 * 为了支持最常见的用例，当由于一方调用{@code到达AndDeregister}而导致的注册方数量变为零时，此方法的默认实现返回{@code true}。
 * 您可以通过重写此方法以始终返回{@code false}来禁用此行为，从而使以后的注册继续进行：
 *
 *  {@code
 * Phaser phaser = new Phaser() {
 *   protected boolean onAdvance(int phase, int parties) { return false; }
 * }}
 *
 * @param phase 进入此方法之前，当前相位编号
 * @param registeredParties 当前注册方的数量
 * @return {@code true} 如果此移相器应终止
 * @author 老马啸西风
 */
protected boolean onAdvance(int phase, int registeredParties) {
    return registeredParties == 0;
}

arriveAndDeregister 到达并且取消注册

/**
 * Arrives at this phaser and deregisters from it without waiting
 * for others to arrive. Deregistration reduces the number of
 * parties required to advance in future phases.  If this phaser
 * has a parent, and deregistration causes this phaser to have
 * zero parties, this phaser is also deregistered from its parent.
 *
 * It is a usage error for an unregistered party to invoke this
 * method.  However, this error may result in an {@code
 * IllegalStateException} only upon some subsequent operation on
 * this phaser, if ever.
 *
 * @return 到达阶段数量，如果终止则为负值
 * @throws IllegalStateException 如果未终止，则已注册或未注册方的数量将变为负数
 * @author 老马啸西风
 */
public int arriveAndDeregister() {
    return doArrive(ONE_DEREGISTER);
}

doArrive 方法

adjust 有两个参数值：

ONE_ARRIVAL for arrive, // 值为 1

ONE_DEREGISTER for arriveAndDeregister // 值为 ONE_ARRIVAL|ONE_PARTY;

/**
 * 方法的主要实现到达+到达并取消注册。
 * 对于仅减少未到达字段的常见情况，进行手动调整以加快并最小化竞赛窗口。
 *
 * @param adjust value to subtract from state;
 * @author 老马啸西风               
 */
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 >> PARTIES_SHIFT;
                if (root == this) {
                    // 这个就是我们重载的方法
                    if (onAdvance(phase, nextUnarrived))
                        n |= TERMINATION_BIT;
                    else if (nextUnarrived == 0)
                        n |= EMPTY;
                    else
                        n |= nextUnarrived;

                    // 计算下一个 phase 
                    int nextPhase = (phase + 1) & MAX_PHASE;
                    n |= (long)nextPhase << PHASE_SHIFT;
                    UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
                    releaseWaiters(phase);
                }

                // 广播注销
                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;
        }
    }
}

小结

希望本文对你有帮助，如果有其他想法的话，也可以评论区和大家分享哦。

各位极客的点赞收藏转发，是老马持续写作的最大动力！