Giraph源码分析（一） —— 启动ZooKeeper服务

    @Override  
    public void setup(Context context)  
      throws IOException, InterruptedException {  
      // Execute all Giraph-related role(s) assigned to this compute node.  
      // Roles can include "master," "worker," "zookeeper," or . . . ?  
      graphTaskManager = new GraphTaskManager<I, V, E, M>(context);  
      graphTaskManager.setup(  
        DistributedCache.getLocalCacheArchives(context.getConfiguration()));  
    }  

    @Override  
      public void run(Context context) throws IOException, InterruptedException {  
        // Notify the master quicker if there is worker failure rather than  
        // waiting for ZooKeeper to timeout and delete the ephemeral znodes  
        try {  
          setup(context);  
          while (context.nextKeyValue()) {  
            graphTaskManager.execute();  
          }  
          cleanup(context);  
          // Checkstyle exception due to needing to dump ZooKeeper failure  
        } catch (RuntimeException e) {  
          graphTaskManager.zooKeeperCleanup();  
          graphTaskManager.workerFailureCleanup();  
        }  
      }  

    /** 
      * Called by owner of this GraphTaskManager on each compute node 
      * @param zkPathList the path to the ZK jars we need to run the job 
      */  
     public void setup(Path[] zkPathList) throws IOException, InterruptedException {  
       context.setStatus("setup: Initializing Zookeeper services.");  
       locateZookeeperClasspath(zkPathList);  
       serverPortList = conf.getZookeeperList();  
       if (serverPortList == null && startZooKeeperManager()) {  
         return; // ZK connect/startup failed  
       }  
       if (zkManager != null && zkManager.runsZooKeeper()) {  
           LOG.info("setup: Chosen to run ZooKeeper...");  
       }  
       context.setStatus("setup: Connected to Zookeeper service " +serverPortList);  
       this.graphFunctions = determineGraphFunctions(conf, zkManager);  
       instantiateBspService(serverPortList, sessionMsecTimeout);  
     }  

    /** 
      * Instantiate and configure ZooKeeperManager for this job. This will 
      * result in a Giraph-owned Zookeeper instance, a connection to an 
      * existing quorum as specified in the job configuration, or task failure 
      * @return true if this task should terminate 
      */  
     private boolean startZooKeeperManager()  
       throws IOException, InterruptedException {  
       zkManager = new ZooKeeperManager(context, conf);  
       context.setStatus("setup: Setting up Zookeeper manager.");  
       zkManager.setup();  
       if (zkManager.computationDone()) {  
         done = true;  
         return true;  
       }  
       zkManager.onlineZooKeeperServers();  
       serverPortList = zkManager.getZooKeeperServerPortString();  
       return false;  
     }  

    /** 
       * Task 0 will call this to create the ZooKeeper server list.  The result is 
       * a file that describes the ZooKeeper servers through the filename. 
       */  
      private void createZooKeeperServerList() throws IOException,  
          InterruptedException {  
        Map<String, Integer> hostnameTaskMap = Maps.newTreeMap();  
        while (true) {  
          FileStatus [] fileStatusArray = fs.listStatus(taskDirectory);  
          hostnameTaskMap.clear();  
          if (fileStatusArray.length > 0) {  
            for (FileStatus fileStatus : fileStatusArray) {    
              String[] hostnameTaskArray =  
                  fileStatus.getPath().getName().split(HOSTNAME_TASK_SEPARATOR);  
         
              if (!hostnameTaskMap.containsKey(hostnameTaskArray[0])) {  
                hostnameTaskMap.put(hostnameTaskArray[0],  
                    new Integer(hostnameTaskArray[1]));  
              }  
            }  
            if (hostnameTaskMap.size() >= serverCount) {  
              break;  
            }  
            Thread.sleep(pollMsecs);  
          }  
        }  
      }  

    private static GraphFunctions determineGraphFunctions(  
         ImmutableClassesGiraphConfiguration conf,  
         ZooKeeperManager zkManager) {  
       // What functions should this mapper do?  
       if (!splitMasterWorker) {  
         if ((zkManager != null) && zkManager.runsZooKeeper()) {  
           functions = GraphFunctions.ALL;  
         } else {  
           functions = GraphFunctions.ALL_EXCEPT_ZOOKEEPER;  
         }  
       } else {  
         if (zkAlreadyProvided) {  
           int masterCount = conf.getZooKeeperServerCount();  
           if (taskPartition < masterCount) {  
             functions = GraphFunctions.MASTER_ONLY;  
           } else {  
             functions = GraphFunctions.WORKER_ONLY;  
           }  
         } else {  
           if ((zkManager != null) && zkManager.runsZooKeeper()) {  
             functions = GraphFunctions.MASTER_ZOOKEEPER_ONLY;  
           } else {  
             functions = GraphFunctions.WORKER_ONLY;  
           }  
         }  
       }  
       return functions;  
     }