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编译生成多个.class文件的原理和作用

Java编译生成多个.class文件的原理和作用

《Java编译生成多个.class文件的原理和作用》作为一名经验丰富的开发者,在Java项目中执行编译后,可能会发现一个.java源文件有时会产生多个.class文件,从技术实现层面详细剖析这一现象... 目录一、内部类机制与.class文件生成成员内部类（常规内部类）局部内部类（方法内部类）匿名内部类二、
阅读更多...
SpringBoot实现数据库读写分离的3种方法小结

SpringBoot实现数据库读写分离的3种方法小结

《SpringBoot实现数据库读写分离的3种方法小结》为了提高系统的读写性能和可用性,读写分离是一种经典的数据库架构模式,在SpringBoot应用中,有多种方式可以实现数据库读写分离,本文将介绍三... 目录一、数据库读写分离概述二、方案一：基于AbstractRoutingDataSource实现动态
阅读更多...
Linux系统配置NAT网络模式的详细步骤(附图文)

Linux系统配置NAT网络模式的详细步骤(附图文)

《Linux系统配置NAT网络模式的详细步骤(附图文)》本文详细指导如何在VMware环境下配置NAT网络模式,包括设置主机和虚拟机的IP地址、网关,以及针对Linux和Windows系统的具体步骤,... 目录一、配置NAT网络模式二、设置虚拟机交换机网关2.1 打开虚拟机2.2 管理员授权2.3 设置子
阅读更多...
Springboot @Autowired和@Resource的区别解析

Springboot @Autowired和@Resource的区别解析

《Springboot@Autowired和@Resource的区别解析》@Resource是JDK提供的注解,只是Spring在实现上提供了这个注解的功能支持,本文给大家介绍Springboot@... 目录【一】定义【1】@Autowired【2】@Resource【二】区别【1】包含的属性不同【2】@
阅读更多...
springboot循环依赖问题案例代码及解决办法

springboot循环依赖问题案例代码及解决办法

《springboot循环依赖问题案例代码及解决办法》在SpringBoot中,如果两个或多个Bean之间存在循环依赖（即BeanA依赖BeanB,而BeanB又依赖BeanA）,会导致Spring的... 目录1. 什么是循环依赖？2. 循环依赖的场景案例3. 解决循环依赖的常见方法方法 1：使用 @La
阅读更多...
Java枚举类实现Key-Value映射的多种实现方式

Java枚举类实现Key-Value映射的多种实现方式

《Java枚举类实现Key-Value映射的多种实现方式》在Java开发中,枚举(Enum)是一种特殊的类,本文将详细介绍Java枚举类实现key-value映射的多种方式,有需要的小伙伴可以根据需要... 目录前言一、基础实现方式1.1 为枚举添加属性和构造方法二、http://www.cppcns.co
阅读更多...
Elasticsearch 在 Java 中的使用教程

Elasticsearch 在 Java 中的使用教程

《Elasticsearch在Java中的使用教程》Elasticsearch是一个分布式搜索和分析引擎,基于ApacheLucene构建,能够实现实时数据的存储、搜索、和分析,它广泛应用于全文... 目录1. Elasticsearch 简介2. 环境准备2.1 安装 Elasticsearch2.2 J
阅读更多...
Java中的String.valueOf()和toString()方法区别小结

Java中的String.valueOf()和toString()方法区别小结

《Java中的String.valueOf()和toString()方法区别小结》字符串操作是开发者日常编程任务中不可或缺的一部分,转换为字符串是一种常见需求,其中最常见的就是String.value... 目录String.valueOf()方法方法定义方法实现使用示例使用场景toString()方法方法
阅读更多...
Java中List的contains()方法的使用小结

Java中List的contains()方法的使用小结

《Java中List的contains()方法的使用小结》List的contains()方法用于检查列表中是否包含指定的元素,借助equals()方法进行判断,下面就来介绍Java中List的c... 目录详细展开1. 方法签名2. 工作原理3. 使用示例4. 注意事项总结结论：List 的 contain
阅读更多...
Java实现文件图片的预览和下载功能

Java实现文件图片的预览和下载功能

《Java实现文件图片的预览和下载功能》这篇文章主要为大家详细介绍了如何使用Java实现文件图片的预览和下载功能,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... Java实现文件（图片）的预览和下载 @ApiOperation("访问文件") @GetMapping("
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2