Master-worker模式是经常使用的并行计算模式,它的核心思想是系统是由两类进程协助工做。Master负责接收和分配任务,worker负责处理子任务。当各个worker子进程处理完成后,会返回结果给master,由master作概括和总结。其好处是能将一个大任务分解成若干个小任务,并行执行,从而提升系统的吞吐量。数据库
1.定义一个要执行的任务类dom
public class Task{ private int id; private int price; //getter/setter()省略 }
2.定义Master负责收发任务性能
//注意worker线程的数量 根据机器的性能来添加 //Runtime.getRuntime().availableProcessors() public class Master{ //1.定义一个盛听任务的容器 private ConcurrentLinkedQueue<Task> workQueue = new ConcurrentLinkedQueue<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 workCount){ 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; } }
3.定义一个worker处理子任务测试
public class Worker implements Runnable{ private ConcurrentLinkedQueue<Task> workQueue; private ConcurrentHashMap<String, Object> resultMap; public void setWorkQueue(ConcurrentLinkedQueue<Task> workQueue) { this.workQueue = workQueue; } public void setResultMap(ConcurrentHashMap<String, Object> resultMap) { this.resultMap = resultMap; } 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 { //处理任务的耗时。。 好比说进行操做数据库。。。 Thread.sleep(500); output = input.getPrice(); } catch (InterruptedException e) { e.printStackTrace(); } return output; } }
4.定义一个测试类this
public class Main{ public static void main(String[] args) { Master master = new Master(); 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.currentTimeMills(); while(true){ if(maste.isComplete()){ long end = System.currentTimeMills(); int priceResult = master.getResult(); System.out.println("最终结果:" + priceResult + ", 执行时间:" + end); break; } } } }
结果以下spa