多线程设计模式（三）：Master-Worker模式

2023-07-17 19:51:16

Master-Worker模式是常用的并行模式之一，它的核心思想是，系统有两个进程协作工作：Master进程，负责接收和分配任务；Worker进程，负责处理子任务。当Worker进程将子任务处理完成后，结果返回给Master进程，由Master进程做归纳汇总，最后得到最终的结果。

一、什么是Master-Worker模式：

该模式的结构图：

多线程设计模式（三）：Master-Worker模式

结构图：

多线程设计模式（三）：Master-Worker模式

Worker：用于实际处理一个任务；

Master：任务的分配和最终结果的合成；

Main：启动程序，调度开启Master。

二、代码实现：

下面的是一个简易的Master-Worker框架实现。

（1）Master部分：

  1. package MasterWorker;
  2. import java.util.HashMap;
  3. import java.util.Map;
  4. import java.util.Queue;
  5. import java.util.concurrent.ConcurrentHashMap;
  6. import java.util.concurrent.ConcurrentLinkedQueue;
  7. public class Master {
  8. //任务队列
  9. protected Queue<Object> workQueue= new ConcurrentLinkedQueue<Object>();
  10. //Worker进程队列
  11. protected Map<String ,Thread> threadMap= new HashMap<String ,Thread>();
  12. //子任务处理结果集
  13. protected Map<String ,Object> resultMap= new ConcurrentHashMap<String, Object>();
  14. //是否所有的子任务都结束了
  15. public boolean isComplete(){
  16. for(Map.Entry<String , Thread> entry:threadMap.entrySet()){
  17. if(entry.getValue().getState()!=Thread.State.TERMINATED){
  18. return false;
  19. }
  20. }
  21. return true ;
  22. }
  23. //Master的构造，需要一个Worker进程逻辑，和需要Worker进程数量
  24. public Master(Worker worker,int countWorker){
  25. worker.setWorkQueue(workQueue);
  26. worker.setResultMap(resultMap);
  27. for(int i=0;i<countWorker;i++){
  28. threadMap.put(Integer.toString(i),  new Thread(worker, Integer.toString(i)));
  29. }
  30. }
  31. //提交一个任务
  32. public void submit(Object job){
  33. workQueue.add(job);
  34. }
  35. //返回子任务结果集
  36. public Map<String ,Object> getResultMap(){
  37. return resultMap;
  38. }
  39. //开始运行所有的Worker进程，进行处理
  40. public  void execute(){
  41. for(Map.Entry<String , Thread> entry:threadMap.entrySet()){
  42. entry.getValue().start();
  43. }
  44. }
  45. }

（2）Worker进程实现：

  1. package MasterWorker;
  2. import java.util.Map;
  3. import java.util.Queue;
  4. public class Worker  implements Runnable{
  5. //任务队列，用于取得子任务
  6. protected Queue<Object> workQueue;
  7. //子任务处理结果集
  8. protected Map<String ,Object> resultMap;
  9. public void setWorkQueue(Queue<Object> workQueue){
  10. this.workQueue= workQueue;
  11. }
  12. public void setResultMap(Map<String ,Object> resultMap){
  13. this.resultMap=resultMap;
  14. }
  15. //子任务处理的逻辑，在子类中实现具体逻辑
  16. public Object handle(Object input){
  17. return input;
  18. }
  19. @Override
  20. public void run() {
  21. while(true){
  22. //获取子任务
  23. Object input= workQueue.poll();
  24. if(input==null){
  25. break;
  26. }
  27. //处理子任务
  28. Object re = handle(input);
  29. resultMap.put(Integer.toString(input.hashCode()), re);
  30. }
  31. }
  32. }

（3）运用这个小框架计算1——100的立方和，PlusWorker的实现：

  1. package MasterWorker;
  2. public class PlusWorker extends Worker {
  3. @Override
  4. public Object handle(Object input) {
  5. Integer i =(Integer)input;
  6. return i*i*i;
  7. }
  8. }

（4）进行计算的Main函数：

  1. package MasterWorker;
  2. import java.util.Map;
  3. import java.util.Set;
  4. public class Main {
  5. /**
  6. * @param args
  7. */
  8. public static void main(String[] args) {
  9. //固定使用5个Worker，并指定Worker
  10. Master m = new Master(new PlusWorker(), 5);
  11. //提交100个子任务
  12. for(int i=0;i<100;i++){
  13. m.submit(i);
  14. }
  15. //开始计算
  16. m.execute();
  17. int re= 0;
  18. //保存最终结算结果
  19. Map<String ,Object> resultMap =m.getResultMap();
  20. //不需要等待所有Worker都执行完成，即可开始计算最终结果
  21. while(resultMap.size()>0 || !m.isComplete()){
  22. Set<String> keys = resultMap.keySet();
  23. String key =null;
  24. for(String k:keys){
  25. key=k;
  26. break;
  27. }
  28. Integer i =null;
  29. if(key!=null){
  30. i=(Integer)resultMap.get(key);
  31. }
  32. if(i!=null){
  33. //最终结果
  34. re+=i;
  35. }
  36. if(key!=null){
  37. //移除已经被计算过的项
  38. resultMap.remove(key);
  39. }
  40. }
  41. }
  42. }

三、总结：

Master-Worker模式是一种将串行任务并行化的方案，被分解的子任务在系统中可以被并行处理，同时，如果有需要，Master进程不需要等待所有子任务都完成计算，就可以根据已有的部分结果集计算最终结果集。

转：http://blog.csdn.net/lmdcszh/article/details/39698189

上一篇：在SQL中修改数据库名称


下一篇：web聊天相关知识