用zookeeper实现简单的master选举

Master选举可以说是ZooKeeper最典型的应用场景了。比如HDFS中Active NameNode的选举、YARN中Active ResourceManager的选举和HBase中Active HMaster的选举等。

针对Master选举的需求，通常情况下，我们可以选择常见的关系型数据库中的主键特性来实现：希望成为Master的机器都向数据库中插入一条相同主键ID的记录，数据库会帮我们进行主键冲突检查，也就是说，只有一台机器能插入成功——那么，我们就认为向数据库中成功插入数据的客户端机器成为Master。

依靠关系型数据库的主键特性确实能够很好地保证在集群中选举出唯一的一个Master。但是，如果当前选举出的Master挂了，那么该如何处理？谁来告诉我Master挂了呢？显然，关系型数据库无法通知我们这个事件。但是，ZooKeeper可以做到！

利用ZooKeepr的强一致性，能够很好地保证在分布式高并发情况下节点的创建一定能够保证全局唯一性，即ZooKeeper将会保证客户端无法创建一个已经存在的ZNode。也就是说，如果同时有多个客户端请求创建同一个临时节点，那么最终一定只有一个客户端请求能够创建成功。利用这个特性，就能很容易地在分布式环境中进行Master选举了。

成功创建该节点的客户端所在的机器就成为了Master。同时，其他没有成功创建该节点的客户端，都会在该节点上注册一个节点变更的Watcher，用于监控当前Master机器是否存活，一旦发现当前的Master挂了，那么其他客户端将会重新进行Master选举。

这样就实现了Master的动态选举

下面的例子里，有三个节点node1、node2、node3参与master选举，先后启动三个节点，代码如下：

Node1.java

package com.zhuyun.election;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

public class Node1 implements Watcher{
ZooKeeper zk;
String hostPort;
String znode;

public Node1(String hostPort,String znode) throws Throwable{
this.hostPort = hostPort;
this.znode = znode;

zk = new ZooKeeper(hostPort, 3000, this);
try {
//每个客户端都创建同一个节点，如果创建成功，则该客户端是master
zk.create(znode, "node1".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
System.out.println("master节点是：node1");
} catch (KeeperException.NodeExistsException e) {
//如果抛出节点存在的异常，则master已经存在，在该节点上添加watcher
System.out.println("master节点是：" + new String(zk.getData(znode, false, null)));
zk.exists(znode, true);
}

}

@Override
public void process(WatchedEvent event) {
try {

if (event.getType() == EventType.NodeDeleted) {
try {
zk.create(znode, "node1".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
System.out.println("master节点是：node1");
} catch (KeeperException.NodeExistsException e) {
System.out.println("master节点是：" + new String(zk.getData(znode, false, null)));
zk.exists(znode, true);
}
}

} catch (KeeperException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

}

public static void main(String[] args) throws Throwable {
new Node1("192.168.10.203:2181", "/com/zhuyun/MasterElection/master");

System.in.read();
}
}

Node2.java

package com.zhuyun.election;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

public class Node2 implements Watcher{
ZooKeeper zk;
String hostPort;
String znode;

public Node2(String hostPort,String znode) throws Throwable{
this.hostPort = hostPort;
this.znode = znode;

zk = new ZooKeeper(hostPort, 3000, this);
try {
//每个客户端都创建同一个节点，如果创建成功，则该客户端是master
zk.create(znode, "node2".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
System.out.println("master节点是：node2");
} catch (KeeperException.NodeExistsException e) {
//如果抛出节点存在的异常，则master已经存在，在该节点上添加watcher
System.out.println("master节点是：" + new String(zk.getData(znode, false, null)));
zk.exists(znode, true);
}

}

@Override
public void process(WatchedEvent event) {
try {

if (event.getType() == EventType.NodeDeleted) {
try {
zk.create(znode, "node2".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
System.out.println("master节点是：node2");
} catch (KeeperException.NodeExistsException e) {
System.out.println("master节点是：" + new String(zk.getData(znode, false, null)));
zk.exists(znode, true);
}
}

} catch (KeeperException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

}

public static void main(String[] args) throws Throwable {
new Node2("192.168.10.203:2181", "/com/zhuyun/MasterElection/master");

System.in.read();
}
}

Node3.java

package com.zhuyun.election;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

public class Node3 implements Watcher{
ZooKeeper zk;
String hostPort;
String znode;

public Node3(String hostPort,String znode) throws Throwable{
this.hostPort = hostPort;
this.znode = znode;

zk = new ZooKeeper(hostPort, 3000, this);
try {
//每个客户端都创建同一个节点，如果创建成功，则该客户端是master
zk.create(znode, "node3".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
System.out.println("master节点是：node3");
} catch (KeeperException.NodeExistsException e) {
//如果抛出节点存在的异常，则master已经存在，在该节点上添加watcher
System.out.println("master节点是：" + new String(zk.getData(znode, false, null)));
zk.exists(znode, true);
}

}

@Override
public void process(WatchedEvent event) {
try {

if (event.getType() == EventType.NodeDeleted) {
try {
zk.create(znode, "node3".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
System.out.println("master节点是：node3");
} catch (KeeperException.NodeExistsException e) {
System.out.println("master节点是：" + new String(zk.getData(znode, false, null)));
zk.exists(znode, true);
}
}

} catch (KeeperException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

}

public static void main(String[] args) throws Throwable {
new Node3("192.168.10.203:2181", "/com/zhuyun/MasterElection/master");

System.in.read();
}
}

三个节点都启动以后，一般是第一个启动的节点成为master，其他节点是slave；当关闭node1的时候，剩下两个节点重新选举出一个master出来，结果类似下面：

master节点是：node1
master节点是：node2