Java并发编程---Master-Worker模式

傷城~ 2022-06-02 06:16 146阅读 0赞

**一.概念**

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

![Center][]

**二.代码示例**

package com.thread.master;
    
    import java.util.Random;
    
    public class Main {
    
    	public static void main(String[] args) {
    		// TODO Auto-generated method stub
            Master master = new Master(new Worker(), 200);    //生成20个worker(worker越多执行时间越快)
            
            //20个线程处理100个元素
            Random r = new Random();
            for(int i=1; i<=100; i++){
            	Task t = new Task();
            	t.setId(i);
            	t.setPrice(r.nextInt(1000));
            	master.submit(t);
            }
            master.execute();
            long start = System.currentTimeMillis();
            
            while(true){
            	if(master.isComplete()){
            		long end = System.currentTimeMillis() - start;
            		int priceResult = master.getResult();
            		System.out.println("最终结果:" + priceResult + ", 执行时间: "  + end);
            		break;
            	}
            }
    	}
    
    }

package com.thread.master;
    
    public class Task {
    
    	private int id;
    	private int price;
    	public int getId() {
    		return id;
    	}
    	public void setId(int id) {
    		this.id = id;
    	}
    	public int getPrice() {
    		return price;
    	}
    	public void setPrice(int price) {
    		this.price = price;
    	}
    	
    }

package com.thread.master;
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.ConcurrentLinkedDeque;
    
    public class Master {
    
    	//1.有一个盛放任务的容器
    	private ConcurrentLinkedDeque<Task> workQueue = new ConcurrentLinkedDeque<Task>();
    	
    	//2.需要有一个盛放worker的集合
    	private HashMap<String, Thread> workers = new HashMap<String, Thread>();
    	
    	//3.需要有一个盛放每一个Worker执行任务的结果集合
    	private ConcurrentHashMap<String, Object> resultMap = new ConcurrentHashMap<String,Object>();
    
    	//4.构造方法
    	public Master(Worker worker , int workerCount) {
    		worker.setWorkQueue(this.workQueue);
    		worker.setResultMap(this.resultMap);
    		
    		for(int i = 0; i<workerCount; i++){
    			this.workers.put(Integer.toString(i), new Thread(worker));
    		}
    	}
    	
    	//5.需要一个提交任务的方法
    	public void submit(Task task){
    		this.workQueue.add(task);
    	}
    	
    	//6.需要有一个执行的方法,启动所有的worker方法区执行任务
    	public void execute(){
    		for(Map.Entry<String, Thread> me : workers.entrySet()){
    			me.getValue().start();
    		}
    	}
    	
    	//7.判断是否运行结束的方法
    	public boolean isComplete(){
    		for(Map.Entry<String, Thread> me : workers.entrySet()){
    			if(me.getValue().getState() != Thread.State.TERMINATED){
    				return false;
    			}
    		}
    		return true;
    	}
    	
    	//8.计算结果方法
    	public int getResult(){
    		int priceResult = 0;
    		for(Map.Entry<String, Object> me : resultMap.entrySet()){
    			priceResult += (Integer) me.getValue();
    		}
    		return priceResult;
    	}
    }

package com.thread.master;
    
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.ConcurrentLinkedDeque;
    
    public class Worker implements Runnable {
    
    	private ConcurrentLinkedDeque<Task> workQueue;
    	private ConcurrentHashMap<String, Object> resultMap;
    	
    	public void setWorkQueue(ConcurrentLinkedDeque<Task> workQueue){
    		this.workQueue = workQueue;
    	}
    	
    	public void setResultMap(ConcurrentHashMap<String, Object> resultMap){
    		this.resultMap = resultMap;
    	}
    	@Override
    	public void run() {
    		// TODO Auto-generated method stub
           while(true){
        	   Task input = this.workQueue.poll();
        	   if (input ==null) break;
        	   Object output = handle(input);      //不为空则处理业务逻辑
        	   this.resultMap.put(Integer.toString(input.getId()), output);
           }
    	}
    	
    	private Object handle(Task input){
    		Object output = null;
    		try {
    			//处理任务的耗时,比如进行操作数据库
    			Thread.sleep(500);
    			output = input.getPrice();
    		} catch (InterruptedException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    		return output;
    	}
    
    }

[Center]: /images/20220602/f962772836004ce98645285a3e31a32a.png