package javatuning.ch4.mstrwkr;import java.util.HashMap;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;public class Master {//任务队列protected Queue<Object> workQueue = new ConcurrentLinkedQueue<Object>();//Worker线程队列protected Map<String,Thread> threadMap=new HashMap<String,Thread>();//子任务处理结果集protected Map<String,Object> resultMap = new ConcurrentHashMap<String,Object>();//是否所有的子任务都结束了public boolean isComplete(){for(Map.Entry<String,Thread> entry:threadMap.entrySet()){if(entry.getValue().getState()!=Thread.State.TERMINATED){return false;}}return true;}//Master的构造，需要一个Worker进程逻辑，和需要的Worker进程数量public Master(Worker worker,int countWorker){worker.setWorkQueue(workQueue);worker.setResultMap(resultMap);for(int i=0;i<countWorker ;i++)threadMap.put(Integer.toString(i), new Thread(worker,Integer.toString(i)));}//提交一个任务public void submit(Object job){workQueue.add(job);}//返回子任务结果集public Map<String,Object> getResultMap(){return resultMap;}//开始运行所有的Worker进程，进行处理public void execute(){for(Map.Entry<String, Thread> entry:threadMap.entrySet()){entry.getValue().start();}}
}
package javatuning.ch4.mstrwkr;import java.util.Map;
import java.util.Queue;public class Worker implements Runnable{//任务队列，用于取得子任务protected Queue<Object> workQueue;//子任务处理结果集protected Map<String,Object> resultMap;public void setWorkQueue(Queue<Object> workQueue) {this.workQueue = workQueue;}public void setResultMap(Map<String, Object> resultMap) {this.resultMap = resultMap;}//子任务处理的逻辑，在子类中实现具体逻辑public Object handle(Object input){return input;}@Overridepublic void run() {while (true) {//获取子任务Object input = workQueue.poll();if (input == null) break;//处理子任务Object re=handle(input);//将处理结果写入结果集resultMap.put(Integer.toString(input.hashCode()), re);}}
}
package javatuning.ch4.mstrwkr;import java.util.Map;
import java.util.Random;
import java.util.Set;import org.junit.Test;public class TestMasterWorker {public class PlusWorker extends Worker{public Object handle(Object input){Integer i=(Integer)input;return i*i*i;}}@Testpublic void testMasterWorker() {Master m=new Master(new PlusWorker(),5);for(int i=0;i<100;i++)m.submit(i);m.execute();int re=0;Map<String,Object> resultMap=m.getResultMap();while(resultMap.size()>0 || !m.isComplete()){Set<String> keys=resultMap.keySet();String key=null;for(String k:keys){key=k;break;}Integer i=null;if(key!=null)i=(Integer)resultMap.get(key);if(i!=null)re+=i;if(key!=null)resultMap.remove(key);}System.out.println("testMasterWorker:"+re);}@Testpublic void testPlus(){int re=0;for(int i=0;i<100;i++){re+=i*i*i;}System.out.println("testPlus:"+re);}}