Worker

Master-Worker设计模式核心思想是将原来串行的逻辑并行化，并将逻辑拆分成很多独立模块并行执行,其中主要包含两个主要组件Master和Worker，Master主要讲逻辑进行查分，拆分为互相独立的部分，同时维护了Worker队列，将每个独立部分下发到多个Worker并行执行，Worker主要进行实际逻辑计算，并将结果返回给Master。其核心框架如下：Master部分实现代码package com.yf.designpattern.masterworker;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 count）{
worker.setResultMap（resultMap）;
worker.setWorkQueue（workQueue）;
for（int i=0;i<count;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（）;
}
}
}Master主要维护了任务队列、Worker队列、结果队列和开启工作线程，添加任务等逻辑，可以通过Master添加任务，获取结果，具体任务执行过程在WorkerWorker的主要结构如下：package com.yf.designpattern.masterworker;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;
} @Override
public void run（） {
while （true） {
//获取子任务
Object input = this.workQueue.poll（）;
if （input == null）
break;
//处理子任务
Object re = this.handle（input）;
//将处理结果写回结果集
this.resultMap.put（Integer.toString（input.hashCode（））, re）;
} }}Worker主要是从Master的任务队列中获取一个任务，并且执行，将结果保存到Master的ResultMap中，每个Worker都持有Master的工作队列和ResultMap。以上就是Master-Worker模式的核心框架，还可以将上述框架进行扩展，扩展为分布式结构，Master与Worker分布在不同机器，Master与Worker之间通过一定协议来进行通信，同时Worker还可以水平扩展为多台，能够支撑大压力、高并发需求。例如现在需要计算1到100，每个数字的立方相加的结果这里可以将计算立方的逻辑交个每个Worker去执行，Worker计算完成后，将结果保存到Master的ResultMap中，Master只需要检查ResultMap是否有元素，有就进行相加计算，而不用等待每个数字都计算完成，Worker的具体实现如下：package com.yf.designpattern.masterworker;public class PlusWorker extends Worker {

@Override
public Object handle（Object input）{
Integer i=（Integer）input;
return i*i*i;
}
}测试代码如下：package com.yf.designpattern.masterworker;import java.util.Map;
import java.util.Set;public class Main { public static void main（String[] args） {
//实例化一个有5个Worker的Master
long start=System.currentTimeMillis（）;
Master m=new Master（new PlusWorker（）,5）;
for（int i=0;i<=1000;i++）{
m.submit（i）;
}
m.execute（）;
int re=0;
Map<String,Object> resultMap=m.getResultMap（）;
//只要有一个Worker计算完成，则开始计算
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;
}
//删除已经计算过的子结果
resultMap.remove（key）;
}
}
long time=System.currentTimeMillis（）-start;
System.out.println（"The result is:"+re+",time:"+time）;
}}