zookeeper 笔记--curator分布式锁

使用ZK实现分布式独占锁,

原理就是利用ZK同级节点的唯一性.

Curator框架下的一些分布式锁工具
InterProcessMutex：分布式可重入排它锁

InterProcessSemaphoreMutex：分布式排它锁

InterProcessReadWriteLock：分布式读写锁

---

使用 Zookeeper 实现 leader 选举

Leader Latch

参与选举的所有节点，会创建一个顺序节点，其中最小的 节点会设置为 master 节点, 没抢到 Leader 的节点都监听前一个节点的删除事件，在前一个节点删除后进行重新抢主，当 master 节点手动调用 close（）方法或者 master 节点挂了之后，后续的子节点会抢占 master。 其中 spark 使用的就是这种方法

LeaderSelector

LeaderSelector 和 Leader Latch 最的差别在于，leader 可以释放领导权以后，还可以继续参与竞争

public class LeaderSelectorClient extends LeaderSelectorListenerAdapter implements Closeable {

    private  String name;  //表示当前的进程
    private  LeaderSelector leaderSelector;  //leader选举的API
    private CountDownLatch countDownLatch=new CountDownLatch(1);

    public LeaderSelectorClient(){

    }

    public LeaderSelectorClient(String name) {
        this.name = name;
    }

    public LeaderSelector getLeaderSelector() {
        return leaderSelector;
    }

    public void setLeaderSelector(LeaderSelector leaderSelector) {
        this.leaderSelector = leaderSelector;
    }

    public void start(){
        leaderSelector.start(); //开始竞争leader
    }

    @Override
    public void takeLeadership(CuratorFramework client) throws Exception {
        //如果进入当前的方法，意味着当前的进程获得了锁。获得锁以后，这个方法会被回调
        //这个方法执行结束之后，表示释放leader权限
        System.out.println(name+"->现在是leader了");
//        countDownLatch.await(); //阻塞当前的进程防止leader丢失
    }

    @Override
    public void close() throws IOException {
        leaderSelector.close();
    }
    private static String CONNECTION_STR="x.x.x.x:2181;

    public static void main(String[] args) throws IOException {
        CuratorFramework curatorFramework = CuratorFrameworkFactory.builder().
                connectString(CONNECTION_STR).sessionTimeoutMs(50000000).
                retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
        curatorFramework.start();
        LeaderSelectorClient leaderSelectorClient=new LeaderSelectorClient("ClientA");
        LeaderSelector leaderSelector=new LeaderSelector(curatorFramework,"/leader",leaderSelectorClient);
        leaderSelectorClient.setLeaderSelector(leaderSelector);
        leaderSelectorClient.start(); //开始选举
        System.in.read();
    }
}

---

ZAB 协议介绍

ZAB 协议包含两种基本模式，

1. 崩溃恢复[master节点选择]

2. 原子广播[数据恢复,数据同步]

当整个集群在启动时，或者当 leader 节点出现网络中断、 崩溃等情况时，ZAB 协议就会进入恢复模式并选举产生新 的 Leader，当 leader 服务器选举出来后，并且集群中有过 半的机器和该 leader 节点完成数据同步后（同步指的是数 据同步，用来保证集群中过半的机器能够和 leader 服务器 的数据状态保持一致），ZAB 协议就会退出恢复模式。

当集群中已经有过半的 Follower 节点完成了和 Leader 状 态同步以后，那么整个集群就进入了消息广播模式。这个时候，在 Leader 节点正常工作时，启动一台新的服务器加入到集群，那这个服务器会直接进入数据恢复模式，和 leader 节点进行数据同步。同步完成后即可正常对外提供非事务请求的处理。

需要注意的是：leader 节点可以处理事务请求和非事务请 求，follower 节点只能处理非事务请求，如果 follower 节 点接收到非事务请求，会把这个请求转发给 Leader 服务器

ZK数据同步流程

ZK数据同步流程过程实际上是一个 简化版本的二阶段(2pc)提交过程

1. leader 接收到消息请求后，将消息赋予一个全局唯一的 64 位自增 id，叫：zxid，通过 zxid 的大小比较既可以实现因果有序这个特征

2. leader 为每个 follower 准备了一个 FIFO 队列（通过 TCP 协议来实现，以实现了全局有序这一个特点）将带有 zxid 的消息作为一个提案（proposal）分发给所有的 follower

3. 当 follower 接收到 proposal，先把 proposal 写到磁盘， 写入成功以后再向 leader 回复一个 ack

4. 当 leader 接收到合法数量（超过半数节点）的 ACK 后，leader 就会向这些 follower 发送 commit 命令，同时会 在本地执行该消息

5. 当 follower 收到消息的 commit 命令以后，会提交该消息

 

---

关于 ZXID

为了保证事务的顺序一致性，zookeeper 采用了递增的事 务 id 号（zxid）来标识事务。

所有的提议（proposal）都 在被提出的时候加上了 zxid。实现中 zxid 是一个 64 位的 数字，它高 32 位是 epoch（ZAB 协议通过 epoch 编号来 区分 Leader 周期变化的策略）用来标识 leader 关系是否改变，每次一个 leader 被选出来，它都会有一个新的 epoch=（原来的 epoch+1），标识当前属于那个 leader 的 统治时期。低 32 位用于递增计数。

epoch：可以理解为当前集群所处的年代或者周期，每个leader 就像皇帝，都有自己的年号，所以每次改朝换代，leader 变更之后，都会在前一个年代的基础上加1。这样就算旧的 leader 崩溃恢复之后，也没有人听他的了，因为 follower 只听从当前年代的 leader 的命令。