Master-Worker模式

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

Master：

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<Task>();

	//2 使用HashMap去承装worker对象
	private HashMap<String,Thread> workers = new HashMap<String,Thread>();
	
	//3 使用一个容器承装每一个worker并发执行任务的结果集,因为worker返回来结果集会是并发的,所有用这个map
	private ConcurrentHashMap<String,Object> resultMap = new ConcurrentHashMap<String,Object>();
	
	//4 构造方法
	public Master(Worker worker,int workerCount){
		
		//在worker中设置好对Master中workQueue,workers的引用
		//每一个worker对象都需要Master的引用。workQueue用于任务的领取，resultMap用于任务的提交
		worker.setWorkerQueue(this.workQueue);
		worker.setResultMap(this.resultMap);
		
		for(int i = 0;i < workerCount; i++ ){
			//key表示每一个worker的名字，value表示线程执行对象
			workers.put("子节点"+ Integer.toString(i),new Thread(worker));
		}
		
	}
	
	//5 提交方法(将任务放入到Master的workQueue里去)
	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();
		}
	}
	
	//8 判断线程是否执行完毕
	public boolean isComplete(){
		
		for(Map.Entry<String, Thread> me : workers.entrySet()){
			if(me.getValue().getState() != Thread.State.TERMINATED){
				return false;
			}
		}
		return true;
	}
	
	//9 返回结果集
	public int getReulst(){
		
		int ret = 0;
		for(Map.Entry<String, Object> me: resultMap.entrySet()){
			//汇总的逻辑(不定，可加，可其他操作，根据业务需求)
			ret += (Integer)me.getValue();
		}
		return ret;
	}
}

Worker：

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

public class Worker implements Runnable{

	private ConcurrentLinkedQueue<Task> workQueue;
	private ConcurrentHashMap<String, Object> resultMap;

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

	public void setResultMap(ConcurrentHashMap<String, Object> resultMap) {
		this.resultMap = resultMap;
	}
	
	@Override
	public void run() {
		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 {
			//表示处理task任务的耗时,可能是数据的加工,也可能是操作数据库的...
			Thread.sleep(500);
			
			output = input.getPrice();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return output;
	}
}

Task：

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;
	}

	
}

主类：

import java.util.Random;

public class Main {

	public static void main(String[] args) {
		
		Master master = new Master(new Worker(), 10);
		
		Random r = new Random();
		for(int i = 1;i<=100;i++){
			Task t = new Task();
			t.setId(i);
			t.setName("任务" + 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 ret = master.getReulst();
				System.out.println("执行结果:" + ret + ",耗时:" + end);
				break;
			}
		}
		
	}
	
}

输出：

执行结果:51783,耗时:5020

其实还有疑问，100个任务，10个Worker，是如何做到负载均衡的，就是怎么实现给10个Worker每个分配10个任务的…