Java多线程编程中Master-Worker模式的详解

Java多线程编程中，常用的多线程设计模式包括：Future模式、Master-Worker模式、Guarded Suspeionsion模式、不变模式和生产者-消费者模式等。这篇文章主要讲述Master-Worker模式，关于其他多线程设计模式的地址如下：

关于其他多线程设计模式的地址如下：

关于Future模式的详解： Java多线程编程中Future模式的详解

关于Guarded Suspeionsion模式的详解： Java多线程编程中Guarded Suspeionsion模式的详解

关于不变模式的详解： Java多线程编程中不变模式的详解

关于生产者-消费者模式的详解：生产者-消费者模式Java详解

1 Master-Worker模式核心思想

Master-Worker模式是常用的并行模式之一，它的核心思想是：系统由两类进程协同工作，即Master进程和Worker进程，Master负责接收和分配任务，Wroker负责处理子任务。当各个Worker进程将子任务处理完成后，将结果返回给Master进程，由Master进程进行汇总，从而得到最终的结果，其具体处理过程如下图所示。



Master进程为主要进程，它维护一个Worker进程队列、子任务队列和子结果集。Worker进程队列中的Worker进程不停从任务队列中提取要处理的子任务，并将结果写入结果集。

2 Master-Worker模式的代码实现

Worker类的Java实现

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 = workQueue.poll();
if(input == null) break;
//处理子任务
Object re = handle(input);
//将处理结果写入结果集
resultMap.put(Integer.toString(input.hashCode()), re);
}
}
}

Master类的Java实现

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 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 boolean isComplete() {
for(Map.Entry<String, Thread> entry : threadMap.entrySet()) {
if(entry.getValue().getState() != Thread.State.TERMINATED)
//存在为完成的线程
return false;
}
return true;
}

//提交一个子任务
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();
}
}
}

3 Master-Worker模式的应用实例

利用Master-Worker模式实现计算立方和的应用。计算1^3+2^3+3^3+…+100^3。

这个计算任务被划分成100个子任务，每个任务仅仅用于计算单独的立方和。

Worker的子类

public class PlusWorker extends Worker { //求立方和
@Override
public Object handle(Object input) {
int i = (Integer)input;
return i * i * i;
}
}

运行
运行的调用函数如下。在主函数中首先通过Master类创建4个Worker工作进程和Worker工作实例PlusWorker。在提交了100个子任务后，边开始子任务的计算。这些子任务中由这4个进程共同完成。Master不用等待所有Worker计算完成才开始汇总，而是子任务在计算的过程中，Master就开始汇总了。

import java.util.Map;
import java.util.Set;

public class Application {
public static void main(String[] args) {
//固定使用4个Workde
Master master = new Master(new PlusWorker(), 4);
for(int i=1; i<=100; i++) //提交100个子任务
master.submit(i);
master.execute(); //开始计算

Map<String, Object> resultMap = master.getResultMap();

int re = 0;  //最终计算结果保存在此
//不需要等待所有Worker都执行完即可
while(true) {
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); //移除已被计算过的项目
if(master.isComplete() && resultMap.size()==0)
break;
}
System.out.println(re);
}
}

运行结果如下：



本文完。转载请注明出处。

参考文献

葛一鸣，Java程序性能优化．清华大学出版社．