override def preStart() {  
    logInfo("Starting Spark master at " + masterUrl)  
    ...  
    //persistenceEngine是持久化Worker，Driver和Application信息的，这样在Master重新启动时不会影响  
    //已经提交Job的运行  
    persistenceEngine = RECOVERY_MODE match {  
      case "ZOOKEEPER" =>  
        logInfo("Persisting recovery state to ZooKeeper")  
        new ZooKeeperPersistenceEngine(SerializationExtension(context.system), conf)  
      case "FILESYSTEM" =>  
        logInfo("Persisting recovery state to directory: " + RECOVERY_DIR)  
        new FileSystemPersistenceEngine(RECOVERY_DIR, SerializationExtension(context.system))  
      case _ =>  
        new BlackHolePersistenceEngine()  
    }  
    //leaderElectionAgent负责Leader的选取。  
    leaderElectionAgent = RECOVERY_MODE match {  
        case "ZOOKEEPER" =>  
          context.actorOf(Props(classOf[ZooKeeperLeaderElectionAgent], self, masterUrl, conf))  
        case _ => // 仅仅有一个Master的集群，那么当前的Master就是Active的  
          context.actorOf(Props(classOf[MonarchyLeaderAgent], self))  
      }  
  } 

val RECOVERY_MODE = conf.get("spark.deploy.recoveryMode", "NONE")  

private[spark] class BlackHolePersistenceEngine extends PersistenceEngine {  
  override def addApplication(app: ApplicationInfo) {}  
  override def removeApplication(app: ApplicationInfo) {}  
  override def addWorker(worker: WorkerInfo) {}  
  override def removeWorker(worker: WorkerInfo) {}  
  override def addDriver(driver: DriverInfo) {}  
  override def removeDriver(driver: DriverInfo) {}  
  
  override def readPersistedData() = (Nil, Nil, Nil)  
}  

class ZooKeeperPersistenceEngine(serialization: Serialization, conf: SparkConf)  
  extends PersistenceEngine  
  with Logging  
{  
  val WORKING_DIR = conf.get("spark.deploy.zookeeper.dir", "/spark") + "/master_status"  
  val zk: CuratorFramework = SparkCuratorUtil.newClient(conf)  
  
  SparkCuratorUtil.mkdir(zk, WORKING_DIR)  
  // 将app的信息序列化到文件WORKING_DIR/app_{app.id}中  
  override def addApplication(app: ApplicationInfo) {  
    serializeIntoFile(WORKING_DIR + "/app_" + app.id, app)  
  }  
  
  override def removeApplication(app: ApplicationInfo) {  
    zk.delete().forPath(WORKING_DIR + "/app_" + app.id)  
  }  

spark.deploy.recoveryMode=ZOOKEEPER  
spark.deploy.zookeeper.url=zk_server_1:2181,zk_server_2:2181  
spark.deploy.zookeeper.dir=/dir     
// OR 通过一下方式设置  
export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER "  
export SPARK_DAEMON_JAVA_OPTS="${SPARK_DAEMON_JAVA_OPTS} 
-Dspark.deploy.zookeeper.url=zk_server1:2181,zk_server_2:2181"

/** 
     * Create a new client 
     * 
     * 
     * @param connectString list of servers to connect to 
     * @param sessionTimeoutMs session timeout 
     * @param connectionTimeoutMs connection timeout 
     * @param retryPolicy retry policy to use 
     * @return client 
     */  
    public static CuratorFramework newClient
(String connectString, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy)  
    {  
        return builder().  
            connectString(connectString).  
            sessionTimeoutMs(sessionTimeoutMs).  
            connectionTimeoutMs(connectionTimeoutMs).  
            retryPolicy(retryPolicy).  
            build();  
    }  

/** 
* LeaderLatchListener can be used to be notified asynchronously about when the state of the LeaderLatch has changed. 
* 
* Note that just because you are in the middle of one of these method calls, it does not necessarily mean that 
* hasLeadership() is the corresponding true/false value. It is possible for the state to change behind the scenes 
* before these methods get called. The contract is that if that happens, you should see another call to the other 
* method pretty quickly. 
*/  
public interface LeaderLatchListener  
{  
  /** 
* This is called when the LeaderLatch's state goes from hasLeadership = false to hasLeadership = true. 
* 
* Note that it is possible that by the time this method call happens, hasLeadership has fallen back to false. If 
* this occurs, you can expect {@link #notLeader()} to also be called. 
*/  
  public void isLeader();  
  
  /** 
* This is called when the LeaderLatch's state goes from hasLeadership = true to hasLeadership = false. 
* 
* Note that it is possible that by the time this method call happens, hasLeadership has become true. If 
* this occurs, you can expect {@link #isLeader()} to also be called. 
*/  
  public void notLeader();  
}  

public class LeaderLatch implements Closeable  
{  
    private final Logger log = LoggerFactory.getLogger(getClass());  
    private final CuratorFramework client;  
    private final String latchPath;  
    private final String id;  
    private final AtomicReference<State> state = new AtomicReference<State>(State.LATENT);  
    private final AtomicBoolean hasLeadership = new AtomicBoolean(false);  
    private final AtomicReference<String> ourPath = new AtomicReference<String>();  
    private final ListenerContainer<LeaderLatchListener> listeners = new ListenerContainer<LeaderLatchListener>();  
    private final CloseMode closeMode;  
    private final AtomicReference<Future<?>> startTask = new AtomicReference<Future<?>>();  
.  
.  
.  
    /** 
     * Attaches a listener to this LeaderLatch 
     * <p/> 
     * Attaching the same listener multiple times is a noop from the second time on. 
     * <p/> 
     * All methods for the listener are run using the provided Executor.  It is common to pass in a single-threaded 
     * executor so that you can be certain that listener methods are called in sequence, but if you are fine with 
     * them being called out of order you are welcome to use multiple threads. 
     * 
     * @param listener the listener to attach 
     */  
    public void addListener(LeaderLatchListener listener)  
    {  
        listeners.addListener(listener);  
    }  

private[spark] class ZooKeeperLeaderElectionAgent(val masterActor: ActorRef,  
    masterUrl: String, conf: SparkConf)  
  extends LeaderElectionAgent with LeaderLatchListener with Logging  {  
  val WORKING_DIR = conf.get("spark.deploy.zookeeper.dir", "/spark") + "/leader_election"  
  // zk是通过CuratorFrameworkFactory创建的ZooKeeper实例  
  private var zk: CuratorFramework = _  
  // leaderLatch：Curator负责选出Leader。  
  private var leaderLatch: LeaderLatch = _  
  private var status = LeadershipStatus.NOT_LEADER  
  
  override def preStart() {  
  
    logInfo("Starting ZooKeeper LeaderElection agent")  
    zk = SparkCuratorUtil.newClient(conf)  
    leaderLatch = new LeaderLatch(zk, WORKING_DIR)  
    leaderLatch.addListener(this)  
  
    leaderLatch.start()  
  } 

override def isLeader() {  
  synchronized {  
    // could have lost leadership by now.  
    //现在leadership可能已经被剥夺了。。详情参见Curator的实现。  
    if (!leaderLatch.hasLeadership) {  
      return  
    }  
  
    logInfo("We have gained leadership")  
    updateLeadershipStatus(true)  
  }  
}  
  
override def notLeader() {  
  synchronized {  
    // 现在可能赋予leadership了。详情参见Curator的实现。  
    if (leaderLatch.hasLeadership) {  
      return  
    }  
  
    logInfo("We have lost leadership")  
    updateLeadershipStatus(false)  
  }  
}  

def updateLeadershipStatus(isLeader: Boolean) {  
    if (isLeader && status == LeadershipStatus.NOT_LEADER) {  
      status = LeadershipStatus.LEADER  
      masterActor ! ElectedLeader  
    } else if (!isLeader && status == LeadershipStatus.LEADER) {  
      status = LeadershipStatus.NOT_LEADER  
      masterActor ! RevokedLeadership  
    }  
  }