master-worker核心思想

master-worker模式優點

master-worker模式結構

master-worker模式主要參與者

master-worker樣例程式碼結構

程式碼

1.master.java

package test;

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

public
 
 class Master {


    //任務佇列
    protected  Queue<Object> workQueue = new ConcurrentLinkedQueue<Object>();

    //worker程序佇列
    protected Map<String,Thread>threadMap = new HashMap<String,Thread>();


    //子任務結果處理集
    protected Map<String,Object>resultMap = new ConcurrentHashMap<String,Object>();


    public
 
 Map<String, Object> getResultMap() {
        return resultMap;
    }


    public Master(Worker worker ,int countWorker){
        worker.setWorkQueue(workQueue);
        worker.setResultMap(resultMap);
        for(int i=0;i<countWorker;i++){
            threadMap.put(Integer.toString(i), new
 
 Thread(worker,Integer.toString(i)));
        }
    }

    public void submit(Object job){
        workQueue.add(job);
    }

    public void execute(){
        for(Map.Entry<String, Thread> entry:threadMap.entrySet()){
            entry.getValue().start();
        }
    }

    public boolean isComplete(){
        for(Map.Entry<String, Thread> entry:threadMap.entrySet()){
            if(entry.getValue().getState()!=Thread.State.TERMINATED){
                return false;
            }
        }
        return true;
    }


}

2.Worker.java

package test;

import java.util.Map;
import java.util.Queue;

public class Worker implements Runnable {

    // 任務佇列，用於取得子任務
    protected Queue<Object> workQueue;

    // 子任務結果處理集
    protected Map<String, Object> resultMap;

    public void setWorkQueue(Queue<Object> workQueue) {
        this.workQueue = workQueue;
    }

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

    public Object handle(Object input) {
        return input;
    }

    @Override
    public void run() {

        while (true) {
            Object input = workQueue.poll();
            if( input == null){
                break;
            }
            //處理子任務
            Object re = handle(input);
            int k =input.hashCode();
            //將處理結果寫入結果集
            resultMap.put(Integer.toString(input.hashCode()), re);
        }
    }

}

3.PlusWorker

package test;

public class PlusWorker extends Worker{


    public Object handle(Object input){
        Integer i =(Integer)input;
        return i*i*i;
    }
}

4.TestMain

package test;

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

public class TestMain {

    public static void main(String[] args) {

        Master m = new Master(new PlusWorker(), 5);
        for (int i = 0; i < 100; i++) {
            m.submit(i);
        }
        m.execute();
        int re = 0;
        Map<String, Object> resultMap = m.getResultMap();
        while (resultMap.size() > 0 || !m.isComplete()) {
            Set<String> keys = resultMap.keySet();
            String key = null;
            for (String k : keys) {
                key = k;
                break;
            }
            Integer i = null;
            if (key != null) {
                i = (Integer) resultMap.get(key);
            }
            if (i != null) {
                re = re + i;
            }
            if (key != null) {
                System.out.println("removekey:"+key);
                resultMap.remove(key);
            }
        }
        System.out.println("re" + re);
    }

}