Java 并行程序设计模式 （Master-Worker模式）

2024-04-13 08:38
文章标签 java 设计模式 模式 worker master 并行程序

本文主要是介绍Java 并行程序设计模式 （Master-Worker模式），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

并行设计模式属于设计优化的一部分，它是对一些常用的多线程结构的总结和抽象。与串行程序相比，并行程序的结构通常更为复杂。因此，合理的使用并行模式在多线程开发中，更具有积极意义。并行程序设计模式主要有 Future模式 、Master-Worker模式、Guarded Suspension模式、不变模式和生产者-消费者模式，本文主要讲解   Master-Worker模式

Master-Worker模式是常用的并行模式之一，核心思想是，系统由两类进程协作：Master进程和Worker进程。Master进程负责接收和分配任务，worker进程负责子任务。当各个Worker进程将子任务处理完成后，将结果返回给Master进程，由Master进程做归纳和汇总，从而得到系统的最终结果。
示意图如下：


Master-Worker的代码实现

Worker实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
import java.util.Map;
import java.util.Queue;
public class Worker implements Runnable {
     //任务队列,用于取得子任务
     protected Queue<Object> workQueue;
     //子任务处理结果集
     protected Map<String, Object> resultMap;
     public void setWorkQueue(Queue<Object> workQueue) {
         this .workQueue = workQueue;
     }
     public void setResultMap(Map<String, Object> resultMap) {
         this .resultMap = resultMap;
     }
     //子任务处理的逻辑，在子类中实现具体逻辑
     public Object handle(Object input){
         return  input;
     }
     public void run() {
         while ( true ){
             Object input = workQueue.poll();
             if  (input ==  null )  break ;
             //处理子任务
             Object result = handle(input);
             System.out.println(input.hashCode());
             //将处理结果写入结果集
             resultMap.put(Integer.toString(input.hashCode()), result);
         }
         
     }
}

Master实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
import java.util.HashMap;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
public class Master {
     //任务队列
     protected Queue<Object> workQueue =  new  ConcurrentLinkedQueue<Object>();
     
     //Worker进程队列
     
     protected Map<String, Thread> threadMap =  new  HashMap<String, Thread>();
     
     //子任务处理结果集
     
     protected Map<String, Object>  resultMap =  new  ConcurrentHashMap<String, Object>();
     
     
     //是否所有子任务都结束了
     public boolean isComplete(){
         for (Map.Entry<String, Thread> entry:threadMap.entrySet()){
             if  (entry.getValue().getState() != Thread.State.TERMINATED) {
                 return  false ;
             }
         }
         return  true ;
     }
     
     //Master的构造，需要一个Worker进程逻辑，和需要的Worker进程数量
     public Master(Worker worker, int countWorker){
         worker.setWorkQueue(workQueue);
         worker.setResultMap(resultMap);
         for (int i=0; i < countWorker; i++){
             threadMap.put(Integer.toString(i),  new  Thread(worker,Integer.toString(i)));
         }
     }
     //提交一个任务
     public void submit(Object job){
         workQueue.add(job);
     }
     
     //返回结果集
     public Map<String, Object> getResultMap(){
         return  resultMap;
     }
     
     //开始运行所有的worker进程，进行处理
     public void execute(){
         for (Map.Entry<String, Thread> entry:threadMap.entrySet()){
             entry.getValue().start();
         }
     }
}

运用这个小框架计算1——100的立方和
PlusWorker的实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import java.util.concurrent.TimeUnit;
public class PlusWorker extends Worker {
     @Override
     public Object handle(Object input) {
         Integer i = (Integer) input;
         try  {
             TimeUnit.SECONDS.sleep(1);
         }  catch  (InterruptedException e) {
             // TODO Auto-generated catch block
             e.printStackTrace();
         }
         return  i * i * i;
     }
}

Test Main方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
import java.util.Map;
import java.util.Set;
/**
  * 计算1^3 + 2^3 + 3^3 +......+ 100^3
  * @author tanlk
  * @date 2017年7月21日下午2:30:17
  */
public class Test {
     public static void main(String[] args) {
         Master master =  new  Master( new  PlusWorker(), 5);  //使用5个worker
         
         //添加worker到workQueue， workQueue是ConcurrentLinkedQueue,5个PlusWorker进程同时抢workQueue里面的数据
         for (int i = 1; i<= 100; i++){
             master.submit(i);
         }
         
         master.execute();
         
         int result = 0;
         
         Map<String, Object> resultMap = master.getResultMap();
         
         //不需要全部结果执行完就可以返回结果
         while (resultMap.size() > 0 || !master.isComplete()){
             Set<String> keys = resultMap.keySet();
             
             String key =  null  ;
             for  (String str : keys) {
                 key = str;
                 break ;
             }
             
             Integer i =  null ;
             if  (key !=  null ) {
                 i = (Integer) resultMap.get(key);
             }
             if  (i !=  null ) {
                 result = result + i;
             }
             if  (key !=  null ) {
                 resultMap.remove(key); //删除已经取出的数据
             }
             
         }
         
         System.out.println(result);
     }
}

总结
Master-Worker模式是一种将串行任务并行化的方案，被分解的子任务在系统中可以被并行处理，同时，如果有需要，Master进程不需要等待所有子任务都完成计算，就可以根据已有的部分结果集计算最终结果集。


类似的框架Fork/Join
 Fork/Join框架是Java 7提供的一个用于并行执行任务的框架，是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。

Fork/Join架构的主要类



RecursiveAction供不需要返回值的任务继续。
RecursiveTask通过泛型参数设置计算的返回值类型。

ForkJoinPool提供了一系列的submit方法，计算任务。ForkJoinPool默认的线程数通过Runtime.availableProcessors()获得，因为在计算密集型的任务中，获得多于处理性核心数的线程并不能获得更多性能提升。

public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
doSubmit(task);
return task;
}

Fork/Join实现并行计算
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.TimeUnit;
/**
  * Fork/Join来优化并行计算
  * 计算1^3 + 2^3 + 3^3 +......+ 100^3
  * 
  * 继承RecursiveTask，重写compute
  * @author tanlk
  * @date 2017年7月21日下午3:24:57
  */
public class Calculate extends RecursiveTask<Integer> {
     /**
      * 
      */
     private static final long serialVersionUID = -3363693028643602343L;
     final static int THRESHOLD = 4;
     private Integer start;
     private Integer length;
     public Calculate(Integer start, Integer length) {
         this .length = length;
         this .start = start;
     }
     @Override
     protected Integer compute() {
         System.out.println( "Calculate.compute() start:" +start+ ",length:"  +length );
         int result = 0;
         
         if  (length < THRESHOLD) {  // 小于临界值，直接计算
             for  (int i = start; i < start + length; i++) {
                 try  {
                     TimeUnit.SECONDS.sleep(1);
                 }  catch  (InterruptedException e) {
                     e.printStackTrace();
                 }
                 result = result + (i * i * i);
             }
         }  else  {  // 分而治之
             int split = length / 2;
             Calculate c1 =  new  Calculate(start, split);
             Calculate c2 =  new  Calculate(start + split, length - split);
             c1.fork();
             c2.fork(); //fork拆分子任务
             result = c1.join() + c2.join(); //join合并子任务结果
         }
         return  result;
     }
     
     
     public static void main(String[] args) throws InterruptedException, ExecutionException {
          Calculate calculate =  new  Calculate(1,100);
          long start = System.currentTimeMillis();
          ForkJoinPool forkJoinPool =  new  ForkJoinPool();
          forkJoinPool.submit(calculate);
          Integer result = calculate.get();
          System.out.println( "结果：" +result + ",耗时：" +(System.currentTimeMillis() - start));
     }
     
}

关于fork/join和Master-Worker模式的区别，欢迎大家留言讨论

我认为Master-Worker 这种方式对于大小相同，且任务大小适中可控的任务来说是不错的。但是当任务大小不一致的时候就会遇到问题。就是说，一个worker可能被缠在冗长的任务中，然后其他的worker闲着没事做。

Fork join并不是预先拆分所有任务，而是在执行时动态的决定拆分


这篇关于Java 并行程序设计模式 （Master-Worker模式）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


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

相关文章

Java 正则表达式的使用实战案例

Java 正则表达式的使用实战案例

《Java正则表达式的使用实战案例》本文详细介绍了Java正则表达式的使用方法,涵盖语法细节、核心类方法、高级特性及实战案例,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要... 目录一、正则表达式语法详解1. 基础字符匹配2. 字符类（[]定义）3. 量词（控制匹配次数）4. 边
阅读更多...
Java Scanner类解析与实战教程

Java Scanner类解析与实战教程

《JavaScanner类解析与实战教程》JavaScanner类（java.util包）是文本输入解析工具,支持基本类型和字符串读取,基于Readable接口与正则分隔符实现,适用于控制台、文件输... 目录一、核心设计与工作原理1.底层依赖2.解析机制A.核心逻辑基于分隔符（delimiter）和模式匹
阅读更多...
Java中的stream流分组示例详解

Java中的stream流分组示例详解

《Java中的stream流分组示例详解》Java8StreamAPI以函数式风格处理集合数据,支持分组、统计等操作,可按单/多字段分组,使用String、Map.Entry或Java16record... 目录什么是stream流1、根据某个字段分组2、按多个字段分组（组合分组）1、方法一：使用 Stri
阅读更多...
Java+AI驱动实现PDF文件数据提取与解析

Java+AI驱动实现PDF文件数据提取与解析

《Java+AI驱动实现PDF文件数据提取与解析》本文将和大家分享一套基于AI的体检报告智能评估方案,详细介绍从PDF上传、内容提取到AI分析、数据存储的全流程自动化实现方法,感兴趣的可以了解下... 目录一、核心流程：从上传到评估的完整链路二、第一步：解析 PDF，提取体检报告内容1. 引入依赖2. 封装
阅读更多...
使用Spring Cache本地缓存示例代码

使用Spring Cache本地缓存示例代码

《使用SpringCache本地缓存示例代码》缓存是提高应用程序性能的重要手段,通过将频繁访问的数据存储在内存中,可以减少数据库访问次数,从而加速数据读取,：本文主要介绍使用SpringCac... 目录一、Spring Cache简介核心特点：二、基础配置1. 添加依赖2. 启用缓存3. 缓存配置方案方案
阅读更多...
Java实现复杂查询优化的7个技巧小结

Java实现复杂查询优化的7个技巧小结

《Java实现复杂查询优化的7个技巧小结》在Java项目中,复杂查询是开发者面临的“硬骨头”,本文将通过7个实战技巧,结合代码示例和性能对比,手把手教你如何让复杂查询变得优雅,大家可以根据需求进行选择... 目录一、复杂查询的痛点：为何你的代码“又臭又长”1.1冗余变量与中间状态1.2重复查询与性能陷阱1.
阅读更多...
深度剖析SpringBoot日志性能提升的原因与解决

深度剖析SpringBoot日志性能提升的原因与解决

《深度剖析SpringBoot日志性能提升的原因与解决》日志记录本该是辅助工具,却为何成了性能瓶颈,SpringBoot如何用代码彻底破解日志导致的高延迟问题,感兴趣的小伙伴可以跟随小编一起学习一下... 目录前言第一章：日志性能陷阱的底层原理1.1 日志级别的“双刃剑”效应1.2 同步日志的“吞吐量杀手”
阅读更多...
Spring创建Bean的八种主要方式详解

Spring创建Bean的八种主要方式详解

《Spring创建Bean的八种主要方式详解》Spring（尤其是SpringBoot）提供了多种方式来让容器创建和管理Bean,@Component、@Configuration+@Bean、@En... 目录引言一、Spring 创建 Bean 的 8 种主要方式1. @Component 及其衍生注解
阅读更多...
SpringBoot通过main方法启动web项目实践

SpringBoot通过main方法启动web项目实践

《SpringBoot通过main方法启动web项目实践》SpringBoot通过SpringApplication.run()启动Web项目,自动推断应用类型,加载初始化器与监听器,配置Spring... 目录1. 启动入口：SpringApplication.run()2. SpringApplicat
阅读更多...
Java利用@SneakyThrows注解提升异常处理效率详解

Java利用@SneakyThrows注解提升异常处理效率详解

《Java利用@SneakyThrows注解提升异常处理效率详解》这篇文章将深度剖析@SneakyThrows的原理,用法,适用场景以及隐藏的陷阱,看看它如何让Java异常处理效率飙升50%,感兴趣的... 目录前言一、检查型异常的“诅咒”：为什么Java开发者讨厌它1.1 检查型异常的痛点1.2 为什么说
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2