fork/Join 线程可以创建多个吗?

Posted 2023-04-28

图一按照官方标准50000个并发，阈值为5000，理论10线程，为什么只能有4线程：0，1，2，3图二是自定义创建多个线程，有多个线程为什么官方标准创建不了更多的线程呢？？？另外我的cpu是双核4线程的，难道和这个有关？？

fork划分的是子任务，不是线程，子任务会被放入双向队列中，这里的一个队列对应一个线程，每个线程分别执行队列中的任务，当有线程提前完成了所有的任务，他会主动的从未完成的任务的线程任务队列的末尾“窃取”任务，也就是“工作窃取”算法。ForkJoin的优点是线程间并行计算，以少数量的线程来完成大量的任务，减少线程竞争cpu的开销。ForkJoinPool的默认线程数为计算机上的处理器数量。 参考技术A 我不太理解你为什么说用fork()创建线程。
我觉得，你不应该把父子进程输出都统一
这样，就像楼上所说的全是hello world
这样不便于分析。

看看如下代码：
#include <stdio.h> #include <stdlib.h> #include <unistd.h> #include <errno.h> #define err_exit(msg) (perror(msg),(exit(1))) int main(void) pid_t pid; pid=fork(); if(pid==0) puts("I am child .."); else if(pid>0) puts("I am parent .."); else err_exit("fork()"); return 0; ``` 结果如下：
I am parent ..
I am child ..
这样似乎还是没有分析出他们具有父子关系。 所以再修改代码：
#include
#include
#include
#include

#define err_exit(msg) (perror(msg),(exit(1)))

int main(void)

pid_t pid;
pid=fork(); if(pid==0) puts("I am child .."); printf("My father pid==%d\n",getppid()); sleep(30); else if(pid>0) puts("I am parent .."); printf("My child pid==%d\n",pid); wait(); else err_exit("fork()"); return 0;

得到结果如下：
I am parent ..
My child pid==3926
I am child ..
My father pid==3925
在程序睡眠期间。我们调用系统工具
pstree -p |grep t27
查到如下结果：
|-gnome-terminal(3503)-+-bash(3511)---t27(3925)---t27(3926)
这样你应该能比较清晰理解父子进程的关系。 希望能帮到你！

Java线程：Fork/Join-Java并行计算框架

并行计算在处处都有大数据的今天已经不是一个新鲜的词汇了，现在已经有单机多核甚至多机集群并行计算，注意，这里说的是并行，而不是并发。严格的将，并行是指系统内有多个任务同时执行，而并发是指系统内有多个任务同时存在，不同的任务按时间分片的方式切换执行，由于切换的时间很短，给人的感觉好像是在同时执行。
Java在JDK7之后加入了并行计算的框架Fork/Join，可以解决我们系统中大数据计算的性能问题。Fork/Join采用的是分治法，Fork是将一个大任务拆分成若干个子任务，子任务分别去计算，而Join是获取到子任务的计算结果，然后合并，这个是递归的过程。子任务被分配到不同的核上执行时，效率最高。伪代码如下：

Result solve(Problem problem) 
    if (problem is small)
        directly solve problem
    else 
        split problem into independent parts
        fork new subtasks to solve each part
        join all subtasks
        compose result from subresults
    

Fork/Join框架的核心类是ForkJoinPool，它能够接收一个ForkJoinTask，并得到计算结果。ForkJoinTask有两个子类，RecursiveTask（有返回值）和RecursiveAction（无返回结果），我们自己定义任务时，只需选择这两个类继承即可。类图如下：

下面来看一个实例：计算一个超大数组所有元素的和。代码如下：

import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

/**
 * @author: shuang.gao  Date: 2015/7/14 Time: 8:16
 */
public class SumTask extends RecursiveTask<Integer> 

    private static final long serialVersionUID = -6196480027075657316L;

    private static final int THRESHOLD = 500000;

    private long[] array;

    private int low;

    private int high;

    public SumTask(long[] array, int low, int high) 
        this.array = array;
        this.low = low;
        this.high = high;
    

    @Override
    protected Integer compute() 
        int sum = 0;
        if (high - low <= THRESHOLD) 
            // 小于阈值则直接计算
            for (int i = low; i < high; i++) 
                sum += array[i];
            
         else 
            // 1. 一个大任务分割成两个子任务
            int mid = (low + high) >>> 1;
            SumTask left = new SumTask(array, low, mid);
            SumTask right = new SumTask(array, mid + 1, high);

            // 2. 分别计算
            left.fork();
            right.fork();

            // 3. 合并结果
            sum = left.join() + right.join();
        
        return sum;
    

    public static void main(String[] args) throws ExecutionException, InterruptedException 
        long[] array = genArray(1000000);

        System.out.println(Arrays.toString(array));

        // 1. 创建任务
        SumTask sumTask = new SumTask(array, 0, array.length - 1);

        long begin = System.currentTimeMillis();

        // 2. 创建线程池
        ForkJoinPool forkJoinPool = new ForkJoinPool();

        // 3. 提交任务到线程池
        forkJoinPool.submit(sumTask);

        // 4. 获取结果
        Integer result = sumTask.get();

        long end = System.currentTimeMillis();

        System.out.println(String.format("结果 %s 耗时 %sms", result, end - begin));
    

    private static long[] genArray(int size) 
        long[] array = new long[size];
        for (int i = 0; i < size; i++) 
            array[i] = new Random().nextLong();
        
        return array;
    

我们通过调整阈值（THRESHOLD），可以发现耗时是不一样的。实际应用中，如果需要分割的任务大小是固定的，可以经过测试，得到最佳阈值；如果大小不是固定的，就需要设计一个可伸缩的算法，来动态计算出阈值。如果子任务很多，效率并不一定会高。
未完待续。。。

---

参考资料

http://gee.cs.oswego.edu/dl/papers/fj.pdf
https://docs.oracle.com/javase/tutorial/essential/concurrency/forkjoin.html
https://www.ibm.com/developerworks/cn/java/j-lo-forkjoin/
http://www.ibm.com/developerworks/cn/java/j-jtp11137.html

本文来自：高爽|Coder，原文地址：http://blog.csdn.net/ghsau/article/details/46287769，转载请注明。