Master-worker设计模式

Master-worker模式是常用的并行计算模式。它的核心思想是系统是由两类进程协助工作：Master进行和worker进程。Master负责接收和分配任务，worker负责处理子任务。当各个worker子进程处理完成后，会返回结果给master，由master做归纳和总结。其好处是能将一个大任务分解成若干个小任务，并行执行，从而提高系统的吞吐量。

代码实现：

Master：

package com.java.day04_mode_masterworker;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

public class Master {

    //1.首先有承装任务的容器
    private ConcurrentLinkedQueue<Task> workQueue= new ConcurrentLinkedQueue<>();
    
    //2.有承装worker的容器
    private HashMap<String, Thread> workers = new HashMap<>();
    
    //3.承装worker结果集的容器
    private ConcurrentHashMap<String , Object> results = new ConcurrentHashMap<>();
    
    //4.构造方法
    public Master(Worker worker,int workerCount){
        worker.setWorkQueue(workQueue);
        worker.setResults(results);
        
        for (int i = 0; i < workerCount; i++) {
            workers.put("节点"+Integer.toString(i), new Thread(worker));
        }
    }
    
    //5.往任务队列里提交任务的方法
    public void submit(Task task){
        this.workQueue.add(task);
    }
    
    //6.启动应用程序，让所有的worker工作
    public void excute(){
        //map的循环方式
        for(Map.Entry<String, Thread> me : workers.entrySet()){
            me.getValue().start();
        }
    }

    //判断线程是否执行完毕:当所有的任务执行完毕，线程也停止
    public boolean isComplete() {
        
        for(Map.Entry<String, Thread> me : workers.entrySet()){
            if(me.getValue().getState() != Thread.State.TERMINATED){
                return false;
            }
        }
        
        return true;
    }

    //返回结果集合
    public long getResult() {
        long result = 0;
        
        for(Map.Entry<String, Object> me : results.entrySet()){
            result+=(Integer)me.getValue();
        }
        return result;
    }
    
    
    
}

Worker：

package com.java.day04_mode_masterworker;

import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

public class Worker implements Runnable{
    
    private ConcurrentLinkedQueue<Task> workQueue;
    private ConcurrentHashMap<String, Object> results;
    

    @Override
    public void run() {
        while(true){
            Task input = workQueue.poll();
            //任务执行完之后，break，线程停止
            if(input == null) break;
            Object output = handle(input);
            
            results.put(Integer.toString( input.getId()), output);
        }
        
    }

    //可以把下面的方法提取出来，在worker里面返回空，然后写一个子类继承worker，在worker里面重写此方法，解耦，灵活度也更高，Object output = 子类.handle(input);
    //对数据的具体操作
    private Object handle(Task input) {
        Object result = null;
        
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        result=input.getPrice();
        
        
        return result;
    }

    public void setWorkQueue(ConcurrentLinkedQueue<Task> workQueue) {
        this.workQueue=workQueue;
    }

    public void setResults(ConcurrentHashMap<String, Object> results) {
        this.results=results;
    }

    
    
}

Task:

package com.java.day04_mode_masterworker;

public class Task {
    private int id;
    private String name;
    private int price;
    public int getId() {
        return id;
    }
    public void setId(int id) {
        this.id = id;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public int getPrice() {
        return price;
    }
    public void setPrice(int price) {
        this.price = price;
    }
    
    
}

Main:

package com.java.day04_mode_masterworker;

public class Main {
    
    public static void main(String[] args) {
        
        //这一块的10（线程数）要根据机器具体的性能来
        //Runtime.getRuntime().availableProcessors():我机器可以用的process数量
        
        //创建master，和所需要的worker
        Master master = new Master(new Worker(),10);
        
        //提交任务
        for (int i = 1; i < 101; i++) {
            Task t = new Task();
            t.setId(i);
            t.setName("任务"+i);
            t.setPrice(i+1000);
            
            master.submit(t);
        }
        
        //执行任务
        master.excute();
        
        long start  = System.currentTimeMillis();
        
        //判断任务是否执行完毕
        while(true){
            if(master.isComplete()){
                long end = System.currentTimeMillis()-start;
                //任务执行完成，返回结果集
                long result = master.getResult();
                System.out.println("执行了"+end+"秒，最终结果为："+result);
                break;
            }
        }
        
        
        
        
        
        
    }
    
    
}

其中上面的代码还可以进行优化