上一篇博客讨论了基于zookeeper的分布式队列的机制，这个机制除了可以做分布式队列以外，稍加修改，还可以做更多的事情，例如接下来要讨论的领导选举和分布式锁的功能等。

领导选举

领导选举的应用场景可以理解为：多个节点同时想干一件事（都想当老大），但最终只有一个节点被授权（老大只可能有一个）

例如：一主多从模式下，如果主节点挂掉了，那么所有的从节点都要竞选成为主节点，但只有一个节点可以成为主节点。

领导选举的特点就是选举是一次性的，只要主节点不挂掉，他就一直是领导。

基于zookeeper的领导选举的实现方案如下：

提前准备一个znode作为一个队列。例如/election

1. 每个节点都调用create()创建一个node，带上sequence和ephemeral标签。名字为/election/queue-
2. 在/election节点调用getChildren()，不设置watch，拿到/election下的整个列表
3. 如果在第1步创建的znode中自增长的数字，如果在列表中是最小的，则它就是leader了
4. 如果发现自己不是leader，则删除在第1步创建的znode

此时，数字最小的那个znode成为了leader，其余的znode都被创建它的节点删除了。如果某个节点得到竞选通知比较晚，此时它再次调用create()，创建出来的znode的数字肯定会更大，按照上面的流程，它会直接删除自己创建的znode并退出选举。这就保证了任何时候只有一个leader。

领导选举的示例代码

我们用java的多线程来模拟一下多个节点，进行一次选举

首先，新建一个类实现Callable接口。这个类参与选举。如果成功竞选，则返回自己创建的znode地址，否则，返回空字符串。

class LeaderElection implements Callable<String> {
    ZooKeeper zooKeeper;
    String znodePath;
    String name;

    String newNodeName = null;
    boolean isLeader = false;

    LeaderElection(ZooKeeper zooKeeper, String znodePath, String name) {
        this.zooKeeper = zooKeeper;
        this.znodePath = znodePath;
        this.name = name;
    }

    @Override
    public String call() throws Exception {
        String fullPath = znodePath + "/queue-";
        newNodeName = zooKeeper.create(fullPath, this.name.getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(this.name + " created node:" + newNodeName);
        List<String> nodeList = zooKeeper.getChildren(this.znodePath, false);
        isLeader = checkIsLeader(nodeList, newNodeName);
        System.out.println(this.name + " is leader:" + isLeader);
        if (!isLeader) {
            zooKeeper.delete(newNodeName, zooKeeper.exists(newNodeName, false).getVersion());
            return "";
        } else {
            return newNodeName;
        }
    }

    private boolean checkIsLeader(List<String> nodeList, String newNodeName) {
        int curValue = getNumber(newNodeName);
        for (String node : nodeList) {
            if (getNumber(node) < curValue) {
                return false;
            }
        }
        return true;
    }

    private int getNumber(String queueNode) {
        queueNode = queueNode.substring(queueNode.length() - 10, queueNode.length());
        return Integer.parseInt(queueNode);
    }

}

接下来，创建一个线程池，并启动10个选举线程，就可以了。

@Test
public void testLeaderSelection() throws IOException, InterruptedException, ExecutionException, KeeperException {
    ZooKeeper zooKeeper = zookeeperClient.connect();

    ExecutorService threadpool = Executors.newCachedThreadPool();
    List<Future<String>> futureList = new LinkedList<>();
    for (int i = 0; i < 10; i++) {
        Callable<String> callable = new LeaderElection(zooKeeper, zookeeperClient.getRestoreLockPath(), "election-" + i);
        Future<String> future = threadpool.submit(callable);
        futureList.add(future);
    }
    for (Future<String> future : futureList) {
        String leaderNode = future.get();
        if (!leaderNode.isEmpty()) {
            System.out.println("leader node is " + leaderNode);
        }
    }

}

运行上面的测试，会打印出谁是领导，有且只有一个。输出如下：

election-5 created node:/jobcenter/restorelocks/queue-0000000090
election-4 created node:/jobcenter/restorelocks/queue-0000000091
election-3 created node:/jobcenter/restorelocks/queue-0000000092
election-1 created node:/jobcenter/restorelocks/queue-0000000093
election-2 created node:/jobcenter/restorelocks/queue-0000000094
election-0 created node:/jobcenter/restorelocks/queue-0000000095
election-6 created node:/jobcenter/restorelocks/queue-0000000096
election-7 created node:/jobcenter/restorelocks/queue-0000000097
election-4 is leader:false
election-5 is leader:true
election-3 is leader:false
election-9 created node:/jobcenter/restorelocks/queue-0000000098
election-8 created node:/jobcenter/restorelocks/queue-0000000099
election-1 is leader:false
election-2 is leader:false
election-0 is leader:false
election-6 is leader:false
election-7 is leader:false
election-9 is leader:false
election-8 is leader:false
leader node is /jobcenter/restorelocks/queue-0000000090

分布式锁

这里所说的分布式锁，指的是：多个节点都需要做一件事，但这件事在任何一个时间点上只能有一个节点在做，如果多个节点同时做的话，会有并发问题，可能造成数据错误。

分布式锁的实现方案跟上面所说的领导选举十分相似，我还列一下这个过程吧

还是要提前准备一个znode作为一个队列。例如/distribute-lock

1. 每个节点都调用create()创建一个node，带上sequence和ephemeral标签。名字为/distribute-lock/lock-
2. 在/distribute-lock节点调用getChildren()，不设置watch，拿到/distribute-lock下的整个列表
3. 如果在第1步创建的znode中自增长的数字，如果在列表中是最小的，则它就是leader了，就可以退出这个过程了。
4. 如果发现自己不是leader，则调用exist()来监听比自己小一号的znode，这一步要设置watch。当watch触发时，goto step 2，来对比自己是不是leader(因为新的leader会在完事以后删掉自己创建的znode)。
5. ……

怎么理解比自己小一号的znode：如果自己创建的znode是lock-0000000023的话，则小一号的不一定就是0000000022。正确的做法是，对整个列表从小到大进行排序，拿到自己的index，如果index是0，则自己就是leader，而比自己小一号的znode就是list[index - 1]。

如果在第3步发现自己是leader了，则表示自己拿到了分布式锁，就可以开始执行某个过程了，执行完了以后删除自己在第1步时创建的znode，表示释放了自己的分布式锁。这时，原本倒数第二小的znode成为了最小的znode，对应的节点就拿到了分布式锁。还有一种情况就是，如果老四监听老三，但老三在成为老大之前退出了，上面的机制会使老四去监听老二。这个过程会不断重复。

由于分布式锁的逻辑跟上面领导选举比较相似，读者可以自己用代码来实现一下玩玩。

参考资料

http://zookeeper.apache.org/doc/current/recipes.html