pendingjob.java

服务器模块主要负责分配计算任务

package primecruncher;
import java.util.*;

public class PendingJob{
    private String jobID;
    private Dispatcher dispatcher;                 //发布者
    private ResultListener resultListener;
    private HashSet jobSet=new HashSet();
/*****************************************************************/
    PendingJob(String jobID,Dispatcher dispatcher,ResultListener resultListener){
        this.jobID=jobID;
        this.dispatcher=dispatcher;
        this.resultListener=resultListener;
    }
/*****************************************************************/
    void addJob(DispatcherJob job){
        JobResult r=new JobResult(job);
        jobSet.add(r);
        System.out.println("Added new job to jobset");
    }
/*****************************************************************/
    String getID(){
        return jobID;
    }
/*****************************************************************/
    ResultListener getResultListener(){
        return resultListener;
    }
/*****************************************************************/
    void getResult(DispatcherJob job, Map resultMap){
        System.out.println("Got a result");
        HashSet jobCopy = null;
        synchronized(jobSet){
            jobCopy=(HashSet)jobSet.clone();
        }
        Iterator it=jobCopy.iterator();
        while(it.hasNext()){
            JobResult rs=(JobResult)it.next();
            if(rs.dispatcherJob.equals(job)){     //将服务端返回的结果集中的任务号与jobset中的任务依次作对比,
             rs.setResult(resultMap);            //看jobset中是否存在这样的任务号
                break;
            }
        }
        areWeReady();
    }
/*****************************************************************/
    private void areWeReady(){
        HashSet s;
        synchronized(jobSet){
            s = (HashSet)jobSet.clone();
        }
        Iterator it=s.iterator();
        while(it.hasNext()){    //判断是否每一个服务端都结果返回,如果有一个没返回,则标记为false
            JobResult r=(JobResult)it.next();
            if(!r.isRead()){
                return;
            }
        }
        JobResult[] results=(JobResult[])s.toArray(new JobResult[s.size()]);   //创建JobResult类型的数组
        Arrays.sort(results);
        Map[] sortedResults=new Map[results.length];
        for(int i=0;i<sortedResults.length;i++){
            sortedResults[i]=results[i].getResult();
        }
        dispatcher.jobComplete(this,sortedResults);                     //调用Dispatcher类的jobComplete()
    }
/*****************************************************************/
    public boolean equals(Object o){
        if(o instanceof PendingJob){
            PendingJob p=(PendingJob)o;
            return jobID.equals(p.jobID);    //将任务号作对比
        }
        return false;
    }
/*****************************************************************/
    public int hashCode(){
        return jobID.hashCode();
    }
/*****************************************************************/







    class JobResult implements Comparable{
        DispatcherJob dispatcherJob;
        Map result;
        boolean ready = false;
/*****************************************************************/
        public int compareTo(Object o){
            if(o instanceof JobResult){
                JobResult r=(JobResult)o;
                return dispatcherJob.compareTo(r.dispatcherJob);
            }
            return 0;
        }
/*****************************************************************/
        JobResult(DispatcherJob dispatcherJob){
            this.dispatcherJob=dispatcherJob;
        }
/*****************************************************************/
        boolean isRead(){
            return ready;
        }
/*****************************************************************/
        void setResult(Map result){
            this.result=result;
            ready=true;
        }
/*****************************************************************/
        Map getResult(){
            return result;
        }
/*****************************************************************/
        public boolean equals(Object o){
            if(o instanceof JobResult){
                JobResult jr=(JobResult)o;
                return dispatcherJob.equals(jr.dispatcherJob);
            }
            return false;
        }
/*****************************************************************/
        public int hashCode(){
            return dispatcherJob.hashCode();
        }
    }
}