ForkJoinPool、CAS原子操作

2024-03-30 01:04
文章标签 操作 原子 cas forkjoinpool

本文主要是介绍ForkJoinPool、CAS原子操作，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

ForkJoinPool

ForkJoinPool是由JDK1.7后提供多线程并行执行任务的框架。可以理解为一种特殊的线程池。

1.任务分割：Fork（分岔），先把大的任务分割成足够小的子任务，如果子任务比较大的话还要对子任务进行继续分割。

2.合并结果：join，分割后的子任务被多个线程执行后，再合并结果，得到最终的完整输出。

类似于分治的思想，把大任务一点点拆分为一个个小任务。

如果要统计1~100之间的和，当然可以直接暴力for循环，不过也可以把它拆分为10个任务，计算1到10的和，11到20的和…

  • ForkJoinTask：主要提供fork和join两个方法用于任务拆分与合并；一般用子类 RecursiveAction（无返回值的任务）和RecursiveTask（需要返回值）来实现compute方法。

public abstract class ForkJoinTask<V> implements Future<V>, Serializable

可以看到，ForkJoinTask实现了Future这个接口，也就是说，我们也可以通过ForkJoinTask来获取线程的状态、结果等。

  • ForkJoinPool：调度ForkJoinTask的线程池；

  • ForkJoinWorkerThread：Thread的子类，存放于线程池中的工作线程（Worker）；

  • WorkQueue：任务队列，用于保存任务；
    ForkJoinPool forkJoinPool=new ForkJoinPool(8);//最多拆分为8个线程java.util.concurrent.ForkJoinTask<Integer> forkJoinTask = forkJoinPool.submit(new ForkJoinTask(1, 100));System.out.println(forkJoinTask.get());static class ForkJoinTask extends RecursiveTask<Integer>{int start;int end;public ForkJoinTask(int start, int end) {this.start = start;this.end = end;}@Overrideprotected Integer compute() {if((end-start)<=10){int count=0;for(int i=start;i<=end;i++){count+=i;}System.out.println("当前线程为："+Thread.currentThread().getName());return count;}else{int mid=start+end>>1;ForkJoinTask subTask1=new ForkJoinTask(start,mid);subTask1.fork();ForkJoinTask subTask2=new ForkJoinTask(mid+1,end);subTask2.fork();return subTask1.join()+subTask2.join();}}}

计算1~100的和，如果end-start小于等于10就直接暴力进行加法运算，如果大于10，就继续拆分。

ForkJoinPool的设计思想

  • 普通线程池内部有两个重要集合：工作线程集合（普通线程），和任务队列。
  • ForkJoinPool也类似，线程集合里放的是特殊线程ForkJoinWorkerThread，任务队列里放的是特殊任务ForkJoinTask
  • 不同之处在于，普通线程池只有一个队列。而ForkJoinPool的工作线程ForkJoinWorkerThread每个线程内都绑定一个双端队列。
  • 在fork的时候，也就是任务拆分，将拆分的task会被当前线程放到自己的队列中。
  • 如果有任务，那么线程优先从自己的队列里取任务执行，以LIFO先进后出方式从队尾获取任务，
  • 当自己队列中执行完后，工作线程会跑到其他队列以work−stealing窃取,窃取方式为FIFO先进先出，减少竞争。

  1. 任务拆分：线程首先将大任务拆分成更小的任务。
  2. 本地队列：拆分出的小任务通常会被放置在执行这个任务的线程的本地队列中。这个队列是一个双端队列（deque）。
  3. 任务窃取：其他闲置的线程可以从这个队列的另一端窃取任务来执行。这意味着，虽然拆分出的任务最初是放在原线程的队列中，但其他线程可以参与处理这些任务。
  4. 负载均衡：通过这种方式，ForkJoinPool试图在其所有线程之间实现负载均衡，从而提高效率。

举例：

假设你有一个大任务：计算从1加到10000的总和。这个任务可以通过拆分成更小的任务来并行处理。

  1. 任务拆分：线程A开始执行这个任务，它决定将任务拆分成两个更小的任务：第一个是计算1到5000的总和，第二个是计算5001到10000的总和。
  2. 放置在本地队列：线程A将这两个任务放入它的本地队列。此时，它开始执行其中一个任务（比如计算1到5000的总和）。
  3. 工作窃取：此时，另一个线程B处于空闲状态，它会查看线程A的队列。线程B发现队列中有待处理的任务（计算5001到10000的总和），于是它将这个任务从队列中窃取并开始执行。
  4. 并行处理：线程A和线程B现在都在执行一个较小的任务。一旦各自的任务完成，结果会被汇总。在这个例子中，两个任务的结果将被加在一起以得到最终的总和。

注意点

使用ForkJoin将相同的计算任务通过多线程执行。但是在使用中需要注意：

  • 注意任务切分的粒度，也就是fork的界限。并非越小越好
  • 判断要不要使用ForkJoin。任务量不是太大的话，串行可能优于并行。因为多线程会涉及到上下文的切换

CAS(比较交换)原子操作

在说CAS之前先说一下什么是原子操作

原子（atom）本意是“不能被进一步分割的最小粒子”，而原子操作（atomic operation）意为"不可被中断的一个或一系列操作" 。

CAS（Compare-and-Swap/Exchange），即比较并替换，是一种实现并发常用到的技术。CAS的整体架构如下：

  • 初始状态：计数器的值为0。
  • 线程A 读取计数器的值，得到0，打算将其增加到1。
  • 线程B 也读取计数器的值，得到0，同样打算将其增加到1。

此时，假设两个线程都尝试执行CAS操作来更新计数器的值。

理想的CAS操作：

  1. 线程A 的CAS操作先执行，它比较当前计数器的值（0）与预期值（0），发现匹配，因此成功将计数器的值更新为1。
  2. 接着，线程B 尝试执行它的CAS操作。这时，它比较当前计数器的值（现在为1）与其预期值（0），发现不匹配，因此不执行更新。

在这个情况下，计数器的最终值是1，这是正确的结果。每个线程都试图将计数器增加1，但只有一个线程成功了，因为CAS操作确保了计数器的每次更新都是基于最新的、有效的值。

如果CAS不按预期行为：

假设当线程B的预期值不匹配时，CAS操作仍然执行了更改，将计数器从1增加到2。

这将导致以下问题：

  • 数据不一致：这意味着两个线程的操作都基于同一个旧值（0），从而错误地假定计数器未被其他线程更改。
  • 结果错误：最终计数器的值变为2，而实际上只应该被增加1次。这是因为线程B没有正确地检测到线程A已经更新了计数器。
public class TestAtomic {public static void main(String[] args) {AtomicInteger atomicInteger=new AtomicInteger(1);atomicInteger.addAndGet(1);atomicInteger.incrementAndGet();}
}

以上两种方法都是给当前值+1，addAndGet(1)表示是在当前值的基础上+1，incrementAndGet表示自增。

源码剖析

    public final int incrementAndGet() {return unsafe.getAndAddInt(this, valueOffset, 1) + 1;}

    public final int getAndAddInt(Object o, long offset, int delta) {int v;do {v = getIntVolatile(o, offset);} while (!compareAndSwapInt(o, offset, v, v + delta));return v;}

    public final native boolean compareAndSwapInt(Object o, long offset,int expected,int x);

offset：当前变量的地址=当前类的地址+偏移量offset

compareAndSwapInt(o, offset, v, v + delta)：判断是否交换成功了，没有成功会去一直获取内存中value的值。

由此可以看出CAS是有一定弊端的，在面临高并发的场景下，可以持续死循环，导致CPU飙高。

CAS虽然很高效的解决了原子操作问题，但是CAS仍然存在三大问题。

  1. 自旋(循环)时间长开销很大，如果CAS失败，会一直进行尝试。如果CAS长时间一直不成功，可能会给CPU带来很大的开销，注意这里的自旋是在用户态/SDK 层面实现的。
  2. 只能保证一个共享变量的原子操作，对多个共享变量操作时，循环CAS就无法保证操作的原子性，这个时候就可以用锁来保证原子性。
  3. ABA问题，在使用CAS前要考虑清楚“ABA”问题是否会影响程序并发的正确性，如果需要解决ABA问题，改用传统的互斥同步可能会比CAS更高效。

ABA问题：

加入A拿到Value，并且进行了修改，A=1（这时候B拿到了A=1，接着A继续执行） —> A=2 —> A=1，这时候B一看，好家伙，Value还是1，该到我改了吧，这下直接A=5了，这就会导致原子性被破坏了。

可以通过引入AtomicStampedReference来解决ABA的问题

  • AtomicStampedReference：原子更新带有版本号的引用类型。
public class TestAtomic {public static void main(String[] args) {AtomicStampedReference<Integer> atomicStampedReference = new AtomicStampedReference<>(1, 1);int expectedReference = 1; // 当前期望的值int newReference = 2; // 新值int expectedStamp = 1; // 当前期望的版本号int newStamp = 2; // 新的版本号boolean wasUpdated = atomicStampedReference.compareAndSet(expectedReference, newReference, expectedStamp, newStamp);if (wasUpdated) {System.out.println("Update successful");} else {System.out.println("Update failed");}}
}

这篇关于ForkJoinPool、CAS原子操作的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


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

相关文章

动手学深度学习【数据操作+数据预处理】

动手学深度学习【数据操作+数据预处理】

import osos.makedirs(os.path.join('.', 'data'), exist_ok=True)data_file = os.path.join('.', 'data', 'house_tiny.csv')with open(data_file, 'w') as f:f.write('NumRooms,Alley,Price\n') # 列名f.write('NA
阅读更多...
线程的四种操作

线程的四种操作

所属专栏：Java学习        1. 线程的开启 start和run的区别： run：描述了线程要执行的任务，也可以称为线程的入口 start：调用系统函数，真正的在系统内核中创建线程（创建PCB，加入到链表中），此处的start会根据不同的系统，分别调用不同的api，创建好之后的线程，再单独去执行run（所以说，start的本质是调用系统api，系统的api
阅读更多...
Java IO 操作——个人理解

Java IO 操作——个人理解

之前一直Java的IO操作一知半解。今天看到一个便文章觉得很有道理（ 原文章），记录一下。 首先，理解Java的IO操作到底操作的什么内容，过程又是怎么样子。          数据来源的操作： 来源有文件，网络数据。使用File类和Sockets等。这里操作的是数据本身，1,0结构。    File file = new File("path");   字
阅读更多...
MySQL——表操作

MySQL——表操作

目录 一、创建表 二、查看表 2.1 查看表中某成员的数据 2.2 查看整个表中的表成员 2.3 查看创建表时的句柄 三、修改表 alter 3.1 重命名 rename 3.2 新增一列 add 3.3 更改列属性 modify 3.4 更改列名称 change 3.5 删除某列 上一篇博客介绍了库的操作，接下来来看一下表的相关操作。 一、创建表 create
阅读更多...
Lua 脚本在 Redis 中执行时的原子性以及与redis的事务的区别

Lua 脚本在 Redis 中执行时的原子性以及与redis的事务的区别

在 Redis 中，Lua 脚本具有原子性是因为 Redis 保证在执行脚本时，脚本中的所有操作都会被当作一个不可分割的整体。具体来说，Redis 使用单线程的执行模型来处理命令，因此当 Lua 脚本在 Redis 中执行时，不会有其他命令打断脚本的执行过程。脚本中的所有操作都将连续执行，直到脚本执行完成后，Redis 才会继续处理其他客户端的请求。 Lua 脚本在 Redis 中原子性的原因
阅读更多...
封装MySQL操作时Where条件语句的组织

封装MySQL操作时Where条件语句的组织

在对数据库进行封装的过程中，条件语句应该是相对难以处理的，毕竟条件语句太过于多样性。 条件语句大致分为以下几种： 1、单一条件，比如：where id = 1; 2、多个条件，相互间关系统一。比如：where id > 10 and age > 20 and score < 60; 3、多个条件，相互间关系不统一。比如：where (id > 10 OR age > 20) AND sco
阅读更多...
PHP7扩展开发之流操作

PHP7扩展开发之流操作

前言 啥是流操作？简单来讲就是对一些文件，网络的IO操作。PHP已经把这些IO操作，封装成流操作。这节，我们将使用PHP扩展实现一个目录遍历的功能。PHP示例代码如下： <?phpfunction list_dir($dir) {if (is_dir($dir) === false) {return;} $dh = opendir($dir);if ($dh == false) {ret
阅读更多...
浙大数据结构：树的定义与操作

浙大数据结构：树的定义与操作

四种遍历 #include<iostream>#include<queue>using namespace std;typedef struct treenode *BinTree;typedef BinTree position;typedef int ElementType;struct treenode{ElementType data;BinTree left;BinTre
阅读更多...
浙大数据结构：04-树7 二叉搜索树的操作集

浙大数据结构：04-树7 二叉搜索树的操作集

这道题答案都在PPT上，所以先学会再写的话并不难。 1、BinTree Insert( BinTree BST, ElementType X ) 递归实现，小就进左子树，大就进右子树。 为空就新建结点插入。 BinTree Insert( BinTree BST, ElementType X ){if(!BST){BST=(BinTree)malloc(sizeof(struct TNo
阅读更多...
hibernate修改数据库已有的对象【简化操作】

hibernate修改数据库已有的对象【简化操作】

陈科肇 直接上代码： /*** 更新新的数据并并未修改旧的数据* @param oldEntity 数据库存在的实体* @param newEntity 更改后的实体* @throws IllegalAccessException * @throws IllegalArgumentException */public void updateNew(T oldEntity,T newEntity
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2