java线程学习之Master-Worker模式

Posted songlove

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了java线程学习之Master-Worker模式相关的知识,希望对你有一定的参考价值。

    Masker-Worker的核心思想是有两类进程(Masker进程和Worker进程)协作完成任务。Masker进程负责接收和分配任务,Worker负责处理子任务,当各个Worker子进程完成任务后会将结果返回给Masker,由Masker做归纳和总结。其好处在于能将一个大任务分解成若干个小任务,并行执行,从而提供系统的吞吐量。

                      技术图片

      技术图片

这个模型主要用于主线程可以分为若干子线程的情形,各子线程之间不会相互影响。

 举个例子,这个例子是创建20个Worker去处理100个任务,每个任务是在0-1000的范围内随即一个整数,最后将这些数字相加。分别有Main,Masker,Worker,Task这四个类

技术图片
 1 import java.util.Random;
 2 
 3 public class Main {
 4 
 5     public static void main(String[] args) {
 6         
 7         Master master = new Master(new Worker(), 20);//生成20个work去处理这个任务
 8         
 9         Random r = new Random();
10         for(int i = 1; i <= 100; i++){
11             Task t = new Task();
12             t.setId(i);
13             t.setPrice(r.nextInt(1000));
14             master.submit(t);
15         }
16         master.execute();
17         long start = System.currentTimeMillis();
18         
19         while(true){
20             if(master.isComplete()){
21                 long end = System.currentTimeMillis() - start;
22                 int priceResult = master.getResult();
23                 System.out.println("最终结果:" + priceResult + ", 执行时间:" + end);
24                 break;
25             }
26         }
27         
28     }
29 }
Main
技术图片
 1 import java.util.HashMap;
 2 import java.util.Map;
 3 import java.util.concurrent.ConcurrentHashMap;
 4 import java.util.concurrent.ConcurrentLinkedQueue;
 5 
 6 public class Master {
 7 
 8     //1 有一个盛放任务的容器
 9     private ConcurrentLinkedQueue<Task> workQueue = new ConcurrentLinkedQueue<Task>();
10     
11     //2 需要有一个盛放worker的集合
12     private HashMap<String, Thread> workers = new HashMap<String, Thread>();
13     
14     //3 需要有一个盛放每一个worker执行任务的结果集合
15     private ConcurrentHashMap<String, Object> resultMap = new ConcurrentHashMap<String, Object>();
16     
17     //4 构造方法
18     public Master(Worker worker , int workerCount){
19         worker.setWorkQueue(this.workQueue);    //因为是消费workQueue里面的数据,所以workQueue放进去
20         worker.setResultMap(this.resultMap);    //消费完之后要把结果集返回给Masker,所以要有resultMap应的引用
21         
22         for(int i = 0; i < workerCount; i ++){
23             this.workers.put(Integer.toString(i), new Thread(worker));//创建一个线程并对它起个名字用i来表示
24         }
25         
26     }
27     
28     //5 需要一个提交任务的方法
29     public void submit(Task task){
30         this.workQueue.add(task);
31     }
32     
33     //6 需要有一个执行的方法,启动所有的worker方法去执行任务
34     public void execute(){
35         for(Map.Entry<String, Thread> me : workers.entrySet()){
36             me.getValue().start();    //循环这已经装好的Works,让它们都起动起来
37         }
38     }
39 
40     //7 判断是否运行结束的方法
41     public boolean isComplete() {
42         for(Map.Entry<String, Thread> me : workers.entrySet()){//循环这写Works线程,判断其状态
43             if(me.getValue().getState() != Thread.State.TERMINATED){
44                 return false;
45             }
46         }    
47         return true;
48     }
49 
50     //8 计算结果方法
51     public int getResult() {
52         int priceResult = 0;
53         for(Map.Entry<String, Object> me : resultMap.entrySet()){
54             priceResult += (Integer)me.getValue();
55         }
56         
57         return priceResult;
58         
59     }
60     
61     
62     
Masker
技术图片
 1 import java.util.concurrent.ConcurrentHashMap;
 2 import java.util.concurrent.ConcurrentLinkedQueue;
 3 
 4 public class Worker implements Runnable {
 5 
 6     private ConcurrentLinkedQueue<Task> workQueue;
 7     private ConcurrentHashMap<String, Object> resultMap;
 8     
 9     public void setWorkQueue(ConcurrentLinkedQueue<Task> workQueue) {
10         this.workQueue = workQueue;
11     }
12 
13     public void setResultMap(ConcurrentHashMap<String, Object> resultMap) {
14         this.resultMap = resultMap;
15     }
16     
17     @Override
18     public void run() {
19         while(true){
20             Task input = this.workQueue.poll();//不断地从Mask的workQueue将任务取出来
21             if(input == null) break;   //为空的话证明它已经消费完了
22             Object output = handle(input);//否则交给handle方法进行处理
23             this.resultMap.put(Integer.toString(input.getId()), output);//将任务的id,和结果集放到resultMap里
24         }
25     }
26 
27     private Object handle(Task input) {
28         Object output = null;
29         try {
30             //处理任务的耗时。。 比如说进行操作数据库。。。
31             Thread.sleep(500);
32             output = input.getPrice();//得到处理完的结果
33         } catch (InterruptedException e) {
34             e.printStackTrace();
35         }
36         return output;
37     }
38 
39 
40 
41 }
Worker
技术图片
 1 public class Task {
 2 
 3     private int id;
 4     private int price ;
 5     public int getId() {
 6         return id;
 7     }
 8     public void setId(int id) {
 9         this.id = id;
10     }
11     public int getPrice() {
12         return price;
13     }
14     public void setPrice(int price) {
15         this.price = price;
16     } 
17     
18 }
Task

运行结果为:

技术图片

执行时间是每个任务休眠0.5s乘以100个任务,除以20个Worker,为2.5s。创建的Worker数并不是越大越好,因为创建Worker也需要花费时间。

 

以上是关于java线程学习之Master-Worker模式的主要内容,如果未能解决你的问题,请参考以下文章

Java多线程Master-Worker模式

并行设计模式-- Master-Worker模式

JAVA基础学习之-AQS的实现原理分析

java线程模型Master-Worker

多线程:多线程设计模式:Master-Worker模式

并行设计模式-- Future模式