【Java多线程】浅谈Java并行处理任务发展历程

我会带着你远行 2022-11-01 00:51 151阅读 0赞

### 文章目录 ###

*   *  一.前言
     *  二.无处不在的并行
     *  三.如何并行
     *   *  3.1.单线程处理
         *  3.2.Thread方式
         *  3.3.线程池方式
         *  3.4.fork/join框架
         *  3.5.并行流方式

## 一.前言 ##

**并行**，即: `多个线程一起运行，来提高系统的整体处理速度` 。

*  为什么使用多个线程就能提高处理速度，因为现在计算机普遍都是`多核处理器`，我们需要充`分利用cpu资源`；如果站的更高一点来看，我们`每台机器都可以是一个处理节点`，`多台机器并行处理`。并行的处理方式可以说无处不在。

本文主要来谈谈Java实现并行处理的方式

## 二.无处不在的并行 ##

Java的垃圾回收器，我们可以看到每一代版本的更新，伴随着GC更短的延迟，从`serial`到`cms`再到现在的`G1`，一直在摘掉Java慢的帽子。消息队列从早期的`ActiveMQ`到现在的`kafka`和`RocketMQ`，引入的`分区`的概念，提高了消息的并行性。数据库单表数据到一定量级之后，访问速度会很慢，我们会对表进行`分表`处理，引入`数据库中间件`；Redis你可能觉得本身处理是单线程的，但是Redis的集群方案中引入了`slot(槽)`的概念；更普遍的就是我们很多的业务系统，通常会部署多台，通过`负载均衡中间件`来进行分发；好了还有其他的一些例子，此处不在一一例举。

[Java垃圾收集器——Serial，Parallel，CMS，G1收集器概述][Java_Serial_Parallel_CMS_G1]  
[JVM垃圾收集器－对比Serial、Parallel、CMS和G1][JVM_Serial_Parallel_CMS_G1]

## 三.如何并行 ##

我觉得并行的**核心**在于`"拆分"`，`把大任务变成小任务，然后利用多核CPU也好，还是多节点也好，同时并行的处理`，Java历代版本的更新，都在为我们开发者提供更方便的并行处理，从开始的`Thread`，到`线程池`，再到`fork/join框架`，最后到`流`处理

下面使用简单的求和例子来看看各种方式是如何并行处理的；

### 3.1.单线程处理 ###

首先看一下最简单的单线程处理方式，直接使用主线程进行求和操作；

public class SingleThread { 
        public static void main(String[] args) { 
        	//生成指定范围大小的的数组
            long[] numbers = LongStream.rangeClosed(1, 10_000_000).toArray();
            long sum = 0;
            for (int i = 0; i < numbers.length; i++) { 
                sum += numbers[i];
            }
            System.out.println("sum = " + sum);
        }
    }

求和本身是一个**计算密集型任务**，但是现在已经是`多核时代`，只用`单线程`，相当于只使用了其中`一个cpu`，`其他cpu被闲置，导致资源的浪费`。

### 3.2.Thread方式 ###

我们把`任务拆分成多个小任务，然后每个小任务分别启动一个线程，分段处理任务`。如下所示：

public class ThreadTest { 
        //分段阈值,即每个线程处理次数
        public static final int threshold = 10_000;
        //要累加的数字集合
        public static long[] numbers;
        //累加结果
        private static long allSum;
    
        public static void main(String[] args) throws Exception { 
            //生成要累加的数字集合
            numbers = LongStream.rangeClosed(1, 10_000_000).toArray();
    
            //线程数 =计算总次数 / 每个线程处理次数
            int taskSize = (int) (numbers.length / threshold);
    
            //循环生成线程
            for (int i = 1; i <= taskSize; i++) { 
                final int key = i;
                new Thread(new Runnable() { 
                    public void run() { 
                        //一个线程处理数组的一段数据 start= (i - * threshold) ,end = key * threshold,类似于分页计算公式
                        sumAll(segmentSum((key - 1) * threshold, key * threshold));
                    }
                }).start();
            }
    
            Thread.sleep(100);
            System.out.println("allSum = " + getAllSum());
        }
    
        //累加每个线程计算的总和
        private static synchronized long sumAll(long threadSum) { 
            return allSum += threadSum;
        }
    
        //获取总和
        public static synchronized long getAllSum() { 
            return allSum;
        }
    
        /** * 分段累加 * @param start 开始下标 * @param end 结束下标 * @return */
        private static long segmentSum(int start, int end) { 
            long sum = 0;
            for (int i = start; i < end; i++) { 
                sum += numbers[i];
            }
            return sum;
        }
    }

上面通过*将一个大的任务，分段切分成一个个小任务。然后通过分段阈值，计算出要生成的线程数以及每一段任务处理的个数*。这种处理就是`创建的线程数过多，而CPU数有限`，更重要的是`求和是一个计算密集型任务`，`启动过多的线程只会带来更多的线程上下文切换`。同时`线程处理完一个任务就终止了，也是对资源的浪费`。另外可以看到主线程不知道何时子任务已经处理完了，需要做额外的处理。所以Java后续引入了**线程池**。

### 3.3.线程池方式 ###

**Java1.5**时引入了`并发包java.concurrent`，其中包括了线程池`ThreadPoolExecutor`，相关代码如下：

public class ExecutorServiceTest { 
        //分段阈值,即每个线程处理次数
        public static final int threshold = 10_000;
        //要累加的数字集合(即)
        public static long[] numbers;
    
        public static void main(String[] args) throws Exception { 
            //生成要累加的数字集合
            numbers = LongStream.rangeClosed(1, 10_000_000).toArray();
    
            //创建固定长度的线程池,核心线程数大于与非核心线程大小相等=cpu核心数+1
            ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() + 1);
    
            //CompletionService实际上可以看做是Executor和BlockingQueue的结合体。CompletionService在接收到要执行的任务时，通过类似BlockingQueue的put和take获得任务执行的结果。CompletionService的一个实现是ExecutorCompletionService，
            CompletionService<Long> completionService = new ExecutorCompletionService<Long>(executor);
    
            //线程数 =计算总次数 / 每个线程处理次数
            int taskSize = numbers.length / threshold;
    
            //循环生成线程
            for (int i = 1; i <= taskSize; i++) { 
                final int key = i;
                completionService.submit(new Callable<Long>() { 
                    @Override
                    public Long call() throws Exception { 
                        //一个线程处理数组的一段数据 start= (i - * threshold) ,end = key * threshold,类似于分页计算公式
                        return segmentSum((key - 1) * threshold, key * threshold);
                    }
                });
            }
    
            long sumValue = 0;
            for (int i = 0; i < taskSize; i++) { 
                //检索并移除表示下一个已完成任务的 Future，如果目前不存在这样的任务，则等待。
                sumValue += completionService.take().get();
            }
    
            // 所有任务已经完成,关闭线程池
            System.out.println("sumValue = " + sumValue);
            executor.shutdown();
        }
    
        /** * 分段累加 * @param start 开始下标 * @param end 结束下标 * @return */
        private static long segmentSum(int start, int end) { 
            long sum = 0;
            for (int i = start; i < end; i++) { 
                sum += numbers[i];
            }
            return sum;
        }
    }

上面已经分析了`计算密集型`业务中`并不是线程越多越好`，这里创建了`JDK默认的线程数:CPU数+1`，这是一个经过大量测试以后给出的一个结果；线程池顾名思义，`可以重复利用现有的线程`

*  同时利用`CompletionService`来`对子任务进行汇总`
 *  合理的使用线程池已经可以充分的并行处理任务，只是在写法上有点繁琐，此时**Java1.7**中引入了`fork/join框架`；

### 3.4.fork/join框架 ###

分支/合并框架的目的是： `以递归的方式将可以并行的任务拆分成更小的任务，然后将每个子任务的结果合并起来生成整体结果`；相关代码如下：

public class ForkJoinTest extends RecursiveTask<Long> { 
        //分段阈值,即每个线程处理次数
        public static final int threshold = 10_000;
        //要累加的数字集合(即)
        private final long[] numbers;
        //当前任务集合开始下标
        private final int start;
        //当前任务集合结束下标
        private final int end;
    
        //构造方法(初始化要累加的数字集合,开始下标,结束下标)
        private ForkJoinTest(long[] numbers, int start, int end) { 
            this.numbers = numbers;
            this.start = start;
            this.end = end;
        }
    
        public static void main(String[] args) { 
            //要累加的数字集合(即)
            long[] numbers = LongStream.rangeClosed(1, 10_000_000).toArray();
    
    
            // 创建包含Runtime.getRuntime().availableProcessors()返回值作为个数的并行线程的ForkJoinPool
            ForkJoinPool forkJoinPool = new ForkJoinPool();
    
            // 提交可分解的PrintTask任务
            //Future<Long> future = forkJoinPool.submit(new ForkJoinTest(numbers, 0, numbers.length));
            //System.out.println("计算出来的总和="+future.get());
    
            //创建ForkJoin 任务
            ForkJoinTask<Long> task = new ForkJoinTest(numbers,0, numbers.length);
            Long sumAll = forkJoinPool.invoke(task);
            System.out.println("计算出来的总和=" + sumAll);
    
            // 关闭线程池
            forkJoinPool.shutdown();
        }
    
        @Override
        protected Long compute() { 
            //总处理次数
            int length = end - start;
            // 当end-start的值小于threshold时候，直接累加
            if (length <= threshold) { 
                long sum = 0;
                for (int i = start; i < end; i++) { 
                    sum += numbers[i];
                }
                return sum;
            }
            
            System.err.println("=====任务分解======");
            // 将大任务从中间切分,然后分解成两个小任务
            int middle = (start + end) / 2;
    
            //任务分解： 将大任务分解成两个小任务
            ForkJoinTest leftTask = new ForkJoinTest(numbers, start, middle);
            ForkJoinTest rightTask = new ForkJoinTest(numbers, middle, end);
    
            // 并行执行两个小任务
            leftTask.fork();
            rightTask.fork();
    
            // 注：join方法会阻塞，因此有必要在两个子任务的计算都开始之后才执行join方法
            // 把两个小任务累加的结果合并起来
            return leftTask.join() + rightTask.join();
        }
    }

执行结果:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70]

`ForkJoinPool`是**ExecutorService接口**的一个实现，子任务分配给线程池中的工作线程，同时需要把任务提交到此线程池中，需要创建`RecursiveTask<R>`的一个子类。

*  大体逻辑就是`通过fork(0进行拆分`，然后`通过join()进行结果的合并`，Java为我们提供了一个框架，我们只需要在里面填充即可，更加方便
 *  有没有更简单的方式，连拆分都省了，自动拆分合并，`Java1.8`中引入了`流`的概念；

### 3.5.并行流方式 ###

Java8引入了stream的概念，可以让我们更好的利用并行，使用流代码如下：

public class StreamTest { 
        public static void main(String[] args) { 
            // 并行流：多个线程同时运行
            System.out.println("sum = " + parallelRangedSum(10_000_000));
            // 顺序流：使用主线程，单线程
            System.out.println("sum = " + sequentialRangedSum(10_000_000));
        }
        
        //并行流
        public static long parallelRangedSum(long n) { 
            return LongStream.rangeClosed(1, n).parallel().reduce(0L, Long::sum);
        }
        //顺序流
        public static long sequentialRangedSum(long n) { 
            return LongStream.rangeClosed(1, n).sequential().reduce(0L, Long::sum);
        }
    }

以上代码是不是非常简单，对于开发者来说完全`不需要手动拆分，使用同步机制等方式`，就可以让任务并行处理，只需要对流使用`parallel()`方法，`系统自动会对任务进行拆分`，当然**前提是没有共享可变状态**

*  **并行流内部使用的也是fork/join框架**

总结  
本文使用一个求和的实例介绍并行处理的方式，可以看到Java一直在为提供更方便的并行处理而努力。

[Java_Serial_Parallel_CMS_G1]: https://www.cnblogs.com/jason1990/p/11732261.html
[JVM_Serial_Parallel_CMS_G1]: https://blog.csdn.net/qq_32038679/article/details/80616836
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxODc3NzI4NzE1_size_16_color_FFFFFF_t_70]: /images/20221024/b49399dd03844a53a31d0c9b3f5088cc.png