并发编程之Master-worker模式

淩亂°似流年 2021-11-11 21:12 240阅读 0赞

**Master-worker模式的简单实现**  
1、master

package com.ming.mw;
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.ConcurrentLinkedQueue;
    
    /**
     * @Auther: ex_chenhm
     * @Date: 2019/8/6 10:00
     * @Description:
     */
    public class Master {
        //存放任务队列
        private ConcurrentLinkedQueue<Task> taskQueue=new ConcurrentLinkedQueue();
        //存放worker对象
        private Map<String,Thread> workers=new HashMap<>();
        //存放结果
        private ConcurrentHashMap<String,Object> resultMap=new ConcurrentHashMap<>();
    
        public Master(Worker worker,int workerCount){
            worker.setTask(this.taskQueue);
            worker.setResultMap(this.resultMap);
            //初始化worker
            for(int i=0;i<workerCount;i++){
                workers.put("worker"+Integer.toString(i),new Thread(worker,"worker"+Integer.toString(i)));
            }
        }
    
        //提交任务
        public void submitTask(Task task){
            taskQueue.add(task);
        }
    
        //启动任务
        public void startTask(){
            for(Map.Entry<String,Thread> entry:workers.entrySet()){
                entry.getValue().start();
            }
        }
    
        //判断任务是否结束
        public boolean isComplete(){
            for(Map.Entry<String,Thread> entry:workers.entrySet()){
                if(entry.getValue().getState()!=Thread.State.TERMINATED){
                    return false;
                }
            }
            return true;
        }
    
        //返回结果集
        public ConcurrentHashMap<String,Object> getResultMap(){
            return resultMap;
        }
    }

2、worker

package com.ming.mw;
    
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.ConcurrentLinkedQueue;
    
    /**
     * @Auther: ex_chenhm
     * @Date: 2019/8/6 10:00
     * @Description:
     */
    public class Worker implements Runnable{
        //存放任务队列
        private ConcurrentLinkedQueue<Task> taskQueue=new ConcurrentLinkedQueue();
        //存放结果
        private ConcurrentHashMap<String,Object> resultMap=new ConcurrentHashMap<>();
    
        public void setTask(ConcurrentLinkedQueue<Task> taskQueue) {
            this.taskQueue=taskQueue;
        }
    
        public void setResultMap(ConcurrentHashMap<String, Object> resultMap) {
            this.resultMap=resultMap;
        }
    
        @Override
        public void run() {
            while (true){
                Task task = taskQueue.poll();
                if(task==null)break;
                //做业务
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                handleTask(task);
            }
        }
    
        protected void handleTask(Task task) {
            System.out.println(task.getId()+"年龄："+task.getAge());
            resultMap.put(task.getId(),task.getAge());
        }
    }

3、task

package com.ming.mw;
    
    /**
     * @Auther: ex_chenhm
     * @Date: 2019/8/6 10:00
     * @Description:
     */
    public class Task {
        private String id;
        private String name;
        private int age;
        private int sex;
    
        public Task(String id, int age) {
            this.id = id;
            this.age = age;
        }
    
        public String getId() {
            return id;
        }
    
        public void setId(String id) {
            this.id = id;
        }
    
        public String getName() {
            return name;
        }
    
        public void setName(String name) {
            this.name = name;
        }
    
        public int getAge() {
            return age;
        }
    
        public void setAge(int age) {
            this.age = age;
        }
    
        public int getSex() {
            return sex;
        }
    
        public void setSex(int sex) {
            this.sex = sex;
        }
    }

4、不同实现的myworker

package com.ming.mw;
    
    /**
     * @Auther: ex_chenhm
     * @Date: 2019/8/6 11:09
     * @Description:
     */
    public class MyWorker extends Worker {
    
        @Override
        protected void handleTask(Task task) {
            //不同实现的业务方法
            System.out.println("不同实现的业务方法");
            super.handleTask(task);
        }
    }

5、测试类

package com.ming.mw;
    
    import java.util.Map;
    
    /**
     * @Auther: ex_chenhm
     * @Date: 2019/8/6 10:31
     * @Description:
     */
    public class TestMain {
        public static void main(String[] str) {
            Master master = new Master(new MyWorker(), 20);
            //创建100个任务
            for (int i = 0; i < 100; i++) {
                master.submitTask(new Task(Integer.toString(i), (int) (Math.random() * 100)));
            }
            master.startTask();
            long startTime=System.currentTimeMillis();
            while (true) {
                if(master.isComplete()){
                    int result = 0;
                    for (Map.Entry<String, Object> entry : master.getResultMap().entrySet()) {
                        result += (int) entry.getValue();
                    }
                    System.out.println("执行结果："+result+" ,执行耗时："+(System.currentTimeMillis()-startTime));
                    break;
                }
            }
        }
    }