死磕 java同步系列之Phaser源码解析

2024-02-15 01:58
文章标签 java 源码 系列 解析 同步 phaser

本文主要是介绍死磕 java同步系列之Phaser源码解析，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

问题

（1）Phaser是什么？

（2）Phaser具有哪些特性？

（3）Phaser相对于CyclicBarrier和CountDownLatch的优势？

简介

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

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

使用方法

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

public class PhaserTest {public static final int PARTIES = 3;public static final int PHASES = 4;public static void main(String[] args) {Phaser phaser = new Phaser(PARTIES) {@Overrideprotected boolean onAdvance(int phase, int registeredParties) {// 【本篇文章由公众号“彤哥读源码”原创，请支持原创，谢谢！】System.out.println("=======phase: " + phase + " finished=============");return super.onAdvance(phase, registeredParties);}};for (int i = 0; i < PARTIES; i++) {new Thread(()->{for (int j = 0; j < PHASES; j++) {System.out.println(String.format("%s: phase: %d", Thread.currentThread().getName(), j));phaser.arriveAndAwaitAdvance();}}, "Thread " + i).start();}}
}

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

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

可以看到，每个阶段都是三个线程都完成了才进入下一个阶段。这是怎么实现的呢，让我们一起来学习吧。

原理猜测

根据我们前面学习AQS的原理，大概猜测一下Phaser的实现原理。

首先，需要存储当前阶段phase、当前阶段的任务数（参与者）parties、未完成参与者的数量，这三个变量我们可以放在一个变量state中存储。

其次，需要一个队列存储先完成的参与者，当最后一个参与者完成任务时，需要唤醒队列中的参与者。

嗯，差不多就是这样子。

结合上面的案例带入：

初始时当前阶段为0，参与者数为3个，未完成参与者数为3；

第一个线程执行到phaser.arriveAndAwaitAdvance();时进入队列；

第二个线程执行到phaser.arriveAndAwaitAdvance();时进入队列；

第三个线程执行到phaser.arriveAndAwaitAdvance();时先执行这个阶段的总结onAdvance()，再唤醒前面两个线程继续执行下一个阶段的任务。

嗯，整体能说得通，至于是不是这样呢，让我们一起来看源码吧。

源码分析

主要内部类

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 waitQNode 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();}
}

先完成的参与者放入队列中的节点，这里我们只需要关注thread和next两个属性即可，很明显这是一个单链表，存储着入队的线程。

主要属性

// 状态变量，用于存储当前阶段phase、参与者数parties、未完成的参与者数unarrived_count
private volatile long state;
// 最多可以有多少个参与者，即每个阶段最多有多少个任务
private static final int  MAX_PARTIES     = 0xffff;
// 最多可以有多少阶段
private static final int  MAX_PHASE       = Integer.MAX_VALUE;
// 参与者数量的偏移量
private static final int  PARTIES_SHIFT   = 16;
// 当前阶段的偏移量
private static final int  PHASE_SHIFT     = 32;
// 未完成的参与者数的掩码，低16位
private static final int  UNARRIVED_MASK  = 0xffff;      // to mask ints
// 参与者数，中间16位
private static final long PARTIES_MASK    = 0xffff0000L; // to mask longs
// counts的掩码，counts等于参与者数和未完成的参与者数的'|'操作
private static final long COUNTS_MASK     = 0xffffffffL;
private static final long TERMINATION_BIT = 1L << 63;// 一次一个参与者完成
private static final int  ONE_ARRIVAL     = 1;
// 增加减少参与者时使用
private static final int  ONE_PARTY       = 1 << PARTIES_SHIFT;
// 减少参与者时使用
private static final int  ONE_DEREGISTER  = ONE_ARRIVAL|ONE_PARTY;
// 没有参与者时使用
private static final int  EMPTY           = 1;// 用于求未完成参与者数量
private static int unarrivedOf(long s) {int counts = (int)s;return (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
}
// 用于求参与者数量（中间16位），注意int的位置
private static int partiesOf(long s) {return (int)s >>> PARTIES_SHIFT;
}
// 用于求阶段数（高32位），注意int的位置
private static int phaseOf(long s) {return (int)(s >>> PHASE_SHIFT);
}
// 已完成参与者的数量
private static int arrivedOf(long s) {int counts = (int)s; // 低32位return (counts == EMPTY) ? 0 :(counts >>> PARTIES_SHIFT) - (counts & UNARRIVED_MASK);
}
// 用于存储已完成参与者所在的线程，根据当前阶段的奇偶性选择不同的队列
private final AtomicReference<QNode> evenQ;
private final AtomicReference<QNode> oddQ;

主要属性为state和evenQ及oddQ：

（1）state，状态变量，高32位存储当前阶段phase，中间16位存储参与者的数量，低16位存储未完成参与者的数量【本篇文章由公众号“彤哥读源码”原创，请支持原创，谢谢！】；

Phaser

（2）evenQ和oddQ，已完成的参与者存储的队列，当最后一个参与者完成任务后唤醒队列中的参与者继续执行下一个阶段的任务，或者结束任务。

构造方法

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) {if (parties >>> 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<QNode>();this.oddQ = new AtomicReference<QNode>();}// 状态变量state的存储分为三段this.state = (parties == 0) ? (long)EMPTY :((long)phase << PHASE_SHIFT) |((long)parties << PARTIES_SHIFT) |((long)parties);
}

构造函数中还有一个parent和root，这是用来构造多层级阶段的，不在本文的讨论范围之内，忽略之。

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

下面我们一起来看看几个主要方法的源码：

register()方法

注册一个参与者，如果调用该方法时，onAdvance()方法正在执行，则该方法等待其执行完毕。

public int register() {return doRegister(1);
}
private int doRegister(int registrations) {// state应该加的值，注意这里是相当于同时增加parties和unarrivedlong adjust = ((long)registrations << PARTIES_SHIFT) | registrations;final Phaser parent = this.parent;int phase;for (;;) {// state的值long s = (parent == null) ? state : reconcileState();// state的低32位，也就是parties和unarrived的值int counts = (int)s;// parties的值int parties = counts >>> PARTIES_SHIFT;// unarrived的值int unarrived = counts & UNARRIVED_MASK;// 检查是否溢出if (registrations > MAX_PARTIES - parties)throw new IllegalStateException(badRegister(s));// 当前阶段phasephase = (int)(s >>> PHASE_SHIFT);if (phase < 0)break;// 不是第一个参与者if (counts != EMPTY) {                  // not 1st registrationif (parent == null || reconcileState() == s) {// unarrived等于0说明当前阶段正在执行onAdvance()方法，等待其执行完毕if (unarrived == 0)             // wait out advanceroot.internalAwaitAdvance(phase, null);// 否则就修改state的值，增加adjust，如果成功就跳出循环else if (UNSAFE.compareAndSwapLong(this, stateOffset,s, s + adjust))break;}}// 是第一个参与者else if (parent == null) {              // 1st root registration// 计算state的值long next = ((long)phase << PHASE_SHIFT) | adjust;// 修改state的值，如果成功就跳出循环if (UNSAFE.compareAndSwapLong(this, stateOffset, s, next))break;}else {// 多层级阶段的处理方式synchronized (this) {               // 1st sub registrationif (state == s) {               // recheck under lockphase = 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".while (!UNSAFE.compareAndSwapLong(this, stateOffset, s,((long)phase << PHASE_SHIFT) | adjust)) {s = state;phase = (int)(root.state >>> PHASE_SHIFT);// assert (int)s == EMPTY;}break;}}}}return phase;
}
// 等待onAdvance()方法执行完毕
// 原理是先自旋一定次数，如果进入下一个阶段，这个方法直接就返回了，
// 如果自旋一定次数后还没有进入下一个阶段，则当前线程入队列，等待onAdvance()执行完毕唤醒
private int internalAwaitAdvance(int phase, QNode node) {// 保证队列为空releaseWaiters(phase-1);          // ensure old queue cleanboolean queued = false;           // true when node is enqueuedint lastUnarrived = 0;            // to increase spins upon change// 自旋的次数int spins = SPINS_PER_ARRIVAL;long s;int p;// 检查当前阶段是否变化，如果变化了说明进入下一个阶段了，这时候就没有必要自旋了while ((p = (int)((s = state) >>> PHASE_SHIFT)) == phase) {// 如果node为空，注册的时候传入的为空if (node == null) {           // spinning in noninterruptible mode// 未完成的参与者数量int unarrived = (int)s & UNARRIVED_MASK;// unarrived有变化，增加自旋次数if (unarrived != lastUnarrived &&(lastUnarrived = unarrived) < NCPU)spins += SPINS_PER_ARRIVAL;boolean interrupted = Thread.interrupted();// 自旋次数完了，则新建一个节点if (interrupted || --spins < 0) { // need node to record intrnode = new QNode(this, phase, false, false, 0L);node.wasInterrupted = interrupted;}}else if (node.isReleasable()) // done or abortedbreak;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 enqqueued = head.compareAndSet(q, node);}else {try {// 当前线程进入阻塞状态，跟调用LockSupport.park()一样，等待被唤醒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;
}

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

（1）增加一个参与者，需要同时增加parties和unarrived两个数值，也就是state的中16位和低16位；

（2）如果是第一个参与者，则尝试原子更新state的值，如果成功了就退出；

（3）如果不是第一个参与者，则检查是不是在执行onAdvance()，如果是等待onAdvance()执行完成，如果否则尝试原子更新state的值，直到成功退出；

（4）等待onAdvance()完成是采用先自旋后进入队列排队的方式等待，减少线程上下文切换；

arriveAndAwaitAdvance()方法

当前线程当前阶段执行完毕，等待其它线程完成当前阶段。

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

public int arriveAndAwaitAdvance() {// Specialization of doArrive+awaitAdvance eliminating some reads/pathsfinal Phaser root = this.root;for (;;) {// state的值long s = (root == this) ? state : reconcileState();// 当前阶段int phase = (int)(s >>> PHASE_SHIFT);if (phase < 0)return phase;// parties和unarrived的值int counts = (int)s;// unarrived的值（state的低16位）int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);if (unarrived <= 0)throw new IllegalStateException(badArrive(s));// 修改state的值if (UNSAFE.compareAndSwapLong(this, stateOffset, s,s -= ONE_ARRIVAL)) {// 如果不是最后一个到达的，则调用internalAwaitAdvance()方法自旋或进入队列等待if (unarrived > 1)// 这里是直接返回了，internalAwaitAdvance()方法的源码见register()方法解析return root.internalAwaitAdvance(phase, null);// 到这里说明是最后一个到达的参与者if (root != this)return parent.arriveAndAwaitAdvance();// n只保留了state中parties的部分，也就是中16位long n = s & PARTIES_MASK;  // base of next state// parties的值，即下一次需要到达的参与者数量int nextUnarrived = (int)n >>> PARTIES_SHIFT;// 执行onAdvance()方法，返回true表示下一阶段参与者数量为0了，也就是结束了if (onAdvance(phase, nextUnarrived))n |= TERMINATION_BIT;else if (nextUnarrived == 0)n |= EMPTY;else// n 加上unarrived的值n |= nextUnarrived;// 下一个阶段等待当前阶段加1int nextPhase = (phase + 1) & MAX_PHASE;// n 加上下一阶段的值n |= (long)nextPhase << PHASE_SHIFT;// 修改state的值为nif (!UNSAFE.compareAndSwapLong(this, stateOffset, s, n))return (int)(state >>> PHASE_SHIFT); // terminated// 唤醒其它参与者并进入下一个阶段releaseWaiters(phase);// 返回下一阶段的值return nextPhase;}}
}

arriveAndAwaitAdvance的大致逻辑为：

（1）修改state中unarrived部分的值减1；

（2）如果不是最后一个到达的，则调用internalAwaitAdvance()方法自旋或排队等待；

（3）如果是最后一个到达的，则调用onAdvance()方法，然后修改state的值为下一阶段对应的值，并唤醒其它等待的线程；

（4）返回下一阶段的值；

总结

（1）Phaser适用于多阶段多任务的场景，每个阶段的任务都可以控制得很细；

（2）Phaser内部使用state变量及队列实现整个逻辑【本篇文章由公众号“彤哥读源码”原创，请支持原创，谢谢！】；

（3）state的高32位存储当前阶段phase，中16位存储当前阶段参与者（任务）的数量parties，低16位存储未完成参与者的数量unarrived；

（4）队列会根据当前阶段的奇偶性选择不同的队列；

（5）当不是最后一个参与者到达时，会自旋或者进入队列排队来等待所有参与者完成任务；

（6）当最后一个参与者完成任务时，会唤醒队列中的线程并进入下一个阶段；

彩蛋

Phaser相对于CyclicBarrier和CountDownLatch的优势？

答：优势主要有两点：

（1）Phaser可以完成多阶段，而一个CyclicBarrier或者CountDownLatch一般只能控制一到两个阶段的任务；

（2）Phaser每个阶段的任务数量可以控制，而一个CyclicBarrier或者CountDownLatch任务数量一旦确定不可修改。

 

这篇关于死磕 java同步系列之Phaser源码解析的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/710171。

相关文章

SpringBoot基于沙箱环境实现支付宝支付教程

SpringBoot基于沙箱环境实现支付宝支付教程

《SpringBoot基于沙箱环境实现支付宝支付教程》本文介绍了如何使用支付宝沙箱环境进行开发测试,包括沙箱环境的介绍、准备步骤、在SpringBoot项目中结合支付宝沙箱进行支付接口的实现与测试... 目录一、支付宝沙箱环境介绍二、沙箱环境准备2.1 注册入驻支付宝开放平台2.2 配置沙箱环境2.3 沙箱
阅读更多...
Redis 内存淘汰策略深度解析(最新推荐)

Redis 内存淘汰策略深度解析(最新推荐)

《Redis内存淘汰策略深度解析(最新推荐)》本文详细探讨了Redis的内存淘汰策略、实现原理、适用场景及最佳实践,介绍了八种内存淘汰策略,包括noeviction、LRU、LFU、TTL、Rand... 目录一、 内存淘汰策略概述二、内存淘汰策略详解2.1 ​noeviction（不淘汰）​2.2 ​LR
阅读更多...
使用Java发送邮件到QQ邮箱的完整指南

使用Java发送邮件到QQ邮箱的完整指南

《使用Java发送邮件到QQ邮箱的完整指南》在现代软件开发中,邮件发送功能是一个常见的需求,无论是用户注册验证、密码重置,还是系统通知,邮件都是一种重要的通信方式,本文将详细介绍如何使用Java编写程... 目录引言1. 准备工作1.1 获取QQ邮箱的SMTP授权码1.2 添加JavaMail依赖2. 实现
阅读更多...
Java嵌套for循环优化方案分享

Java嵌套for循环优化方案分享

《Java嵌套for循环优化方案分享》介绍了Java中嵌套for循环的优化方法,包括减少循环次数、合并循环、使用更高效的数据结构、并行处理、预处理和缓存、算法优化、尽量减少对象创建以及本地变量优化,通... 目录Java 嵌套 for 循环优化方案1. 减少循环次数2. 合并循环3. 使用更高效的数据结构4
阅读更多...
IDEA与JDK、Maven安装配置完整步骤解析

IDEA与JDK、Maven安装配置完整步骤解析

《IDEA与JDK、Maven安装配置完整步骤解析》：本文主要介绍如何安装和配置IDE（IntelliJIDEA）,包括IDE的安装步骤、JDK的下载与配置、Maven的安装与配置,以及如何在I... 目录1. IDE安装步骤2.配置操作步骤3. JDK配置下载JDK配置JDK环境变量4. Maven配置下
阅读更多...
java两个List的交集,并集方式

java两个List的交集,并集方式

《java两个List的交集,并集方式》文章主要介绍了Java中两个List的交集和并集的处理方法,推荐使用Apache的CollectionUtils工具类,因为它简单且不会改变原有集合,同时,文章... 目录Java两个List的交集,并集方法一方法二方法三总结java两个List的交集,并集方法一
阅读更多...
Spring AI集成DeepSeek三步搞定Java智能应用的详细过程

Spring AI集成DeepSeek三步搞定Java智能应用的详细过程

《SpringAI集成DeepSeek三步搞定Java智能应用的详细过程》本文介绍了如何使用SpringAI集成DeepSeek,一个国内顶尖的多模态大模型,SpringAI提供了一套统一的接口,简... 目录DeepSeek 介绍Spring AI 是什么？Spring AI 的主要功能包括1、环境准备2
阅读更多...
Spring AI集成DeepSeek实现流式输出的操作方法

Spring AI集成DeepSeek实现流式输出的操作方法

《SpringAI集成DeepSeek实现流式输出的操作方法》本文介绍了如何在SpringBoot中使用Sse（Server-SentEvents）技术实现流式输出,后端使用SpringMVC中的S... 目录一、后端代码二、前端代码三、运行项目小天有话说题外话参考资料前面一篇文章我们实现了《Spring
阅读更多...
Spring AI与DeepSeek实战一之快速打造智能对话应用

Spring AI与DeepSeek实战一之快速打造智能对话应用

《SpringAI与DeepSeek实战一之快速打造智能对话应用》本文详细介绍了如何通过SpringAI框架集成DeepSeek大模型,实现普通对话和流式对话功能,步骤包括申请API-KEY、项目搭... 目录一、概述二、申请DeepSeek的API-KEY三、项目搭建3.1. 开发环境要求3.2. mav
阅读更多...
Springboot的自动配置是什么及注意事项

Springboot的自动配置是什么及注意事项

《Springboot的自动配置是什么及注意事项》SpringBoot的自动配置（Auto-configuration）是指框架根据项目的依赖和应用程序的环境自动配置Spring应用上下文中的Bean... 目录核心概念：自动配置的关键特点：自动配置工作原理：示例：需要注意的点1.默认配置可能不适合所有场景
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2