首先看一下LeaderlatchListener。它在LeaderLatch状态变化的时候被通知：

1. 在该节点被选为Leader的时候，接口isLeader()会被调用
2. 在节点被剥夺Leader的时候，接口notLeader()会被调用

因为通知是异步的。因此有可能在接口被调用的时候。这个状态是准确的。须要确认一下LeaderLatch的hasLeadership()是否的确是true/false。这一点在接下来Spark的实现中能够得到体现。

/**
* 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();
}

LeaderLatch负责在众多连接到ZooKeeper Cluster的竞争者中选择一个Leader。

Leader的选择机制能够看ZooKeeper的详细实现，LeaderLatch这是完毕了非常好的封装。我们只须要要知道在初始化它的实例后。须要通过

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);
    }

通过addListener能够将我们实现的Listener加入到LeaderLatch。在Listener里。我们在两个接口里实现了被选为Leader或者被剥夺Leader角色时的逻辑就可以。

4. ZooKeeperLeaderElectionAgent的实现

实际上因为有Curator的存在。Spark实现Master的HA就变得非常easy了，ZooKeeperLeaderElectionAgent实现了接口LeaderLatchListener。在isLeader()确认所属的Master被选为Leader后，向Master发送消息ElectedLeader，Master会将自己的状态改为ALIVE。

当noLeader()被调用时，它会向Master发送消息RevokedLeadership时，Master会关闭。

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()
  }

在prestart中，启动了leaderLatch来处理选举ZK中的Leader。就如在上节分析的，基本的逻辑在isLeader和noLeader中。

  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)
    }
  }

updateLeadershipStatus的逻辑非常easy，就是向Master发送消息。

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
    }
  }

5. 设计理念

为了解决Standalone模式下的Master的SPOF，Spark採用了ZooKeeper提供的选举功能。Spark并没有採用ZooKeeper原生的Java API，而是採用了Curator，一个对ZooKeeper进行了封装的框架。採用了Curator后，Spark不用管理与ZooKeeper的连接，这些对于Spark来说都是透明的。

Spark只使用了100行代码，就实现了Master的HA。当然了，Spark是站在的巨人的肩膀上。谁又会去反复发明轮子呢？

请您支持：