浅入浅出zookeeper

zookeeper是我们日常开发中每天都能接触到的组件，但是好像很多人对其缺乏了解，所以心血来潮写了这篇文章。首先简单介绍一下zookeeper。zookeeper最开始是hadoop的子项目，后来升级为了apache的顶级项目。我想很多人刚开始接触zookeeper，都会带有这样的疑问，zookeeper的出现是为了解决什么样的问题?为了解决这样的问题，zookeeper引入了什么样的手段？这样的手段是通过什么原理去解决这个问题的？带着这样的疑惑，开始正文。友情提示：如果阅读过程中感到不适，可以直接跳到第五部分总结。

一. 起源

zookeeper起源于曾经的互联网巨头雅虎，基于Google的Chubby实现。zookeeper的出现是为了解决单点问题，所谓的单点问题大意指的是如果服务仅仅部署在一台机器上，万一机器故障，整个服务将会不可用。所以，在生产环境中，我们会需要若干个备胎，正如mysql的主从，redis的主从，集群所做的那样。zookeeper要做的就是快速解决备胎转正问题。
关于zookeeper的命名，也有一段趣闻，由于雅虎的项目很多都是由动物的名字命名的，所以给这个项目起了zookeeper这个名字，译为动物管理员，想要用其管理雅虎的各个项目。

zookeeper在创立之初设定了几个目标：

简单的数据模型：目的是为了让使用者清晰明了的使用。
可以构建集群：这个也很好理解，本身zookeeper就是用于辅助管理分布式系统，所谓打铁还需自身硬，zookeeper自己也需要保证自己的高可用。
顺序访问：客户端的每个更新请求，zookeeper都会分配一个全局唯一ID，这个也很好理解，作为一个集群，如果没有一个唯一ID保证顺序，则可能会出现并发问题。
高性能：zookeeper将全量数据存储在内存中。
以上四点都是概念性的东西，只要理解一点即可，zookeeper想要又好又快的解决备胎转正问题，而且想要尽量减少纠纷。

二. zookeeper的写入过程简析

这部分想要简述zookeeper是如何进行数据写入的，包括集群模式下的数据同步。

1.zookeeper节点的数据结构

如果有zookeeper的使用经验，会知道zookeeper的节点路径非常类似于linux的文件系统，事实上它是如何进行存储的呢？
其实从本质上讲，zookeeper也是一个kv存储数据库，数据存储在DataTree这个数据结构中，核心存储是一个ConcurrentHashMap，其中key值是节点路径，value值为一个DataNode类型的数据。可以说，zookeeper的所有操作都是对这个Map的操作。在DataTree中存放数据的成员变量为：private final NodeHashMap nodes;下面可以看到其构造过程：

public NodeHashMapImpl(DigestCalculator digestCalculator) {
    this.digestCalculator = digestCalculator;
    nodes = new ConcurrentHashMap<>();
    hash = new AdHash();
    digestEnabled = ZooKeeperServer.isDigestEnabled();
}

正如那句俗语，美丽的外表千篇一律，其实redis的存储方式和zookeeper也是类似的，redis数据库的所有数据存放于下面的这个结构体中：

typedef struct redisDb {
    dict *dict;                 /* The keyspace for this DB */
    dict *expires;              /* Timeout of keys with a timeout set */
    dict *blocking_keys;        /* Keys with clients waiting for data (BLPOP)*/
    dict *ready_keys;           /* Blocked keys that received a PUSH */
    dict *watched_keys;         /* WATCHED keys for MULTI/EXEC CAS */
    int id;                     /* Database ID */
    long long avg_ttl;          /* Average TTL, just for stats */
    unsigned long expires_cursor; /* Cursor of the active expire cycle. */
    list *defrag_later;         /* List of key names to attempt to defrag one by one, gradually. */
} redisDb;

第一个成员变量为redis自建的数据结构：字典，底层实现是一个hashtable，redis数据库的key值为一个字符串，value值是一个如下数据结构的redisObject及结构体：

typedef struct redisObject {
    unsigned type:4;
    unsigned encoding:4;
    unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or
                            * LFU data (least significant 8 bits frequency
                            * and most significant 16 bits access time). */
    int refcount;
    void *ptr;
} robj;

类似于上文的DataNode。

2. 一个创建节点请求

server端对客户端请求的处理是类似的，大体都是用IO多路复用技术，对不同类型事件做相应的处理，具体的实现还是很复杂的，这里将重点放在zookeeper对其内部对象的操作上，而不是底层的IO上。以创建节点为例：DataTree结构中有一个成员函数: public void createNode(final String path, byte[] data, List<ACL> acl, long ephemeralOwner, int parentCVersion, long zxid, long time, Stat outputStat) throws KeeperException.NoNodeException, KeeperException.NodeExistsException用于创建节点：大致步骤为：
首先拿到需要创建节点的父节点信息：

String parentName = path.substring(0, lastSlash);
String childName = path.substring(lastSlash + 1);
StatPersisted stat = createStat(zxid, time, ephemeralOwner);
DataNode parent = nodes.get(parentName);
if (parent == null) {
    throw new KeeperException.NoNodeException();
}

其次将节点添加其父节点的子节点记录中，然后将节点放到nodes中，

DataNode child = new DataNode(data, longval, stat);
parent.addChild(childName);
nodes.postChange(parentName, parent);
nodeDataSize.addAndGet(getNodeSize(path, child.data));
nodes.put(path, child);

父节点的addChild函数只是将子节点的名称记录在父节点的一个set中，

private Set<String> children = null;

public synchronized boolean addChild(String child) {
    if (children == null) {
        children = new HashSet<String>(8);
    }
    return children.add(child);
}

所以得出结论：父节点只存储子节点的节点路径，并不存储节点的实际内容。若想要拿到节点值，还需要从nodes这个map中获取，时间复杂度为O(1)。

最后，如果有客户端在监听这个节点事件，则将事件通知客户端：

dataWatches.triggerWatch(path, Event.EventType.NodeCreated);
childWatches.triggerWatch(parentName.equals("") ? "/" : parentName, Event.EventType.NodeChildrenChanged);

上述过程只是大体过程，省略了有些对节点事物ID的校验流程，但不影响理解。

3. 集群数据的写入过程

集群和主从是比较相似的概念，都是用若干机器保证服务的高可用。主从架构通常设计为主库处理事务请求，从库处理非事务请求，比如我们日常使用的redis，mysql就是如此，一主多从，主写从读（事务请求大体上指的是增，删，改，非事务请求为查询请求）。一般而言，集群应该是一个多主机集合的概念，比如redis的集群设计就是通过hash槽，将不同key值映射到不同的主机上，集群机器数目变更的同时要对应hash槽的重新分配。zookeeper的集群有点类似于主从，所有的事务请求需要交付Leader处理。其实这里可以发现，无论是redis还是zookeeper，一个事务必须交由单独的一台主机进行处理，不存在一对多的关系。先介绍一下zookeeper节点的三种角色：

Leader: 处理事务请求，保证事务顺序；调度集群内部服务。
Follower：转发事务请求给Leader节点；参与投票选举Leader；参与事务的proposal投票；
Observer：转发事务请求给Leader节点。

虽然Leader节点决定着集群的整体状态，但是每一个事务请求都需要过半的公民（具有选举权）投票通过才行，这个过程类似于2PC（对2PC没有了解的童鞋可以去看一下分布式事务解决方案），这个投票的流程被称为Proposal。这个过程中Leader会将提议放在投票箱中(final ConcurrentMap<Long, Proposal> outstandingProposals = new ConcurrentHashMap<Long, Proposal>();)，同时广播给其他Follower节点，Follower收到后，开始Sync，完成后发送ACK消息给Leader，Leader收到过半消息后，进行COMMIT流程。这些步骤间涉及到的消息类型大致有：

    /**
     * This message type is sent to a leader to request and mutation operation.
     * The payload will consist of a request header followed by a request.
     */
    static final int REQUEST = 1; -- 非Leader节点转发事务请求给Leader

    /**
     * This message type is sent by a leader to propose a mutation.
     */
    public static final int PROPOSAL = 2; --Leader节点发出投票消息

    /**
     * This message type is sent by a follower after it has synced a proposal.
     */
    static final int ACK = 3; --Follower节点发送同意应答

    /**
     * This message type is sent by a leader to commit a proposal and cause
     * followers to start serving the corresponding data.
     */
    static final int COMMIT = 4; -- leader节点发出提交指令。

    /**
     * This message type informs observers of a committed proposal.
     */
    static final int INFORM = 8; -- 由于Observer节点没有不参与投票，没有事务记录，Leader需要发送特定的INFORM消息告诉它进行数据同步。

基于上述概念，假如一个非Leader节点收到一个事务请求，事务请求的大致流程：

节点发送REQUEST请求同步到Leader节点。
节点对请求进行判断。
Leader节点发出PROPOSAL请求。
Follower节点返回ACK应答。
Leader节点在收到超过半数的ACK应答后开始COMMIT事务。
Follower节点收到COMMIT消息进行事务提交。
Observer节点由于不参与投票，没有记录事务相关信息，所以Leader要发INFORM消息给Observer。

WARNING: 上述只是大致流程，省略了大量细节。

三. Watcher机制

zookeeper一个非常重要的应用就是它的发布/订阅功能，watcher机制是其实现这一功能的关键。zookeeper允许客户端向服务端针对某些特定场景注册一个watcher，在这些场景被触发时，服务端会回调watcher进行通知。下面展示一个监听事件的始末：

1. watcher属性

接口类watcher是一个事件处理器，其中有两个比较重要的枚举，分别定义了通知状态和事件类型。

enum KeeperState {
    // 客户端失联
    Disconnected(0),
    // 客户端与服务端处于连接状态
    SyncConnected(3),
    // 权限有问题
    AuthFailed(4),
    // 客户端连接着一个只读的server
    ConnectedReadOnly(5),
    // 客户端会话超时
    Expired(-112),
    // 客户端已关闭
    Closed(7);
}

// 看命名大概都能知道是什么类型的事件了
enum EventType {
    None(-1),
    NodeCreated(1),
    NodeDeleted(2),
    NodeDataChanged(3),
    NodeChildrenChanged(4),
    DataWatchRemoved(5),
    ChildWatchRemoved(6),
    PersistentWatchRemoved (7);
}

2. watcher注册

客户端可以通过getData, getChildren,exists方法进行注册watcher，大致流程为：

public void getData(final String path, Watcher watcher, DataCallback cb, Object ctx) {
    final String clientPath = path;
    PathUtils.validatePath(clientPath);

    // the watch contains the un-chroot path
    WatchRegistration wcb = null;
    if (watcher != null) {
        wcb = new DataWatchRegistration(watcher, clientPath);
    }

    final String serverPath = prependChroot(clientPath);

    RequestHeader h = new RequestHeader();
    h.setType(ZooDefs.OpCode.getData);
    GetDataRequest request = new GetDataRequest();
    request.setPath(serverPath);
    request.setWatch(watcher != null);
    GetDataResponse response = new GetDataResponse();
    cnxn.queuePacket(h, new ReplyHeader(), request, response, cb, clientPath, serverPath, ctx, wcb);
}

// 内部注册流程
public void register(int rc) {
    if (shouldAddWatch(rc)) {
        Map<String, Set<Watcher>> watches = getWatches(rc);
        synchronized (watches) {
            Set<Watcher> watchers = watches.get(clientPath);
            if (watchers == null) {
                watchers = new HashSet<Watcher>();
                watches.put(clientPath, watchers);
            }
            watchers.add(watcher);
        }
    }
}

可以看到一个注册的过程分为两个步骤，第一步把需要监听的事件自己记录，第二步将需要监听的事件发送到服务端上。

在客户端的监听请求送达服务端后，服务端通过两个map对监听事件进行管理：

private final Map<String, Set<Watcher>> watchTable = new HashMap<>();

private final Map<Watcher, Set<String>> watch2Paths = new HashMap<>();

一个是从节点路径维度关联watcher，一个是从watcher维度关联节点路径。上文中已经看到过了，在节点发生事物变更时，会触发服务端的watcher监听trigger，这时候服务端就要把发生的事件通知客户端。这里需要注意的一点是：

// 服务端回调客户端的内容
public WatchedEvent(EventType eventType, KeeperState keeperState, String path) {
    this.keeperState = keeperState;
    this.eventType = eventType;
    this.path = path;
}

服务端回调客户端仅仅包含最基础的内容，节点路径，节点状态，事件类型，也就是说客户端只会知道这个节点发生了自己感兴趣的事件类型，想要拿到节点变更后的信息，则需要再次请求server端get，这种操作非常的轻量级。

四. 实际应用

上文对zookeeper的原理做了非常非常简略的介绍，下面看一下其实际应用有哪些。

1. 管理配置信息

如果将某些配置注册在zookeeper上，那么在服务启动的时候先会拿到目前的配置信息，同时向服务端注册一个watcher，通过上文的描述我们知道，在节点发生变更时，服务端会主动通知客户端。如此一来，配置信息发生变更的时候，客户端会及时拿到最新的配置。

2. 命名服务

基于zookeeper，应用可以基于命名拿到真实数据，比如一些rpc框架，应用可以根据名字拿到服务提供者的真实信息，比如ip端口，URI等。
再比如为一个应用配置一个数据库节点，如果mysql主库挂了，在从库顶上后，应用可以迅速重新连接，不用更改任何配置。

3. 分布式锁

可以基于zookeeper实现一个比较完美的分布式锁，前两天一位同事分享了基于redis的分布式锁，基于redis的分布式锁一个比较困难的问题是锁续期的问题：

锁时间设置过长，如果线程挂了锁就一直不会释放。
锁的时间过短，则可能任务没做完锁就释放了。

为了解决上述问题还需要引入新的线程去监控任务线程。
如果使用zookeeper，每个资源使用方只需要建立临时节点同时注册watcher，等待节点变更通知即可，

假如正常执行完，任务线程会将节点剔除。
假如拿锁的线程在执行过程中挂掉，zookeeper会将其自动剔除，则下一个ID可以拿到锁。

这种操作在锁竞争压力大的时候，可能会出现惊群效应，依然可以通过zk解决，这里就不展开了。

4. 其他

zookeeper在很多大家耳熟能详的应用中有着作用，例如：我们日常使用的dubbo，推荐使用zookeeper作为注册中心，kafka使用zookeeper对broker进行管理，以及用zookeeper维护broker和topic的对应关系。

五. 总结

本文简述了zookeeper的几个基本概念：

节点的数据结构: ConcurrentHashMap。
节点的写入过程：维持节点父子间的对应关系，同时将节点假如map大家庭中。
Watcher机制：客户端向服务端注册需要监听的事件，服务端在检查到事件发生时通知客户端，客户端再去服务端查询到底发生了啥。
zookeeper应用：分布式锁，命名服务，注册中心等。

本文只是简述了zookeeper的非常基本的概念，甚至连Leader选举都没有讲，如果有兴趣的小伙伴欢迎自己查阅资料或者阅读源码研究，感谢阅读。