[大数据]Zookeeper入门学习

Zookeeper入门学习

Zookeeper工作机制

Zookeeper从设计模式的角度来理解，是一个基于观察者模式设计的分布式服务管理框架，它负责存储和管理大家都关心的数据，然后接受观察者的注册，一旦这些数据的状态发生变化，Zookeeper就将负责通知已经在zk上注册的那些观察者，并做出相应的反应。

Zookeeper相当于：文件系统（存储数据：服务器信息等） + 通知机制（给观察者，即服务发送通知）

Zookeeper特点

1. Zookeeper是一个领导者（Leader），**多个跟随者（Follower）**组成的集群。
2. 集群中只要有半数以上的节点存活，zk集群就能正常提供服务，所以zk适合安装奇数台服务器。
3. 全局数据一致：每个Server保存一份相同的数据副本，Client无论连接到哪个Server，数据都是一致的
4. 更新请求顺序执行，来自同一个Client的更新请求按其发送顺序依次执行
5. 数据更新原子性，一次数据更新要么成功，要么失败
6. 实时性，数据同步具有实时性

Zookeeper数据结构

ZooKeeper 数据模型的结构与 Unix 文件系统很类似，整体上可以看作是一棵树，每个

节点称做一个 ZNode。每一个 ZNode 默认能够存储1MB的数据，每个ZNode都可以通过

其路径唯一标识。

Zookeeper应用场景

提供的服务包括：统一命名服务、统一配置管理、统一集群管理、服务器节点动态上下线，软负载均衡等

统一命名服务

在分布式环境下，经常需要地应用/服务进行统一明明，便于识别，例如：IP不容易记住，而域名容易记住（类似nginx）

统一配置管理

• 分布式环境中，配置文件同步很常见：
  • 一般要求一个集群中，所有节点的配置信息是一致的，比如Kafka集群。
  • 对配置文件修改后，希望能快速同步到各个节点上
• 配置管理可交由Zookeeper实现
  • 可将配置信息写入Zookeeper中的一个ZNode
  • 各个客户端服务器监听这个ZNode（如下图）

统一集群管理

• 分布式环境中，实时掌握每个节点的状态是必要的
  • 可根据节点实时状态做出一些调整
• Zookeeper可以实现实时监控节点状态变化
  • 可将节点信息写入Zookeeper中的一个ZNode
  • 监听这个ZNode可获取它的实时状态变化

服务器动态上下线

客户端能实时洞察到服务器上下线的变化：

软负载均衡

在Zookeeper中记录每台服务器的访问数，让访问数最少的服务器去处理最新的客户端请求：

Zookeeper安装及配置参数解读

百度

配置参数解读

1. tickTime = 2000：通信心跳时间，Zookeeper服务器与客户端心跳时间，单位毫秒

   

2. initLimit = 10：LF初始通信时限，即：Leader和Follower初始连接时能容忍的最多心跳数（tickTime的数量）.

   

3. syncLimit = 5：LF同步通信时限，即：Leader和Follower之间通信时间如果超过syncLimit * tickTime，Leader认为Follwer死掉，从服务器列表中删除Follwer。

   

4. dataDir：保存Zookeeper中的数据：注意默认的tmp目录，容易被Linux系统定期删除，所以一般不用默认的tmp目录

5. clientPort = 2181：客户端的默认连接端口，通常不做修改

Zookeeper集群安装中遇到的问题

1. 伪集群启动失败：https://www.cnblogs.com/zhangzl419/p/13200325.html

2. 配置文件中在配置后添加注释导致报错：

   

Zookeeper选举机制（面试重点）

几个概念

SID：服务器ID。用来唯一标识一台zk集群中的机器，每台机器不能重复，和配置集群时设置的myid一致。

ZXID：事务ID。ZXID时一个事务ID，用来标识一次服务器状态的变更。在某一时刻，集群中的每台机器的ZXID值不一定完全一致，这和zk服务器对于客户端的“更新请求”的处理逻辑有关。

Epoch：每个Leader任期的代号。没有Leader时，同一轮投票过程中的逻辑时钟值是相同的。每投完一次票这个数据就会增加。

选举机制——第一次启动

选举流程：

1. 服务器1启动，发起一次选举。服务器1给自己投一票。此时服务器1一票，不够板书以上（3票），选举无法完成，服务器1的状态保持为LOOKING；
2. 服务器2启动，再发起一次选举。服务器1和2分别投自己一票并交换选票信息：此时服务器1发现服务器2的myid比自己目前投票推举的（服务器1）大，更改选票为推举服务器2。此时服务器1为0票，服务器2为2票，没有半数以上的结果，选举无法完成，服务器1, 2的状态保持为LOOKING；
3. 服务器3启动，发起一次选举，此时服务器1和2都会更改选票为服务器3。此次投票结果：服务器1为0票，服务器2为0票，服务器3为3票。此时服务器3的票数已经超过板书，服务器3当选为Leader。服务器1,2更改状态为Following，服务器3更改状态为Leading；
4. 服务器4启动：发起一次选举。此时服务器1、2、3已经不是LOOKING状态，不会更改选票信息。交换选票信息结果：服务器3为3票，服务器4为1票。此时服务器4服从多数，更改选票信息为服务器3，并更改状态为FOLLOWING；
5. 服务器5启动：同4

选举机制——非第一次启动

当zk集群有以下两种情况之一，就会开始进入Leader选举：

1. 服务器初始化启动（上面那种情况）
2. 服务器运行期间无法和Leader保持连接

而当一台机器进入Leader选举流程时，当前集群也可能会处于以下两种状态

1. 集群中本来就已经存在一个Leader

   对于第一种已经存在Leader的情况，机器试图去选举Leader时，会被告知当前服务器的Leader信息，对于该机器来说，仅仅需要和Leader机器建立连接，并进行状态同步即可。

2. 集群中不存在Leader

   假设zk集群有五台服务器组成，SID分别为1,2,3,4,5，ZXID分别为8,8,8,7,7，并且此时SID为3的服务器是Leader。某一时刻，3和5机器挂了，因此开始进行Leader选举，剩下机器将遵循以下原则进行选举：

   1. Epoch大的直接胜出
   2. epoch相同，事务ID大的直接胜出
   3. 事务ID相同，服务器ID大的直接胜出

Zookeeper客户端命令行操作

连接

/opt/module/zk1/apache-zookeeper-3.5.7-bin/bin/zkCli.sh  -server 192.168.59.110:2182  #连接集群中某个客户端

显示所有操作命令

help

ZNode节点数据信息

查看当前znode中包含的内容

[zk: 192.168.59.110:2182(CONNECTED) 6] ls /
[zookeeper]

查看当前节点详细数据

[zk: 192.168.59.110:2182(CONNECTED) 7] ls -s /
[zookeeper]cZxid = 0x0
ctime = Thu Jan 01 08:00:00 CST 1970
mZxid = 0x0
mtime = Thu Jan 01 08:00:00 CST 1970
pZxid = 0x0
cversion = -1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 1

• czxid：创建节点的事务 zxid

  每次修改 ZooKeeper 状态都会产生一个 ZooKeeper 事务 ID。事务 ID 是 ZooKeeper 中所有修改总的次序。每次修改都有唯一的 zxid，如果 zxid1 小于 zxid2，那么 zxid1 在 zxid2 之前发生。

• ctime：znode 被创建的毫秒数（从 1970 年开始）

• mzxid：znode 最后更新的事务 zxid

• mtime：znode 最后修改的毫秒数（从 1970 年开始）

• pZxid：znode 最后更新的子节点 zxid

• cversion：znode 子节点变化号，znode 子节点修改次数

• dataversion：znode 数据变化号

• aclVersion：znode 访问控制列表的变化号

• ephemeralOwner：如果是临时节点，这个是 znode 拥有者的 session id。如果不是临时节点则是 0。

• dataLength：znode 的数据长度

• numChildren：znode 子节点数量

节点类型（持久、短暂、有序号、无序号）

• 持久（Persistent）：客户端和服务器端断开连接后，创建的节点不删除
• 短暂（Ephemeral）：客户端和服务器断开连接后，创建的节点自己删除

1. 持久化目录节点：客户端与zk断开连接后，该节点依旧存在
2. 持久化顺序编号目录节点：客户端与zk断开连接过后，该节点依旧存在，只是zk给该节点名称进行顺序编号
3. 临时目录节点：客户端与zk断开连接后，该节点被删除
4. 临时顺序编号目录节点：客户端与zk断开连接后，该节点被删除，只是zk给该节点名称进行顺序编号~

说明：创建znode时设置顺序标识，znode名称后会附加一个值，顺序号以单调递增，由父节点维护

注意：在分布式系统中，顺序号可以被用于对所有的时间进行全局排序，这样客户端可以通过顺序号判断事件的顺序。

分别创建2个普通节点（永久节点+不带序号）

[zk: 192.168.59.110:2182(CONNECTED) 11] create /sanguo "diaochan"
Created /sanguo
# 获取节点的值
[zk: 192.168.59.110:2182(CONNECTED) 17] get -s /sanguo
diaochan
cZxid = 0x400000003
ctime = Sun Aug 15 10:37:00 CST 2021
mZxid = 0x400000003
mtime = Sun Aug 15 10:37:00 CST 2021
pZxid = 0x400000005
cversion = 2
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 8
numChildren = 2

创建带序号的节点（永久节点 + 带序号）

# 创建带序号的节点
[zk: 192.168.59.110:2182(CONNECTED) 18] create -s /sanguo/wuguo "zhangfei"
Created /sanguo/wuguo0000000002 #带序号
[zk: 192.168.59.110:2182(CONNECTED) 19] create -s /sanguo/wuguo "zhangfei"
Created /sanguo/wuguo0000000003
[zk: 192.168.59.110:2182(CONNECTED) 20] create -s /sanguo/wuguo "zhangfei"
Created /sanguo/wuguo0000000004

如果原来没有序号节点，序号从 0 开始依次递增。如果原节点下已有 2 个节点，则再排

序时从 2 开始，以此类推。

创建短暂节点（短暂节点 + 不带序号or带序号）

# 创建带序号
[zk: 192.168.59.110:2182(CONNECTED) 22] create -e /sanguo/wuguo  "xuyou" 
Created /sanguo/wuguo
# 创建不带序号
[zk: 192.168.59.110:2182(CONNECTED) 23] create -e -s /sanguo/wuguo "xuyou"
Created /sanguo/wuguo0000000006

查看创建结果

[zk: 192.168.59.110:2182(CONNECTED) 26] ls /sanguo
shuguo            weiguo            wuguo             wuguo0000000002   wuguo0000000003   wuguo0000000004   wuguo0000000006
# 关闭客户端之后重新连接
[zk: 192.168.59.110:2182(CONNECTED) 1] ls /sanguo
[shuguo, weiguo, wuguo0000000002, wuguo0000000003, wuguo0000000004]

修改节点数据的值

[zk: 192.168.59.110:2183(CONNECTED) 22] set /sanguo "刘备"
[zk: 192.168.59.110:2183(CONNECTED) 23] get -s /sanguo
刘备 #修改成功

监听器原理

客户端注册监听它关心的目录节点，当目录节点发生变化（数据改变、节点删除、子目录节点增加删除）时，ZooKeeper 会通知客户端。监听机制保证 ZooKeeper 保存的任何的数据的任何改变都能快速的响应到监听了该节点的应用程序。

常见的监听：

1. 监听节点数据的变化：get path [watch]
2. 监听子节点增减的变化：ls path [watch]

节点的值变化监听

1. 在2182上注册监听/sanguo节点的数据变化

   [zk: 192.168.59.110:2182(CONNECTED) 0] get -w /sanguo
   

2. 在2183上修改/sanguo节点的数据

   [zk: 192.168.59.110:2183(CONNECTED) 24] set /sanguo "xixi"
   

3. 观察2182上的监听

   WATCHER::
   WatchedEvent state:SyncConnected type:NodeDataChanged path:/sanguo
   

注意：在2182上多次修改/sanguo的值，2183上不会再收到监听。因为注册一次，只能监听一次。想再次监听需要再次注册。

节点的子节点变化监听

1. 在2182上注册监听/sanguo节点的数据变化

   [zk: 192.168.59.110:2182(CONNECTED) 0] ls -w /sanguo
   

2. 在2183上修改/sanguo节点的数据

   [zk: 192.168.59.110:2183(CONNECTED) 24] create /sanguo/jin "simayi"
   

3. 观察2182上的监听

   WATCHER::
   WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/sanguo
   

节点的删除与查看

1. 删除节点

   [zk: 192.168.59.110:2183(CONNECTED) 28] delete /sanguo/jin
   

2. 递归删除节点：递归删除当前节点和下面的子节点

   [zk: 192.168.59.110:2183(CONNECTED) 29] deleteall /sanguo
   

客户端API操作

环境搭建

1. 创建一个工程

2. 添加pom依赖：junit、zookeeper（注意版本对应）、log4j-core

3. 配置log4j配置

   log4j.rootLogger=INFO, stdout 
   log4j.appender.stdout=org.apache.log4j.ConsoleAppender 
   log4j.appender.stdout.layout=org.apache.log4j.PatternLayout 
   log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n 
   log4j.appender.logfile=org.apache.log4j.FileAppender 
   log4j.appender.logfile.File=target/spring.log 
   log4j.appender.logfile.layout=org.apache.log4j.PatternLayout 
   log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n
   

4. 创建Zookeeper客户端

   public class ZookeeperClient {
       // 配置Zookeeper地址：注意：逗号前后不能有空格
       private static String connectString = "192.168.59.110:2181,192.168.59.110:2182,192.168.59.110:2183";
       // 超时时间
       private static int sessionTimeout = 2000;
       private ZooKeeper zkClient = null;
       @Before
       public void init() throws Exception {
           zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
               @Override
               public void process(WatchedEvent watchedEvent) {
                   // 收到事件通知后的回调函数（用户的业务逻辑）
                   System.out.println(watchedEvent.getType() + "--"
                           + watchedEvent.getPath());
                   // 再次启动监听
                   try {
                       List<String> children = zkClient.getChildren("/",
                               true);
                       for (String child : children) {
                           System.out.println(child);
                       }
                   } catch (Exception e) {
                       e.printStackTrace();
                   } }
               }
           );
       }
   }
   

5. 创建子节点

   // 创建子节点
   @Test
   public void create() throws Exception {
       // 参数 1：要创建的节点的路径； 参数 2：节点数据 ； 参数 3：节点权限 ； 参数 4：节点的类型
       String nodeCreated = zkClient.create("/sanguo", "guanyu".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
   }
   

   记一次报错：KeeperException$ConnectionLossException: KeeperErrorCode = ConnectionLoss for

   原因：sessionTimeout设置时间短了

   测试：在 192.168.59.110 的 zk 客户端上查看创建节点情况

   [zk: 192.168.59.110:2183(CONNECTED) 34] ls /
   [sanguo, zookeeper]
   

获取子节点并监听节点变化

1. 程序

   // 获取子节点
   @Test
   public void getChildren() throws Exception {
       List<String> children = zkClient.getChildren("/", true);
       for (String child : children) {
           System.out.println(child);
       }
       // 延时阻塞
       Thread.sleep(Long.MAX_VALUE);
   }
   

2. 执行可以在控制台看到如下输出

   zookeeper

   sanguo

3. 在linux创建一个节点，观察IDEA控制台

   [zk: 192.168.59.110:2183(CONNECTED) 36] create /sanguo/shuguo "liubei"
   

4. 删除节点，观察IDEA控制台

   [zk: 192.168.59.110:2183(CONNECTED) 36] delete /sanguo/shuguo
   

判断Znode是否存在

//判断ZNode是否存在
@Test
public void exist() throws KeeperException, InterruptedException {
    Stat stat = zkClient.exists("/hello", false);
    System.out.println(stat == null ? "not exist" : "exist");
}

客户端向服务端写数据流程

写流程之写入请求直接发送给Leader节点

半数以上写入就算成功，发送ack

写流程之写入请求直接发送给Follower节点

其实就是Follower将请求发送给Server

案例1:服务器动态上下线监听

需求

某分布式系统中，主节点可以有多台，可以动态上下线，任意一台客户端都能实现实时感知到主节点服务器的上下线。

具体实现

1. 首先在zk中创建/servers节点

   [zk: 192.168.59.110:2183(CONNECTED) 56] create /servers
   Created /servers
   

2. 服务器端向Zookeeper注册代码

   public class DistributeServer {
       private static final String CONNECT_STRING = "192.168.59.110:2181,192.168.59.110:2182,192.168.59.110:2183";
       private static final Integer SESSION_TIMEOUT = 200000;
       private ZooKeeper zk;
       private static final String PATH_NODE = "/servers";
   
       public static void main(String[] args) throws InterruptedException, KeeperException, IOException {
           DistributeServer server = new DistributeServer();
           // 获取链接
           server.getConnect();
           // 注册
           server.register(args[0]);
           // 启动业务逻辑（睡觉）
           Thread.sleep(20000);
       }
   
       // 注册
       public void register(String serverName) throws KeeperException, InterruptedException {
           zk.create(PATH_NODE + "/" + serverName, serverName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
           // 打印上线信息
           System.out.println(serverName + " is online ...... ");
       }
       // 获取zk连接
       public void getConnect() throws IOException {
           zk = new ZooKeeper(CONNECT_STRING, SESSION_TIMEOUT, new Watcher() {
               @Override
               public void process(WatchedEvent event) {} 
           });
       }
   }
   

3. 设置监听的客户端

   public class DistributeClient {
       private static final String CONNECT_STRING = "192.168.59.110:2181,192.168.59.110:2182,192.168.59.110:2183";
       private static final Integer SESSION_TIMEOUT = 200000;
       private ZooKeeper zk;
       private static final String PATH_NODE = "/servers";
   
       public static void main(String[] args) throws InterruptedException, KeeperException, IOException {
           DistributeClient client = new DistributeClient();
           // 获取链接
           client.getConnect();
           // 获取监听的服务器列表
           client.getServerList();
           // 启动业务逻辑（睡觉）
           Thread.sleep(Integer.MAX_VALUE);
       }
   
       private void getServerList() throws KeeperException, InterruptedException {
           List<String> children = zk.getChildren(PATH_NODE, true);
           ArrayList<String> servers = new ArrayList<>();
           children.forEach(chi -> {
               try {
                   byte[] data = zk.getData(PATH_NODE + "/" + chi, false, null);
                   servers.add(new String(data));
               } catch (Exception e) {
                   e.printStackTrace();
               }
           });
           System.out.println(servers);
       }
   
       // 获取zk连接
       public void getConnect() throws IOException {
           zk = new ZooKeeper(CONNECT_STRING, SESSION_TIMEOUT, new Watcher() {
               @Override
               public void process(WatchedEvent event) {
                   // 再次获取集合
                   try {
                       getServerList();
                   } catch (Exception e) {
                       e.printStackTrace();
                   }
               }
           });
       }
   }
   

4. 测试

   1. 先启动 DistributeClient 客户端
   2. 在linux上的/servers目录上创建临时带序号节点
   3. 观察IDEA控制台变化
   4. 启动 DistributeServer
   5. 观察IDEA控制台变化
   6. 在IDEA减少服务器
   7. 观察控制台变化

案例2：Zookeeper分布式锁

分布式锁概念

比如说"进程 1"在使用该资源的时候，会先去获得锁，"进程 1"获得锁以后会对该资源保持独占，这样其他进程就无法访问该资源，"进程 1"用完该资源以后就将锁释放掉，让其他进程来获得锁，那么通过这个锁机制，我们就能保证了分布式系统中多个进程能够有序的访问该临界资源。那么我们把这个分布式环境下的这个锁叫作分布式锁。

分布式锁案例分析

原生Zookeeper实现分布式锁

Lock类

public class DistributeLock {
    private static final String CONNECT_STRING = "192.168.59.110:2181,192.168.59.110:2182,192.168.59.110:2183";
    private static final Integer SESSION_TIMEOUT = 200000;
    private static final String LOCK_PATH = "/locks";
    private final ZooKeeper zk;
    private CountDownLatch connLatch = new CountDownLatch(1);
    private CountDownLatch waitLatch = new CountDownLatch(1);
    private String waitPath;
    private String currentNode;

    public DistributeLock() throws IOException, InterruptedException, KeeperException {
        // 1,初始化Zookeeper连接
        zk = new ZooKeeper(CONNECT_STRING, SESSION_TIMEOUT, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                // 判断zk有没有连接成功
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    connLatch.countDown(); // 释放资源
                }
                // 释放锁条件：waitPath触发删除操作,而且正好是前一个节点
                if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) {
                    waitLatch.countDown(); // 释放资源
                }
            }
        });
        connLatch.await();
        // 2,判断locks节点是否存在，不存在则创建
        Stat exists = zk.exists(LOCK_PATH, true);
        // 3,不存在就创建
        if (exists == null) {
            zk.create(LOCK_PATH, "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    public void zkLock() throws KeeperException, InterruptedException {
        // 在locks下创建当前线程的临时节点
        currentNode = zk.create(LOCK_PATH + "/seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        // 获取当前所有节点
        List<String> children = zk.getChildren(LOCK_PATH, false);
        // 如果只有一个节点，肯定是最小
        if (children.size() == 1) {
            return;  // 获取锁成功
        } else { // 不最小节点，
            Collections.sort(children);
            // 获取当前节点
            String seqString = currentNode.replace("/locks/", "");
            // 判断该节点是否最小，是最小则获取锁
            int index = children.indexOf(seqString);
            // 判断索引位置
            if (index == -1) {
                System.out.println("数据异常!!!");
            } else if (index == 0){ // 是第一个节点
                return;
            } else {
                // 不是第一个节点则需要监听前一个节点的状态
                waitPath = "/locks/" + children.get(index - 1);
                // 获取前一节点状态 (需要监听)
                zk.getData(waitPath, true, null);
                // 等待
                waitLatch.await();
                // 获取到锁了
                return;
            }
        }
    }

    public void zkUnLock() throws KeeperException, InterruptedException {
        // 删除当前临时节点
        zk.delete(currentNode, -1);
    }
}

测试类

public class DistributeLockTest {
    public static void main(String[] args) throws InterruptedException, IOException, KeeperException {
        final DistributeLock distributeLock1 = new DistributeLock();
        final DistributeLock distributeLock2 = new DistributeLock();

        new Thread(() -> {
            try {
                distributeLock1.zkLock();
                System.out.println("线程1获取到了锁...");
                Thread.sleep(10*1000);
                distributeLock1.zkUnLock();
                System.out.println("线程1释放了锁...");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();

        new Thread(() -> {
            try {
                distributeLock2.zkLock();
                System.out.println("线程2获取到了锁...");
                Thread.sleep(10*1000);
                distributeLock2.zkUnLock();
                System.out.println("线程2释放了锁...");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
    }
}

Curator框架实现分布式锁

原生Java API开发存在的问题

1. 会话的连接是异步的，需要自己去处理，比如使用CountDownLatch
2. Watch需要重复注册，不然就不能生效
3. 开发的复杂性还是比较高的
4. 不支持多点删除和创建，需要自己递归

Curator 案例实操

添加maven依赖

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>4.3.0</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>4.3.0</version>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-client</artifactId>
    <version>4.3.0</version>
</dependency>

代码实现案例

public class CuratorTest {

    private static final String CONNECT_STRING = "192.168.59.110:2181,192.168.59.110:2182,192.168.59.110:2183";

    private static final String LOCK_NODE = "/locks";

    public static void main(String[] args) {
        // 创建分布式锁 1
        final InterProcessLock lock1 = new
                InterProcessMutex(getCuratorFramework(), LOCK_NODE);
        // 创建分布式锁 2
        final InterProcessLock lock2 = new
                InterProcessMutex(getCuratorFramework(), LOCK_NODE);

        new Thread(() -> {
            try {
                lock1.acquire();
                System.out.println("线程1获取到了锁...");
                Thread.sleep(10*1000);
                lock1.release();
                System.out.println("线程1释放了锁...");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();

        new Thread(() -> {
            try {
                lock2.acquire();
                System.out.println("线程2获取到了锁...");
                Thread.sleep(10*1000);
                lock2.release();
                System.out.println("线程2释放了锁...");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();

    }

    /**
     * 获取 Curator 分布式锁
     */
    private static CuratorFramework getCuratorFramework() {
        ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000, 3);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(CONNECT_STRING)
                .connectionTimeoutMs(20000)
                .sessionTimeoutMs(20000)
                .retryPolicy(policy).build();
        // 开启连接
        client.start();
        System.out.println("zookeeper 初始化完成...");
        return client;
    }
}

原理、应用相关

算法

Paxos算法

介绍

Paxos算法：一种基于消息传递且具有高度容错特性的一致性算法

解决的问题：就是如何快速正确的在一个分布式系统中对某个数据值达成一致，且保证无论发生任何异常，都不会破坏整个系统的一致性。（异常可能包括机器宕机、网络异常等）

Paxos算法描述

• 在一个Paxos系统中，首先将所有节点划分为Proposer（提议者），Acceptor（接受者）和Learner（学习者）。注意，每个节点都可以身兼数职。
  

• 一个完整的Paxos算法流程分为三个阶段：

  1. Prepare准备阶段

     1. Proposer向多个Acceptor发出Propose请求Promise（承诺）
     2. Acceptor针对收到的Peopose请求进行Promise（许诺）

  2. Accept接受阶段

     1. Proposer收到多数Acceptor承诺的Promise后，向Acceptor发出Propose请求
     2. Accrptor针对收到的Propose请求进行Accept处理

  3. Learn学习阶段

     Proposer将形成的决议发送给所有Learner

Paxos算法流程

1. Prepare：Proposer生成全局唯一递增的Proposal ID（可以理解为事务ID），向所有Acceptor发送Propose请求，这里无需携带提案内容，只携带Proposal ID即可。
2. Promise：Acceptor收到Propose请求后，做出"两个承诺，一个应答"
   1. 不再接受Proposal ID小于等于当前请求的Propose请求
   2. 不再接受Proposal ID小于当前的Accept请求
   3. 不违背以前做出的承诺下，回复已经Accept过的提案中Proposal ID最大的那个提案的Value和Proposal ID，没有则返回空值。
3. Propose：Proposer收到多数Acceptor的Promise应答后，从应答中选择Proposal ID最大的提案的value，作为本次要发起的提案。如果所有应答的提案value均为空值，则可以自己随意决定提案Value。然后携带当前Proposal ID，向所有Acceptor发送Propose请求。
4. Accept：Acceptor收到Propose请求后，在不违背自己之前作出的承诺下，接受并持久化当前Proposal ID和提案value
5. Learn：Proposer收到多数Acceptor的Accept后，决议形成，将形成的决议发送给所有Learner。

三种情况

情况1：

情况2：

情况3(多个提案者活锁)：

Zab（Zookeeper Atomic Broadcast）协议

什么是ZAB算法

Zab借鉴了Paxos算法，是特别为Zookeeper设计的支持崩溃恢复的原子广播协议。基于该协议，Zookeeper设计为只有一台客户端（Leader）负责处理外部的写事务请求，然后Leader客户端将数据同步到其他Follower节点。即Zookeeper只有一个Leader可以发起提案。

消息广播

1. 客户端发起一个写操作请求
2. Leader服务器将客户端的请求转化为事务Proposal提案，同事为每个Proposal分配一个全局的ID，即zxid
3. Leader服务器为每个Follower服务器分配一个单独的队列，然后将要广播的Proposal依次放到队列中去，并且根据FIFO策略进行消息发送
4. Follow接收到Proposal后，会首先将其以事务日志的方式写入本地磁盘中，写入成功后想Leader反馈一个Ack响应消息
5. Leader接收到超过半数以上的Follow的Ack响应消息后，即认为消息发送成功，可以发送commit消息
6. Leader向所有Follower广播commit消息，同时自身也会完成事务提交。Follower接收到commit消息后，会将上一条事务提交
7. Zookeeper采用Zab协议的核心，就是只要有一台服务器提交了Proposal，就要确保所有的服务器最终都能正确提交Proposal
   

Zab协议针对事务的请求处理过程类似于一个两阶段提交过程：

1. 广播事务阶段
2. 广播提交阶段

这两阶段提交，有可能因为Leader宕机带来数据不一致，比如：

1. Leader发起一个事务Proposal后就宕机，Follow都没有Proposal
2. Leader收到半数ACK宕机，没来得及向Follower发送commit

崩溃恢复

一旦Leader服务器出现崩溃或者由于网络原因导致Leader服务器失去了与过半Follower的联系，那么就会进入崩溃恢复模式。

1. 假设两种服务器异常情况
   1. 假设一个事务在Leader提出之后Leader挂了
   2. 一个事务在Leader上提交了，并且过半的Follower都响应ack了，但是Leader在Commit消息发出之前挂了
2. Zab协议崩溃恢复要满足以下两个要求
   1. 确保已经被Leader提交的提案Proposal，必须最终被所有的Follower服务器提交。（已经产生的提案，Follower必须执行）
   2. 确保丢弃已经被Leader提出的，但是没有被提交的Proposal。（丢弃胎死腹中的提案）

崩溃恢复主要包括两部分：Leader选举和数据恢复

Leader选举：

1. 新选出来的Leader不能包含未提交的Proposal。即新Leader必须都是已经提交了Proposal的Follower服务器节点
2. 新选举的Leader节点中含有最大的zxid。这样做的好处是可以避免Leader服务器检查Proposal的提交和丢弃工作。

数据恢复

1. 完成Leader选举后，在正式开始工作之前（接收事务请求，然后提出新的Proposal），Leader服务器会首先确认事务日志中所有的Proposal是否已经被集群中过半的服务器Commit。
2. Leader服务器需要确保所有的Follower服务器能够接收到每一条事务的Proposal，并且能将所有已经提交的事务Proposal应用到内存数据中。等到Follower将所有尚未同步的事务Proposal都从Leader服务器上同步过，并且应用到内存数据中以后，Leader才会把该Follow加入到真正可用的Follower列表中。

异常提案处理

Zab数据同步过程中，如何处理需要丢弃的Proposal？

在Zab的事务编号zxid设计中，zxid是一个64位的数字。其中低32位可以看成一个简单的单增计数器，针对客户端每一个事务请求，Leader在产生新的Proposal事务时，都会对该计数器加1。而高32位则代表了Leader周期的epoch编号。
epoch编号可以理解为当前集群所处的年代，或者周期。每次Leader变更之后都会在 epoch的基础上加1，这样旧的Leader崩溃恢复之后，其他Follower也不会听它的了，因为 Follower只服从epoch最高的Leader命令。每当选举产生一个新的 Leader，就会从这个Leader服务器上取出本地事务日志充最大编号Proposal的zxid，并从zxid中解析得到对应的epoch编号，然后再对其加1，之后该编号就作为新的epoch 值，并将低32位数字归零，由0开始重新生成zxid。Zab协议通过epoch编号来区分Leader变化周期，能够有效避免不同的Leader错误的使用了相同的zxid编号提出了不一样的Proposal的异常情况。基于以上策略，当一个包含了上一个Leader周期中尚未提交过的事务Proposal的服务器启动时，当这台机器加入集群中， 以Follower角色连上Leader服务器后，Leader 服务器会根据自己服务器上最后提交的 Proposal来和Follower服务器的Proposal进行比对，比对的结果肯定是Leader要求Follower进行一个回退操作，回退到一个确实已经被集群中过半机器Commit的最新Proposal。

CAP理论

CAP理论告诉我们，一个分布式系统不可能同时满足以下三种

一致性（C:Consistency） 可用性（A:Available） 分区容错性（P:Partition Tolerance）
这三个基本需求，最多只能同时满足其中的两项，因为P是必须的，因此往往选择就在CP或者AP中。

1）一致性（C:Consistency）
在分布式环境中，一致性是指数据在多个副本之间是否能够保持数据一致的特性。在一致性的需求下，当一个系统在数据一致的状态下执行更新操作后，应该保证系统的数据仍然处于一致的状态。

2）可用性（A:Available）
可用性是指系统提供的服务必须一直处于可用的状态，对于用户的每一个操作请求总是能够在有限的时间内返回结果。

3）分区容错性（P:Partition Tolerance）
分布式系统在遇到任何网络分区故障的时候，仍然需要能够保证对外提供满足一致性和可用性的服务，除非是整个网络环境都发生了故障。

ZooKeeper保证的是CP

1）ZooKeeper不能保证每次服务请求的可用性。（注：在极端环境下，ZooKeeper可能会丢弃一些请求，消费者程序需要重新请求才能获得结果）。所以说，ZooKeeper不能保证服务可用性。

2）进行Leader选举时集群都是不可用。

源码

未更新…

参考链接：https://www.bilibili.com/video/BV1to4y1C7gw?from=search&seid=13243157056880113151