并发编程中级篇二----并行设计模式----Master-Wroker模式

RebeccaZhong 发布于2019-08-16 10:46 / 3504人阅读

摘要：模式是常用的并行计算模式，它的核心思想是系统是由两类进程协助工作。负责接收和分配任务，负责处理子任务。当各个子进程处理完成后，会返回结果给，由做归纳和总结。其好处是能将一个大任务分解成若干个小任务，并行执行，从而提高系统的吞吐量。

Master-worker模式是常用的并行计算模式，它的核心思想是系统是由两类进程协助工作。Master负责接收和分配任务，worker负责处理子任务。当各个worker子进程处理完成后，会返回结果给master，由master做归纳和总结。其好处是能将一个大任务分解成若干个小任务，并行执行，从而提高系统的吞吐量。

1.定义一个要执行的任务类

    public class Task{
        
        private int id;
        
        private int price;
        
        //getter/setter()省略
    }

2.定义Master负责收发任务

    //注意worker线程的数量 根据机器的性能来添加
    //Runtime.getRuntime().availableProcessors()
    public class Master{
        
        //1.定义一个盛放任务的容器
        private ConcurrentLinkedQueue workQueue = 
        new ConcurrentLinkedQueue();
                
        //2.需要一个盛放worker的集合
        private HashMap workers = 
        new HashMap();
        
        //3.需要一个盛放每一个worker执行任务的结果集合
        private ConcurrentHashMap resultMap = 
        new ConcurrentHashMap();
        
        //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 me : workers.entrySet()){
                me.getValue().start();
            }
        }
        
        //7判断运行是否结束的方法
        public boolean isComplete(){
            for(Map.Entry me : workers.entrySet()){
                if(me.getValue().getState() != Thread.State.TERMINATED){
                    return false;
                }
            }        
            return true;
        }
        
        //8 计算结果方法
        public int getResult() {
            int priceResult = 0;
            for(Map.Entry me : resultMap.entrySet()){
                priceResult += (Integer)me.getValue();
            }
            return priceResult;
        }
    }

3.定义一个worker处理子任务

    public class Worker implements Runnable{
        
        private ConcurrentLinkedQueue workQueue;
        private ConcurrentHashMap resultMap;
        
        public void setWorkQueue(ConcurrentLinkedQueue workQueue) {
            this.workQueue = workQueue;
        }
    
        public void setResultMap(ConcurrentHashMap 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.定义一个测试类

    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;
                }
            }        
         }
    }

结果如下

GPU云服务器云服务器 java设计模式并发模式编程 python编程模式 bs模式编程语言

文章版权归作者所有，未经允许请勿转载,若此文章存在违规行为，您可以联系管理员删除。

转载请注明本文地址：https://www.ucloud.cn/yun/70413.html

并发编程中级篇二----并行设计模式----Future模式

摘要：模式类似于用户提交商品订单，下单成功以后后台异步的执行耗时的业务在包中接口是线程模式的实现，可以来进行异步计算。 Future模式类似于用户提交商品订单，下单成功以后后台异步的执行耗时的业务在java.util.concurrent包中.Future接口是Java线程Future模式的实现，可以来进行异步计算。 showImg(https://segmentfault.com/img/...

lx1036 2019-08-16 10:46 评论0 收藏0
并发编程中级篇三----并行设计模式----生产者-消费者模式

摘要：生产者消费者模式是一个经典的多线程设计模式，它为多线程的协作提供了良好的解决方案。生产者消费者模式中的内存缓冲区的主要功能是数据在多线程间的共享。生产者-消费者模式是一个经典的多线程设计模式，它为多线程的协作提供了良好的解决方案。在生产者-消费者模式中，通常有两类线程，即若干个生产者线程和若干个消费者线程。生产者线程负责提交用户请求，消费者线程负责处理用户请求。生产者和消费者之间通过...

Aldous 2019-08-16 10:47 评论0 收藏0
学习Java必读的10本书籍

摘要：学习编程的本最佳书籍这些书涵盖了各个领域，包括核心基础知识，集合框架，多线程和并发，内部和性能调优，设计模式等。擅长解释错误及错误的原因以及如何解决简而言之，这是学习中并发和多线程的最佳书籍之一。 showImg(https://segmentfault.com/img/remote/1460000018913016); 来源 | 愿码(ChainDesk.CN)内容编辑愿码Slo...

masturbator 2019-08-16 16:46 评论0 收藏0
你和阿里资深架构师之间，差的不仅仅是年龄（进阶必看）

摘要：导读阅读本文需要有足够的时间，笔者会由浅到深带你一步一步了解一个资深架构师所要掌握的各类知识点，你也可以按照文章中所列的知识体系对比自身，对自己进行查漏补缺，觉得本文对你有帮助的话，可以点赞关注一下。目录一基础篇二进阶篇三高级篇四架构篇五扩导读：阅读本文需要有足够的时间，笔者会由浅到深带你一步一步了解一个资深架构师所要掌握的各类知识点，你也可以按照文章中所列的知识体系对比自身，对自己...

huaixiaoz 2019-08-15 15:24 评论0 收藏0
2021年10月6日软考中级笔记---软件设计师（B站视频笔记）

摘要：死锁问题可以预防，可以解决死锁年月日数据库的完整性约束实体完整性约束定义主键。主键不能为空，不能重复参照完整性约束外键。系统需要进行多次连接，才能进行查询操作，是的系统效率大大下降。 ...

superPershing 2021-10-08 10:05 评论0 收藏0