forkjoin框架的简单使用

淡淡的烟草味﹌ 2023-09-27 13:04 140阅读 0赞

**目录**

1.说明

2.使用场景

3.实例

--------------------

### 1.说明 ###

是一个分而治之的任务框架，如一个任务需要多线程执行，分割成很多块计算的时候，可以采用这种方法。  
动态规范：和分而治之不同的是，每个小任务之间互相联系。  
工作密取：分而治之分割了每个任务之后，某个线程提前完成了任务，就会去其他线程偷取任务来完成，加快执行效率。同时，第一个分配的线程是从队列中的头部拿任务，当完成任务的线程去其他队列拿任务的时候是从尾部拿任务，所以这样就避免了竞争。  
在Java的Fork/Join框架中，使用两个类完成上述操作:  
　　1.ForkJoinTask:我们要使用Fork/Join框架，首先需要创建一个ForkJoin任务。该类提供了在任务中执行fork和join的机制。通常情况下我们不需要直接集成ForkJoinTask类，只需要继承它的子类，Fork/Join框架提供了两个子类：  
　　　　a.RecursiveAction：用于没有返回结果的任务  
　　　　b.RecursiveTask:用于有返回结果的任务  
　　2.ForkJoinPool:ForkJoinTask需要通过ForkJoinPool来执行.他其实也是一个线程池。它使用了一个无限队列来保存需要执行的任务，而线程的数量则是通过构造函数传入，如果没有向构造函数中传入希望的线程数量，那么当前计算机可用的CPU数量会被设置为线程数量作为默认值。  
注意：ForkJoinPool的invoke方法是同步阻塞的，excute方法是异步的。  
Fork/Join框架的实现原理:  
　　ForkJoinPool由ForkJoinTask数组和ForkJoinWorkerThread数组组成，ForkJoinTask数组负责将存放程序提交给ForkJoinPool，ForkJoinWorkerThread负责执行这些任务。  
原文链接：https://blog.csdn.net/baihaiyang99/article/details/118419560

### 2.使用场景 ###

Fork/Join框架适合能够进行拆分再合并的计算密集型（CPU密集型）任务。ForkJoin框架  
是一个并行框架，因此要求服务器拥有多CPU、多核，用以提高计算能力。  
如果是单核、单CPU，不建议使用该框架，会带来额外的性能开销，反而比单线程的执行效率低。当然不是因为并行的任务会进行频繁的线程切换，因为Fork/Join框架在进行线程池初始化的时候默认线程数量为Runtime.getRuntime().availableProcessors()，单CPU单核的情况下只会产生一个线程，并不会造成线程切换，而是会增加Fork/Join框架的一些队列、池化的开销。  
比如：数据迁移到数据库，解析excel等等可以拆分完成的任务都可以使用到forkjoin。

### 3.实例 ###

创建要执行的任务

package com.example.demo.forkjoin;
    
    import java.util.concurrent.RecursiveTask;
    
    /**
     * @author linaibo
     * @version 1.0
     * Create by 2022/12/3 15:21
     */
    
    public class ForkJoinCalculate extends RecursiveTask<Long> {
    
        private long start;
        private long end;
        private final long THRESHOLD = 10000;
    
        public ForkJoinCalculate(long start, long end) {
            this.start = start;
            this.end = end;
        }
    
        @Override
        protected Long compute() {
    
            long length = end - start;
            if (length <= THRESHOLD) {
                long sum = 0l;
                for (long i = start; i < end; i++) {
                    sum += i;
                }
                return sum;
            } else {
    
                long middle = (start + end) / 2;
                ForkJoinCalculate left = new ForkJoinCalculate(start, middle);
                left.fork();
                ForkJoinCalculate right = new ForkJoinCalculate(middle + 1, end);
                right.fork();
                return left.join() + right.join();
            }
        }
    }

执行此任务

package com.example.demo.forkjoin;
    
    import org.junit.jupiter.api.Test;
    
    import java.time.Duration;
    import java.time.Instant;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.ForkJoinTask;
    
    /**
     * @author linaibo
     * @version 1.0
     * Create by 2022/12/3 15:32
     */
    
    public class ForkJoinCalculateTest {
    
        @Test
        public void test() {
            //使用forkjoin
            Instant start = Instant.now();
            ForkJoinPool pool = new ForkJoinPool();
            ForkJoinTask<Long> task = new ForkJoinCalculate(0, 1000000000l);
            Long invoke = pool.invoke(task);
            System.out.println(invoke);
            Instant end = Instant.now();
            System.out.println("耗费时间：" + Duration.between(start, end).getNano());//311000000
    
        }
    
        @Test
        public void test2() {
            //普通for循环
            Instant start = Instant.now();
            long sum = 0l;
            for (long i = 0; i < 1000000000l; i++) {
                sum += i;
            }
            System.out.println(sum);
            Instant end = Instant.now();
            System.out.println("耗费时间：" + Duration.between(start, end).getNano());//332000000
    
        }
    
    }