多线程学习之路-学习master-worker设计模式

前端时间主要精力在Java的迁移，就是把,以前项目的.net的代码迁移成Java代码，先完成的差很少了，主要考虑服务的优化，想起了Java的中master-worker模式，这个比较适用于多个计算的同时进行的优化，在网上找了一些学习的视频，作了下面的一个练习，这个练习master-worker的练习的例子，不推荐实际项目这么写；

跟以前同样先写一下思路；

首先申明Master和Worder两个工具类，先说一下Master的实现思路

1.Master是负责调度的一个工具类，里面确定应该有装载任务的的集合，这里使用了ConcurrentLinkedQueue;

2.Master是还须要负责给worker分配任务，这里使用了HashMap来装载worker对象

3.Master还须要负责对每个worker完成的结果汇总，由于每一个worker都会操做这个对象，因此定义了ConcurrentHashMap；

4.在Master的构造函数中把worker对象添加Master的引用，用于任务的领取和执行结果的提交；

5.后面分别是启动方法、提交方法、收集执行结果的方法，详细能够看一下代码；

package MasterWorkerTest;

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

public class Master {

    // 1.应该以一个橙装任务的集合
    private ConcurrentLinkedQueue<Task> workQueue = new ConcurrentLinkedQueue<Task>();

    // 2. 使用hashmap承装全部的worker对象
    private HashMap<String, Thread> workers = new HashMap<String, Thread>();

    // 3.使用一个容器承装每个worker并不是执行任务的结果集
    private ConcurrentHashMap<String, Object> resultMap = new ConcurrentHashMap<>();

    // 4.构造方法
    public Master(Worker worker, int workerCount) {

        // 每一个worker对象都须要有master的引用workerQueue用于任务的领取，resultMap用于任务的提交
        worker.setWorkerQueue(this.workQueue);
        worker.setResultMap(this.resultMap);

        for (int i = 0; i < workerCount; i++) {
            // key表示每一个worker的名字，value表示线程执行对象
            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<String, Thread> me : workers.entrySet()) {
            me.getValue().start();
        }
    }

    // 7. 判断是否执行完
    public boolean isComplete() {
        for (Map.Entry<String, Thread> me : workers.entrySet()) {
            if (me.getValue().getState() != Thread.State.TERMINATED) {
                return false;
            }
        }
        return true;
    }

    // 返回结果集数
    public int getReult() {
        int ret = 0;
        for (Map.Entry<String, Object> me : resultMap.entrySet()) {
            ret += (Integer) me.getValue();
        }

        return ret;
    }

}

下面的worker类

worker类其实就是实现了一个Runable接口，模拟业务逻辑操做；

package MasterWorkerTest;


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

public class Worker implements Runnable {

    private ConcurrentLinkedQueue<Task> workQueue;
    private ConcurrentHashMap<String, Object> resultMap;

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

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

    @Override
    public void run() {
        // TODO Auto-generated method stub
        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) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

        return outPut;
    }

}

下面是mian类，用于测试刚才写的代码

 

package MasterWorkerTest;

import java.util.Random;

public class Main {

    public static void main(String[] args) {

        Random r = new Random();
        Master master = new Master(new Worker(), 10);
        for (int i = 1; i < 101; i++) {
            Task task = new Task();
            task.setId(i);
            task.setName("任务:" + i);
            task.setPrice(r.nextInt(1000));
            master.submit(task);
        }

        master.execute();

        long start = System.currentTimeMillis();
        while (true) {
            if (master.isComplete()) {
                long end = System.currentTimeMillis() - start;
                int ret = master.getReult();
                System.out.println("最终结果:" + ret + " 耗时:" + end);
                break;
            }

        }
    }

}

漏了一个实体类，这个是测试的返回实体，如今项目建议写成泛型；

package MasterWorkerTest;

public class Task {
    private int id;
    private String name;
    private int price;

    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 getPrice() {
        return price;
    }

    public void setPrice(int price) {
        this.price = price;
    }
}

 测试的实例到这里就写完了，代码直接拷贝运行，接下来就要考虑接入到实际的项目中去了，等之后总结了这个模式的优势和缺点之后再来跟新，代码是根据视频教程写的;欢迎你们转载和交流；