zookeeper主要用于解决分布式环境下的服务协调问题，通常应用场景为：

注册中心: dubbo、motan等

配置中心：disconf

负载均衡：节点值可以为多个服务地址，根据不同负载策略选取服务地址

分布式锁

集群管理工具：根据节点排序用来选举master等

znode

zookeeper结构

zoo_

zookeeper可以理解为一个文件树结构，每一个节点称为一个znode，每个znode上都能存储自己的信息。

znode类型

持久节点和临时节点

znode节点可以是持久（persistent）节点，还可以是临时（ephemeral）节点。持久的znode，如/path，只能通过调用delete来进行删除。临时的znode与之相反，当创建该节点的客户端崩溃或关闭了与ZooKeeper的连接时，这个节点就会被删除。

持久znode是一种非常有用的znode，可以通过持久类型的znode为应用保存一些数据，即使znode的创建者不再属于应用系统时，数据也可以保存下来而不丢失。例如，在主-从模式例子中，需要保存从节点的任务分配情况，即使分配任务的主节点已经崩溃了。

一个临时znode，在以下两种情况下将会被删除：

1.当创建该znode的客户端的会话因超时或主动关闭而中止时。

2.当某个客户端（不一定是创建者）主动删除该节点时。

因为临时的znode在其创建者的会话过期时被删除，所以目前不允许临时节点拥有子节点。

有序节点

一个znode还可以设置为有序（sequential）节点。一个有序znode节点被分配唯一个单调递增的整数。当创建有序节点时，一个序号会被追加到路径之后，例如/tasks/task-1。

有序znode通过提供了创建具有唯一名称的znode的简单方式。同时也通过这种方式可以直观地查看znode的创建顺序。

总之，znode一共有4种类型：持久节点、临时节点、持久有序节点、临时有序节点。

Watch机制

zookeeper中，client获取sever信息的方式是基于通知（notification）的机制：

客户通过对znode设置监视点（watch）来接收通知。监视点是一个单次触发的操作，意即监视点会触发一个通知。为了接收多个通知，客户端必须在每次通知后设置一个新的监视点。

如图，当节点/tasks发生变化时，客户端会收到一个通知，并从ZooKeeper读取一个新值。

clipboard

zookeeper集群

集群角色

Leader

Leader服务器是整个zookeeper集群的核心，主要的工作任务为：

1. 事物请求的唯一调度和处理者，保证集群事物处理的顺序性

2. 集群内部各服务器的调度者

Follower

Follower 角色的主要职责是：

1. 处理客户端非事务请求、转发事务请求给 leader 服务器

2. 参与事务请求 Proposal 的投票

3. 参与 Leader 选举的投票

Observer

Observer是观察者的角色，观察zookeeper集群中的最新状态变化并将这些状态变化同步到observer服务器上。

Observer的工作原理与follower角色基本一致，而它和follower角色唯一的不同在于observer不参与任何形式的投票。

集群事务处理

集群中，客户端会随机连接到集群中一个节点，如果是读请求，就直接从当前节点中读取数据，如果是写请求，那么请求会被转发给 leader 提交事务。

leader把客户端的事务请求转化成一个事务 Proposal（提议），并把这个 Proposal 分发给集群中的所有 Follower 服务器。

之后 Leader 服务器需要等待所有Follower服务器的反馈，一旦超过半数的Follower服务器进行了正确的反馈，那么Leader就会再次向所有的Follower服务器发送Commit消息，要求各个follower节点对前面的一个 Proposal 进行提交。

这个事务处理过程是一个类2PC过程。

2PC

两阶段提交（2 phase commit），属于分布式事务，事务成功操作需要两阶段执行。

2pc

阶段一：提交事务请求

1. 事务询问

协调者向所有的参与者发送事务内容，询问是否可以执行事务提交操作，并开始等待各参与者的响应。

2. 执行事务

各个参与者节点执行事务操作，并将Undo和Redo信息记录到事务日志中，尽量把提交过程中所有消耗时间的操作和准备都提前完成，确保后面100%成功提交事务。

3. 各个参与者向协调者反馈事务询问的响应

如果各个参与者成功执行了事务操作，那么就反馈给参与者yes的响应，表示事务可以执行；如果参与者没有成功执行事务，就反馈给协调者no的响应，表示事务不可以执行。

这个阶段有点类似协调者组织各个参与者对一次事务操作的投票表态过程，因此2pc协议的第一个阶段称为“投票阶段”，即各参与者投票表名是否需要继续执行接下去的事务提交操作。

阶段二：事务提交

在这个阶段，协调者会根据各参与者的反馈情况来决定最终是否可以进行事务提交操作：

如果所有参与者返回yes，调用confirm接口进行commit

如果有返回no，则调用cancel接口进行rollback

zookeeper的2pc是简化的2pc，因为有过半的参与者返回yes的话，协调者就会发起commit操作，zookeeper中leader也是一个执行事务的参与者。

ZAB 协议

ZooKeeper主要依赖原子广播（Zookeeper Atomic Broadcast）协议来实现分布式数据一致性。ZAB协议包含崩溃恢复和消息广播两种基本模式。

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

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

消息广播

消息广播就是zookeeper实现的2PC。

1. leader 接收到消息请求后，会为消息赋予一个全局唯一的64位自增id，即zxid，并将该消息放入一个FIFO队列保证消息执行的有序性。

2. 将带有 zxid的消息作为一个提案（proposal）分发给所有的follower

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

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

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

崩溃恢复

当Leader 节点崩溃或由于网络问题导致Leader服务器失去了与过半的Follower节点的联系，那么此时的leader不再是合法的leader了，就会进入到崩溃恢复模式。为了保证程序正确运行，整个恢复过程结束后要选举出一个新的Leader。

需要解决两个问题:

1. 已经被处理的消息不能丢失

当leader收到合法数量follower的ACK后，向各个follower广播commit命令。但是如果未在所有follower收到commit命令前leader挂了，导致有一部分服务器没有执行commit。

新的leader选举后要保证这样的消息不能丢失。

2. 被丢弃的消息不能再次出现

当leader接收到消息请求生成proposal后就挂了，其他follower并没有收到此proposal，因此经过恢复模式重新选了leader后，这条消息是会被跳过的。如果此时之前挂了的leader重新启动并注册成了follower，它必须删除这些过时的消息。

ZXID

zxid是事务id，zookeeper采用了递增的事务id来标识事务。所有的提议（proposal）在被提出时都被加上了zxid。zxid是一个64位的数字，高32位是epoch，每次leader被选出来，都会有一个新的epoch（原来的epoch+1）；低 32 位用于递增计数。

启动zookeeper集群，在$dataDir/VERSION-2路径下有currentEpoch文件，文件中记录的是当前的 epoch。当leader节点停机，新的leader节点会被选举出来，这时候currentEpoch会有变化。

leader 选举

Leader 选举分为启动时leader选举、运行时选举。

启动时选举

每个节点启动的时候状态都处于LOOKING状态，集群初始化阶段，当一台服务器Server1启动时，它本身是无法进行和Leader选举的，当第二台服务器Server2启动时，这个时候两台机器可以相互通信，于是进入 Leader 选举过程。

(1) 每个 Server 发出一个投票。由于是初始情况，Server1和 Server2都会将自己作为Leader服务器来进行投票，每次投票会包含所推举的服务器的myid和ZXID、epoch，然后各自将票发给集群中其他机器。

(2) 集群的每个服务器收到投票后，首先判断该投票的有效性，如检查是否是本轮投票（epoch）、是否来自LOOKING状态的服务器。

(3) 针对每一个投票，服务器都需要将别人的投票和自己的投票进行比对，比对规则如下：

i. 优先检查 ZXID。ZXID 比较大的服务器优先作为Leader

ii. 如果 ZXID 相同，那么就比较 myid。myid 较大的服务器作为Leader服务器

(4) 统计投票。每次投票后，服务器都会统计投票信息，判断是否已经有过半机器接受到相同的投票信息。

(5) 改变服务器状态。一旦确定了Leader，每个服务器就会更新自己的状态，如果是Follower，那么就变更为FOLLOWING，如果是Leader，就变更为LEADING。

运行时选举

当集群中的 leader 服务器出现宕机或者不可用的情况时，进入新一轮的Leader选举：

变更状态：Leader挂后，余下的非follower都将自己状态更新为LOOKING，然后开始Leader选举。之后的过程和启动时选举过程一致。

运行时选举需要确保已经被leader提交的事务能够提交、同时丢弃已经被跳过的事务Proposal：

新选举的Leader有最大的zxid事务Proposal，那么这个新Leader一定具有已经提交的提案。

新leader会将epoch号+1，老的leader挂了以后重启，会作为follower接入集群，新leader会让它将拥有旧的epoch号的未被COMMIT的proposal清除。

附：reentrantLock单点锁和curator分布式锁实现对比

package com.faith.test;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 单点锁和分布式锁
 */
public class LockDemo {

    /**
     * 公共变量
     */
    static int count = 1;

    /**
     * 测试
     * @param args
     */
    public static void main(String[] args) {
        reentrantLock();
        distrubuteLock();
    }

    /**
     * 分布式锁
     */
    public static void distrubuteLock() {
        CuratorFramework curatorFramework = CuratorFrameworkFactory.builder()
                .connectString("192.168.160.132:2181") // zookeeper地址
                .sessionTimeoutMs(5000) // session超时时间
                .retryPolicy(new ExponentialBackoffRetry(1000, 10))
                .build();

        curatorFramework.start();

        final CountDownLatch countdown = new CountDownLatch(10);
        final InterProcessMutex lock = new InterProcessMutex(curatorFramework, "/lock");

        for(int i = 0; i < 10; i++) {
            new Thread(() -> {
                {
                    try {
                        countdown.await();
                        lock.acquire();
                        System.out.println(count++);
                    } catch (Exception e) {
                    } finally {
                        try {
                            lock.release();
                        } catch (Exception e) {
                        }
                    }
                }
            },"t" + i).start();
            countdown.countDown();
        }
    }

    /**
     * 单点锁
     */
    public static void reentrantLock() {
        final CountDownLatch countdown = new CountDownLatch(1);
        final ReentrantLock reentrantLock = new ReentrantLock();

        for(int i = 0; i < 10; i++) {
            new Thread(() -> {
                {
                    try {
                        countdown.await();
                        reentrantLock.lock();
                        System.out.println(count++);
                    } catch (Exception e) {
                    } finally {
                        reentrantLock.unlock();
                    }
                }
            },"t" + i).start();
            countdown.countDown();
        }
    }
}

参考：《分布式过程协同技术详解》、《从Paxos到Zookeeper》

zookeeper