多线程调优-Master-Worker模式

简介

Master-Worker是常见的并行设计模式。
核心思想是：系统中有两个进程协议工作。Master进程负责接收和分配任务，Worker负责处理子任务。当worker进程将子任务处理完毕之后，将结果返回给Master进程，由Master进行归纳和汇总。
优势：可以将大任务分解成若干小任务，并发执行，从而提高系统性能。对于客户端（请求发起者）来说，任务一旦提交，Master进程就会立刻分配任务并立刻返回，并不会等待系统处理完再返回，其处理过程是异步的。

架构：Master是主要进程，维护这一个Worker进程队列，子任务队列，子结果集。Worker进程不断的从人物队列中获取要处理的人物，并将子进程的处理结果添加到结果集中。
Master主要负责任务的分配和最终结果的合并。Worker用于实际处理一个任务。客户端进程负责启动系统，调度开启Master

计算10之内整数的立方并求和

客户端

package com.cyan.base.MasterWorker;

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

/**
 * @Auther: Administrator
 * @Date: 2018/12/20 18:22
 * @Description:
 */
public class Client {
    public static void main(String[] args) {
        //创建Master对象，指定需要创建的Worker进程和数量
        Master master = new Master(new PlusWorker(), 5);
        for (int i = 0; i < 10; i++) {
            //提交任务给Master任务队列
            master.submit(i);
        }
        //执行所有的Worker进程
        master.execute();
        //获取最终的执行结果
        Integer result = master.getResult();
        System.out.println("结果是："+result);
    }
}

Master

package com.cyan.base.MasterWorker;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedDeque;
import static java.lang.Thread.State.TERMINATED;
/**
 * @Auther: Administrator
 * @Date: 2018/12/20 17:21
 * @Description:
 */
public class Master {

    //任务队列
    protected Queue<Object> taskQueue = new ConcurrentLinkedDeque<>();
    //worker进程队列
    protected Map<String,Thread> workers = new HashMap<>();
    //结果集队列
    protected Map<String, Object> result = new HashMap<>();

    //构造器创建worker进程，需要 制定worker进程的任务列表和结果集，并创建制定数量的worker
    public Master(Worker worker,int workerCount) {
        worker.setTaskQueue(taskQueue);
        worker.setResultSet(result);
        for (int i = 0; i < workerCount; i++) {
            workers.put(String.valueOf(i), new Thread(worker, "worker " + i + "   "));
        }
    }

    //是否所有的进程都执行完毕
    public boolean isComplete(){
        for (Map.Entry<String, Thread> entry : workers.entrySet()) {
            if (entry.getValue().getState() != TERMINATED) {
                return false;
            }
        }
        return true;
    }

    //提交任务
    public void submit(Object obj){
        Boolean aBoolean = Optional.ofNullable(obj)
                .map(task -> taskQueue.add(task)).get();
        //        taskQueue.offer(obj);
    }

    //返回结果集
    public Integer getResult(){
        Integer sum = 0;
        //此处判断，  结果集大于0  或者 worker线程没有全部执行完时，一直进行累加计算
        while(result.size()>0 || !this.isComplete()){
            Set<String> keys = result.keySet();
            String key = null;
            for (String k : keys) {
                key = k;
                break;
            }
            Integer re = null;
            //获取当前任务的结果
            if (key!=null) {
                re = (Integer) result.get(key);
            }
            //对结果进行处理
            if (re!=null) {
                sum += re;
            }
            //将当前子任务结果从结果集移除
            if (key!=null) {
                result.remove(key);
            }
        }
        return sum;
    }
    //运行所有的worker
    public void execute(){
//        for (Map.Entry<String, Thread> entry : workers.entrySet()) {
//            entry.getValue().start();
//        }
        workers.values().forEach(
                worker -> worker.start()
        );
    }

}

Worker

package com.cyan.base.MasterWorker;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Queue;

/**
 * @Auther: Administrator
 * @Date: 2018/12/20 18:05
 * @Description: worker进程是多线程异步执行
 */
public class Worker implements Runnable{
    //任务列表和结果集也可以通过构造器进行初始化

    //任务列表   此处不做初始化，为其提供set方法进行初始化
    protected Queue<Object> taskQueue;
    //子任务结果集
    protected Map<String, Object> resultSet = new HashMap<>();

    public void setTaskQueue(Queue<Object> taskQueue) {
        this.taskQueue = taskQueue;
    }

    public void setResultSet(Map<String, Object> resultSet) {
        this.resultSet = resultSet;
    }

    //业务处理流程
    public Object handler(Object obj){
        System.out.println("Worker " + Thread.currentThread().getName() + "   obj" + obj);
        return obj;
    }

    @Override
    public void run() {
        while (true) {
            Object object = taskQueue.poll();
            if (Objects.isNull(object)) {
                return;
            }
            Object result = handler(object);
            resultSet.put(String.valueOf(object.hashCode()),result);
        }
    }

}

PlusWorker

package com.cyan.base.MasterWorker;

/**
 * @Auther: Administrator
 * @Date: 2018/12/20 18:20
 * @Description:
 */
public class PlusWorker extends Worker {

    @Override
    public Object handler(Object obj) {
        Integer i = (Integer) obj;
        System.out.println("Thread : " + Thread.currentThread().getName() + "  handler :" + obj);
        return i * i * i;
    }
}

多线程调优-Master-Worker模式

简介

计算10之内整数的立方并求和

客户端

Master

PlusWorker

相关推荐