IT教程 ·

并发编程之Master-Worker模式

Java 添加、读取、删除Excel文档属性

我们晓得,单个线程盘算是串行的,只需等上一个使命终了以后,才实行下一个使命,所以实行效力是比较低的。

那末,假如用多线程实行使命,就能够在单元时候内实行更多的使命,而Master-Worker就是多线程并行盘算的一种完成体式格局。

它的头脑是,启动两个历程协同事情:Master和Worker历程。

Master担任使命的吸收和分派,Worker担任详细的子使命实行。每一个Worker实行完使命以后把效果返回给Master,最后由Master汇总效果。(实在也是一种分而治之的头脑,和forkjoin盘算框架有相似之处,参看:并行使命盘算框架forkjoin)

Master-Worker事情示意图以下:

并发编程之Master-Worker模式 IT教程 第1张

下面用Master-Worker完成盘算1-100的平方和,思绪以下:

  1. 定义一个Task类用于存储每一个使命的数据。
  2. Master生产牢固个数的Worker,把一切worker存放在workers变量(map)中,Master须要存储一切使命的行列workqueue(ConcurrentLinkedQueue)和一切子使命返回的效果集resultMap(ConcurrentHashMap)。
  3. 每一个Worker实行本身的子使命,然后把效果存放在resultMap中。
  4. Master汇总resultMap中的数据,然后返回给Client客户端。
  5. 为了扩大Worker的功用,用一个MyWorker继续Worker重写使命处置惩罚的详细要领。

Task类:

package com.thread.masterworker;
public class Task {
    private int id;
    private String name;
    private int num;

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getNum() {
        return num;
    }

    public void setNum(int num) {
        this.num = num;
    }
}

Master完成:

package com.thread.masterworker;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

public class Master {
    //一切使命的行列
    private ConcurrentLinkedQueue<Task> workerQueue = new ConcurrentLinkedQueue<Task>();

    //一切worker
    private HashMap<String,Thread> workers = new HashMap<String,Thread>();

    //同享变量,worker返回的效果
    private ConcurrentHashMap<String,Object> resultMap = new ConcurrentHashMap<String,Object>();

    //组织要领,初始化一切worker
    public Master(Worker worker,int workerCount){
        worker.setWorkerQueue(this.workerQueue);
        worker.setResultMap(this.resultMap);

        for (int i = 0; i < workerCount; i++) {
            Thread t = new Thread(worker);
            this.workers.put("worker-"+i,t);
        }
    }

    //使命的提交
    public void submit(Task task){
        this.workerQueue.add(task);
    }

    //实行使命
    public int execute(){
        for (Map.Entry<String, Thread> entry : workers.entrySet()) {
            entry.getValue().start();
        }

        //一向轮回,直到效果返回
        while (true){
            if(isComplete()){
                return getResult();
            }
        }

    }

    //推断是不是一切线程都已实行终了
    public boolean isComplete(){
        for (Map.Entry<String, Thread> entry : workers.entrySet()) {
            //只需有恣意一个线程没有终了,就返回false
            if(entry.getValue().getState() != Thread.State.TERMINATED){
                return false;
            }
        }
        return true;
    }

    //处置惩罚效果集返回终究效果
    public int getResult(){
        int res = 0;
        for (Map.Entry<String,Object> entry : resultMap.entrySet()) {
            res += (Integer) entry.getValue();
        }
        return res;
    }

}

父类Worker:

package com.thread.masterworker;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

public class Worker implements Runnable {

    private ConcurrentLinkedQueue<Task> workerQueue;

    private ConcurrentHashMap<String,Object> resultMap;

    public void setWorkerQueue(ConcurrentLinkedQueue<Task> workerQueue) {
        this.workerQueue = workerQueue;
    }

    public void setResultMap(ConcurrentHashMap<String, Object> resultMap) {
        this.resultMap = resultMap;
    }

    @Override
    public void run() {
        while(true){
            //从使命行列中掏出一个使命
            Task task = workerQueue.poll();
            if(task == null) break;
            //处置惩罚详细的使命
            Object res = doTask(task);
            //把每次处置惩罚的效果放到效果集内里,此处直接把num值作为效果
            resultMap.put(String.valueOf(task.getId()),res);
        }

    }

    public Object doTask(Task task) {
        return null;
    }
}

子类MyWorker继续父类Worker,重写doTask要领完成详细的逻辑:

package com.thread.masterworker;

public class MyWorker extends Worker {
    @Override
    public Object doTask(Task task) {
        //停息0.5秒,模仿使命处置惩罚
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //盘算数字的平方
        int num = task.getNum();
        return num * num;
    }
}

客户端Client:

package com.thread.masterworker;

import java.util.Random;

public class Client {
    public static void main(String[] args) {

        Master master = new Master(new MyWorker(), 10);

        //提交n个使命到使命行列里
        for (int i = 0; i < 100; i++) {
            Task task = new Task();
            task.setId(i);
            task.setName("使命"+i);
            task.setNum(i+1);
            master.submit(task);
        }

        //实行使命
        long start = System.currentTimeMillis();
        int res = master.execute();
        long time = System.currentTimeMillis() - start;
        System.out.println("效果:"+res+",耗时:"+time);
    }
}

以上,我们用10个线程去实行子使命,终究由Master做盘算乞降(1-100的平方和)。每一个线程停息500ms,盘算数字的平方值。

统共100个使命,分10个线程并行盘算,相当于每一个线程均分10个使命,一个使命的时候也许为500ms,故10个使命为5000ms,再加上盘算平方值的时候,故稍大于5000ms。效果以下,

效果:338350,耗时:5084

3D点云配准算法简述

参与评论