什么是Zookeeper

分布式开源协调系统，数据模型简单，可以实现同步，配置管理，分组管理，分命名空间管理等。

技术本质

一个原子消息传递系统，它使所有服务器保持同步

FLP(3个科学家名字命名) 理论角度，Zookeeper是 SF的 ，使用超时机制来保证Livessnes

扩展FLP：

基于消息的异步通信系统，即使只有一个进程失败，也没有一个确定性算法，保证存活节点达到一致

确定性算法: 给定一个输入，一定会产生相同的输出

基于消息的异步通信系统 : 没有统一时钟，不能时间同步，不能使用超时，不能探测失败，消息可以任意延迟，消息可以乱序

扩展FLP不可能三角

Safety 系统中非故障节点达成了一致和合法的共识

Liveness 系统中非故障节点在有效时间能够达成共识

Fault Tolerance 协议必须在节点故障的时候同样有效

CAP理论角度 zookeeper 是CP的：

当leader节点down，剩余节点则会发起选举，选举期间导致短暂不可用。保证了强一致性而无法保证高可用性

数据模型

Watches监控节点变化

Data Access，Znode数据原子读写

Ephemeral Nodes临时节点，生命周期，与创建节点的进程一样

Sequence Nodes 创建的节点有序

client开发步骤

设计 path

选择znode类型

设计znode存储的内容

设计Watch client 关注什么事件，事件发生后如何处理

集群选举算法 - 最小节点获胜

算法说明

当 Leader 节点挂掉的时候，持有最小编号 znode 的集群节点成为新的 Leader

设计path

集群公用父节点，例如：/examples/leader

设计Znode类型

所有服务，需要在父节点下创建znode，由于持有最小编号 znode是leader，选择用ephemeral_sequential 类型 znode

设计Znode存储的内容

一般服务部署在多台机子上，这里我们可以按需定制，一般会存储ip

设计Watch

节点启动或者重连后，在 parent 目录下创建自己的 ephemeral_sequntial znode

如果自己的 znode 编号是最小的，则成为Leader，否则 watch parent 目录；

当 parent 目录有节点删除的时候，首先判断其是否是 Leader 节点，然后再看其编号是否正好比自己小1，如果是则自己成为 Leader，如果不是继续 watch。

Curator Framework实现

Curator Framework简化zookeeper，原始api的工具

内部实现： LeaderSelector 最小节点获胜。

pom如下：

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.14</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>4.1.0</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-client</artifactId>
    <version>4.1.0</version>
</dependency>

实例代码：

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.leader.*;
import java.io.Closeable;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 从Curator Framework 的sample中fork
 */
public class LeaderSelectClient extends LeaderSelectorListenerAdapter
 implements Closeable {
    private final String name;
    private final LeaderSelector leaderSelector;
    private final AtomicInteger leaderCount = new AtomicInteger();

    public LeaderSelectClient( 
               CuratorFramework client,
               String path,
               String name) {
        this.name = name;
        leaderSelector = new LeaderSelector(client, path, this);
        // 重新选举时自动入队
        leaderSelector.autoRequeue();
    }
    public void start() throws IOException {
        // 调用这个方法开启后台选举
        leaderSelector.start();
    }

    @Override
    public void close() throws IOException {
        leaderSelector.close();
    }
    @Override
    public void takeLeadership(CuratorFramework client)
     throws Exception {
        // 如果，不想开启下次选举，这个方法应该一致 不返回
        final int waitSeconds = (int) (5 * Math.random()) + 1;
        long sessionId = client.getZookeeperClient()
                          .getZooKeeper().getSessionId();
        System.out.println(name + " is now the leader. Waiting "
            + waitSeconds + " seconds...");
        System.out.println(name + " has been leader " 
         + leaderCount.getAndIncrement() 
         + " time(s) before.sessionId:"+ sessionId);
        try {
            Thread.sleep(TimeUnit.SECONDS.toMillis(waitSeconds));
            while(true);
        } catch (InterruptedException e) {
            System.err.println(name + " was interrupted.");
            Thread.currentThread().interrupt();
        } finally {
            System.out.println(name + " relinquishing leadership.\n");
        }
    }
}

测试代码：

import com.google.common.collect.Lists;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.List;

public class LeaderSelectorExample {
    private static final int CLIENT_QTY = 10;

    private static final String PATH = "/examples/leader";

    public static void main( String[] args ) throws Exception {

        List<CuratorFramework> clients = Lists.newArrayList();
        List<LeaderSelectClient> examples = Lists.newArrayList();

        try {
            for ( int i = 0; i < CLIENT_QTY; ++i ) {
                CuratorFramework client = CuratorFrameworkFactory.newClient( 
                          "127.0.01:2181",
                           new ExponentialBackoffRetry( 1000, 3 ) 
                );
                LeaderSelectClient example = 
                           new LeaderSelectClient(
                               client,
                               PATH,
                               "Client #" + i );
                clients.add( client );
                examples.add( example );
                client.start();
                example.start();
            }

            System.out.println( "Press enter/return to quit\n" );
            new BufferedReader( new InputStreamReader( System.in ) )
                .readLine();
        } finally {
            System.out.println( "Shutting down..." );
            examples.forEach( CloseableUtils::closeQuietly );
            clients.forEach( CloseableUtils::closeQuietly );
        }
    }
}

本地运行结果