请注意，此篇文章并不是介绍Zookeeper集群内部Leader的选举机制，而是应用程序使用Zookeeper作为选举。
使用Zookeeper进行选举，主要用到了Znode的两个性质：
- 临时节点（EPHEMERAL）
- 序列化节点（SEQUENCE）
每一个临时的序列化节点代表着一个客户端（client），也就是选民。主要的设计思路如下：
首先，创建一个选举的节点，我们叫做/election。 然后，每有一个客户端加入，就创建一个子节点/election/n_xxx，这个节点是EPHEMERAL并且SEQUENCE，xxx就是序列化产生的单调递增的数字。 在所有子节点中，序列数字做小的被选举成Leader。
上面的并不是重点，重点是Leader失败的检测，Leader失败后，一个新的客户端（client）将被选举成Leader。实现这个过程的一个最简单的方式是 所有的客户端（client）都监听Leader节点，一旦Leader节点消失，将通知所有的客户端（client）执行Leader选举过程，序列数字最小的将被选举成Leader。 这样实现看似没有问题，但是当客户端（client）数量非常庞大时，所有客户端（client）都将在/election节点执行getChildren()，这对Zookeeper 的压力是非常大的。为了避免这种“惊群效应”，我们可以让客户端只监听它前一个节点（所有序列数字比当前节点小，并且是其中最大的那个节点）。 这样，Leader节点消失后，哪个节点收到了通知，哪个节点就变成Leader，因为所有节点中，没有比它序列更小的节点了。
具体步骤如下：
- 使用EPHEMERAL和SEQUENCE创建节点/election/n_xxx，我们叫做z。
- C为/election的子节点集合，i是z的序列数字。
- 监听/election/n_j，j是C中小于i的最大数字。
接收到节点消失的事件后：
- C为新的/election的子节点集合
- 如果z是集合中最小的节点，则z被选举成Leader
- 如果z不是最小节点，则继续监听/election/n_j，j是C中小于i的最大数字。
具体代码如下：
public class Candidate implements Runnable, Watcher {
//zk
private ZooKeeper zk;
//临时节点前缀
private String perfix = "n_";
//当前节点
private String currentNode;
//前一个最大节点
private String lastNode;
/**
* 构造函数
* @param address zk地址
*/
public Candidate(String address) {
try {
this.zk = new ZooKeeper(address, 3000, this);
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 加入选举
*/
@Override
public void run() {
try {
//创建临时节点
currentNode = zk.create("/zookeeper/election/" + perfix, Thread.currentThread().getName().getBytes(),
ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
//选举
election();
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
/**
* 从小到大排序临时节点
* @param children
* @return
*/
private List<String> getSortedNode(List<String> children) {
return children.stream().sorted(((o1, o2) -> {
String sequence1 = o1.split(perfix)[1];
String sequence2 = o2.split(perfix)[1];
BigDecimal decimal1 = new BigDecimal(sequence1);
BigDecimal decimal2 = new BigDecimal(sequence2);
int result = decimal1.compareTo(decimal2);
return result;
})).collect(toList());
}
/**
* 选举过程
*/
private void election(){
try{
while (true){
//获取/election节点中的所有子节点
List<String> children = zk.getChildren("/zookeeper/election", false);
//所有子节点排序（从小到大）
List<String> sortedNodes = getSortedNode(children);
//获取最小节点
String smallestNode = sortedNodes.get(0);
//当前节点就是最小节点，被选举成Leader
if (currentNode.equals("/zookeeper/election/"+smallestNode)) {
System.out.println(currentNode + "被选举成Leader。");
Thread.sleep(5000);
//模拟Leader节点死去
System.out.println(currentNode+"已离去");
zk.close();
break;
}
//当前节点不是最小节点，监听前一个最大节点
else {
//前一个最大节点
lastNode = smallestNode;
//找到前一个最大节点，并监听
for (int i = 1; i < sortedNodes.size(); i++) {
String z = sortedNodes.get(i);
//找到前一个最大节点，并监听
if (currentNode.equals("/zookeeper/election/"+z)) {
zk.exists("/zookeeper/election/" + lastNode, true);
System.out.println(currentNode+"监听"+lastNode);
//等待被唤起执行Leader选举
synchronized (this){
wait();
}
break;
}
lastNode = z;
}
}
}
}catch (Exception e) {
e.printStackTrace();
}
}
/**
* 观察器通知
* @param event
*/
@Override
public void process(WatchedEvent event) {
//监听节点删除事件
if (event.getType().equals(Event.EventType.NodeDeleted)) {
//被删除的节点是前一个最大节点，唤起线程执行选举
if (event.getPath().equals("/zookeeper/election/" + lastNode)) {
System.out.println(currentNode+"被唤起");
synchronized (this){
notify();
}
}
}
}
}
我们将启动5个线程作为参选者，模拟每一个Leader死去，并重新选举的过程。启动程序如下：
public class Application {
private static final String ADDRESS = "149.28.37.147:2181";
public static void main(String[] args) throws InterruptedException {
setLog();
ExecutorService es = Executors.newFixedThreadPool(5);
for (int i=0;i<5;i++){
es.execute(new Candidate(ADDRESS));
}
es.shutdown();
}
/**
* 设置log级别为Error
*/
public static void setLog(){
//1.logback
LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
//获取应用中的所有logger实例
List<Logger> loggerList = loggerContext.getLoggerList();
//遍历更改每个logger实例的级别,可以通过http请求传递参数进行动态配置
for (ch.qos.logback.classic.Logger logger:loggerList){
logger.setLevel(Level.toLevel("ERROR"));
}
}
}
运行结果如下：
/zookeeper/election/n_0000000133被选举成Leader。 /zookeeper/election/n_0000000134监听n_0000000133 /zookeeper/election/n_0000000137监听n_0000000136 /zookeeper/election/n_0000000135监听n_0000000134 /zookeeper/election/n_0000000136监听n_0000000135 /zookeeper/election/n_0000000133已离去 /zookeeper/election/n_0000000134被唤起 /zookeeper/election/n_0000000134被选举成Leader。 /zookeeper/election/n_0000000134已离去 /zookeeper/election/n_0000000135被唤起 /zookeeper/election/n_0000000135被选举成Leader。 /zookeeper/election/n_0000000135已离去 /zookeeper/election/n_0000000136被唤起 /zookeeper/election/n_0000000136被选举成Leader。 /zookeeper/election/n_0000000136已离去 /zookeeper/election/n_0000000137被唤起 /zookeeper/election/n_0000000137被选举成Leader。 /zookeeper/election/n_0000000137已离去
Zookeeper作为选举的应用就介绍完了，项目示例请参考：https://github.com/liubo-tech/zookeeper-application。