ZooKeeper 高级应用

        更好的阅读体验 
       
      
     
    
   
     \huge{\color{red}{更好的阅读体验}} 
    
   
 更好的阅读体验

概述

ZooKeeper 是 Apache 软件基金会的一个软件项目，它为大型分布式计算提供开源的分布式配置服务、同步服务和命名注册，在架构上，通过冗余服务实现高可用性（CP）。

ZooKeeper 的设计目标是将那些复杂且容易出错的分布式一致性服务封装起来，构成一个高效可靠的原语集，并以一系列简单易用的接口提供给用户使用。

基础回顾

数据结构

ZooKeeper 本身是一个树形目录服务（名称空间），非常类似于标准文件系统，

key-value

的形式存储。名称

key

由斜线

/

分
割的一系列路径元素，例如：

/node

，ZooKeeper 名称空间中的每个节点都是由一个路径来标识的。

注意：

• 每个路径下的节点 key （完整路径，名称）是唯一的，即同一级节点 key 名称是唯一的
• 每个节点中存储了节点 value 和对应的状态属性，其中属性可能有多个

节点类型：

• PERSISTENT：默认创建节点的类型，持久的节点
• PERSISTENT_SEQUENTIAL：持久顺序节点，会在路径后加上单调递增的后缀，适用于分布式锁和分布式选举，创建时添加 -s 参数
• EPHEMERAL：临时节点（不可拥有子节点），和会话绑定，断开服务后自动失效，创建时添加 -e 参数
• EPHEMERAL_SEQUENTIAL：临时顺序节点（不可在拥有子节点），会加上后缀，会话断开后删除，创建时添加 -e -s 参数
• CONTAINER：容器节点，当子节点都被删除后，Container 也随即删除，创建时添加 -c 参数
• PERSISTENT_WITH_TTL：如果该节点在 TTL 内没有被修改或没有子节点则过期删除，创建时添加 -t 参数

基础操作

节点操作的基础命令：

• ls：查看某个路径下目录列表，可选参数 -s 返回状态信息， -w 监听节点变化，-R 递归查看某路径下目录列表
• create：创建节点并赋值，可选参数和节点的类型相照应，注意临时节点不能创建子节点
• set：修改节点存储的数据
• get：获取节点数据和状态信息，可选参数 -s 返回状态信息， -w 返回数据并对对节点进行事件监听
• stat：查看节点状态信息，也可选 -w 参数
• delete/deleteall：删除某节点，如果某节点不为空，则不能用 delete 命令删除

注意：-w 监听节点只能生效一次，在节点信息变化后返回变化信息并失效

分布式锁

原理实现

ZooKeeper 实现简单的分布式锁：

• 注册临时节点，谁注册成功谁获取锁，其他监听该节点的删除事件
• 一旦该临时节点被删除，通知其他客户端，再次重复该流程

但是上述方式存在问题——羊群效应：

• 当临时节点释放时，会通知到所有监听该节点的服务
• 多个服务又会同时发起重新注册的请求，导致 ZooKeeper 服务压力较大

高级实现

为了解决上面产生的问题，我们给出更为完善的方案：

• 所有服务注册临时顺序节点，并写入基本信息
• 所有服务获取节点列表并判断自己的节点是否是最小的那个，如果是说明获取到了锁
• 未获取锁的客户端添加对前一个节点删除事件的监听
• 锁释放/持有锁的客户端宕机 后，节点被删除，下一个节点的客户端收到通知，重复上述流程

基于上述解决方案，我们再将临时顺序节点的创建进行细分，分为分为读锁节点和写锁节点：

• 对于读锁节点而言，其只需要关心前一个写锁节点的释放
• 对于写锁节点而言，其只需要关心前一个节点的释放，而不需要关心前一个节点是写锁节点还是读锁节点

这样就基于 ZooKeeper 实现了共享锁和排他锁，在使用时，我们一般利用 Curator 客户端实现：

importorg.apache.curator.RetryPolicy;importorg.apache.curator.framework.CuratorFramework;importorg.apache.curator.framework.CuratorFrameworkFactory;importorg.apache.curator.framework.recipes.locks.InterProcessLock;importorg.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex;importorg.apache.curator.retry.ExponentialBackoffRetry;importorg.apache.curator.utils.CloseableUtils;importorg.junit.After;importorg.junit.Before;importorg.junit.Test;publicclassDistributedLockDemo{// ZooKeeper 锁节点路径, 分布式锁的相关操作都是在这个节点上进行privatefinalString lockPath ="/distributed-lock";// ZooKeeper 服务地址, 单机格式为:(127.0.0.1:2181),// 集群格式为:(127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183)privateString connectString;// Curator 客户端重试策略privateRetryPolicy retry;// Curator 客户端对象privateCuratorFramework client;// client2 用户模拟其他客户端privateCuratorFramework client2;// 初始化资源@Beforepublicvoidinit()throwsException{// 设置 ZooKeeper 服务地址为本机的 2181 端口
        connectString ="192.168.200.168:2181";// 重试策略// 初始休眠时间为 1000ms, 最大重试次数为 3
        retry =newExponentialBackoffRetry(1000,3);// 创建一个客户端, 60000(ms)为 session 超时时间, 15000(ms)为链接超时时间
        client =CuratorFrameworkFactory.newClient(connectString,60000,15000, retry);
        client2 =CuratorFrameworkFactory.newClient(connectString,60000,15000, retry);// 创建会话
        client.start();
        client2.start();}// 释放资源@Afterpublicvoidclose(){CloseableUtils.closeQuietly(client);}@TestpublicvoidsharedLock()throwsException{// 创建共享锁finalInterProcessLock lock =newInterProcessSemaphoreMutex(client, lockPath);// lock2 用于模拟其他客户端finalInterProcessLock lock2 =newInterProcessSemaphoreMutex(client2, lockPath);newThread(newRunnable(){@Overridepublicvoidrun(){// 获取锁对象try{
                    lock.acquire();System.out.println("======== client1 get lock ========");// 测试锁重入Thread.sleep(5*1000);
                    lock.release();System.out.println("======== client1 release lock ========");}catch(Exception e){
                    e.printStackTrace();}}}).start();newThread(newRunnable(){@Overridepublicvoidrun(){// 获取锁对象try{
                    lock2.acquire();System.out.println("======== client2 get lock ========");Thread.sleep(5*1000);
                    lock2.release();System.out.println("======== client2 release lock ========");}catch(Exception e){
                    e.printStackTrace();}}}).start();Thread.sleep(20*1000);}}

• InterProcessMutex：分布式可重入排它锁（可重入可以借助 LocalMap 存计数器）
• InterProcessSemaphoreMutex：分布式排它锁
• InterProcessMultiLock：将多个锁作为单个实体管理的容器
• InterProcessReadWriteLock：分布式读写锁

集群应用

集群节点配置

对于搭建 ZooKeeper 集群的节点往往采用奇数个：

• 保证容错：需要保证集群能够有半数进行投票，例如： - 三台集群，至少 2 台正常运行才行（3的半数为 1.5，半数以上最少为 2）- 因此，正常运行可以允许 1 台服务器挂掉
• 防止脑裂：需要保证集群在通信不可达的情况下分裂产生小集群，例如： - 3 台集群，投票选举半数为 1.5，1 台服务裂开，和另外 2 台服务器无法通信- 这时候 2 台服务器的集群（2票大于半数1.5票），所以可以选举出leader，而 1 台服务器的集群无法选举

综上可知，搭建集群所需的最少节点配置为 3，如果是 4 台，则发生脑裂时会造成没有 leader 节点的错误。

选举算法

ZooKeeper 采用的是基于 Paxos 算法的 ZAB 协议，这里先提一下 Paxos 算法：

• Paxos 是一个分布式选举算法，该算法定义了三种角色
• Proposer：提案发起者
• Acceptor：提案接收者，可同意或不同意
• Learners：虽然不同意提案，但也只能被动接收学习；或者是后来的，只能被动接受
• 该算法的提案遵循少数服从多数的原则，即过半原则

ZAB 协议在 Paxos 算法基础上进行了扩展，全称为原子消息广播协议（ZooKeeper Atomic Broadcast）：

• ZAB 协议支持原子广播、崩溃恢复，保证 Leader 广播的变更序列被顺序的处理，该协议下的节点有四种状态
• LOOKING：系统刚启动时或者Leader崩溃后正处于选举状态
• FOLLOWING：表示 Follower 节点所处的状态，同步leader状态，参与投票
• LEADING：表示 Leader 所处状态
• OBSERVING：观察状态，同步leader状态，不参与投票
• 该算法下，也遵循半原则

我们查看 ZooKeeper 的源码，在

FastLeaderElection.java

中：

protectedbooleantotalOrderPredicate(long newId,long newZxid,long newEpoch,long curId,long curZxid,long curEpoch){LOG.debug("id: {}, proposed id: {}, zxid: 0x{}, proposed zxid: 0x{}",
            newId,
            curId,Long.toHexString(newZxid),Long.toHexString(curZxid));if(self.getQuorumVerifier().getWeight(newId)==0){returnfalse;}/*
         * We return true if one of the following three cases hold:
         * 1- New epoch is higher
         * 2- New epoch is the same as current epoch, but new zxid is higher
         * 3- New epoch is the same as current epoch, new zxid is the same
         *  as current zxid, but server id is higher.
         *//*
         * 对应上面代码的解释（两个节点之间使用比较的方法来决定选票给谁，三种比较规则）
         * 1- 比较 epoche(zxid高32bit):
         *     如果其他节点的epoche比自己的大，选举 epoch大的节点（理由：epoch 表示年代【投票次数越多，数据越新】，epoch越大表示数据越新）
         *     代码：(newEpoch > curEpoch)；
         * 2- 比较 zxid，:
         *     如果纪元相同，就比较两个节点的zxid的大小，选举 zxid大的节点（理由：zxid 表示节点所提交事务最大的id，zxid越大代表该节点的数据越完整）
         *     代码：(newEpoch == curEpoch) && (newZxid > curZxid)；
         * 3- 比较 serviceId:
         *     如果 epoch和zxid都相等，就比较服务的serverId，选举 serviceId大的节点（理由： serviceId 表示机器性能，他是在配置zookeeper集群时确定的，所以我们配置zookeeper集群的时候可以把服务性能更高的集群的serverId设置大些，让性能好的机器担任leader角色）
         *     代码 ：(newEpoch == curEpoch) && ((newZxid == curZxid) && (newId > curId))。
         */return((newEpoch > curEpoch)||((newEpoch == curEpoch)&&((newZxid > curZxid)||((newZxid == curZxid)&&(newId > curId)))));}

集群数据读写

读请求：

• 当 Client 向 ZooKeeper 发出读请求时
• 无论是 Leader 还是 Follower，都直接返回查询结果

写请求-Leader：

• Client 向 Leader 发出写请求时
• Leader 将数据写入到本节点，并将数据发送到所有的 Follower 节点，等待 Follower 节点返回
• 当 Leader 接收到一半以上节点（包含自己）返回写成功的信息之后，直接向 Client 返回成功

写请求-Follower：

• Client 向 Follower 发出写请求时
• Follower 节点将请求转发给 Leader
• Leader 将数据写入到本节点，并将数据发送到所有的 Follower 节点，等待 Follower 节点返回
• 当 Leader 接收到一半以上节点（包含自己）返回写成功的信息之后，直接向转发请求的 Follower 返回成功
• Follower 接收到 Leader 请求后向 Client 返回成功