几篇介绍ZooKeeper的文章

aaahd

　　http://agapple.iteye.com/blog/1111377
　　http://www.iteye.com/topic/1118626

　　



ZooKeeper解惑

最近针对ZK一些比较疑惑的问题，再看了一下相关代码，列举如下。这里只列官方文档中没有的，或者不清晰的。以zookeeper-3.3.3为基准。以下用ZK表示ZooKeeper。

一个ZooKeeper对象，代表一个ZK Client。应用通过ZooKeeper对象中的读写API与ZK集群进行交互。一个简单的创建一条数据的例子，只需如下两行代码：

ZooKeeper zk = new ZooKeeper(serverList, sessionTimeout, watcher);
zk.create("/test", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

Client和ZK集群的连接和Session的建立过程

ZooKeeper对象一旦创建，就会启动一个线程（ClientCnxn）去连接ZK集群。ZooKeeper内部维护了一个Client端状态。

    public enum States {
CONNECTING, ASSOCIATING, CONNECTED, CLOSED, AUTH_FAILED;
…}

第一次连接ZK集群时，首先将状态置为CONNECTING，然后挨个尝试连接serverlist中的每一台Server。Serverlist在初始化时，顺序已经被随机打乱：

Collections.shuffle(serverAddrsList)

这样可以避免多个client以同样的顺序重连server。重连的间隔毫秒数是0-1000之间的一个随机数。

一旦连接上一台server，首先发送一个ConnectRequest包，将ZooKeeper构造函数传入的sessionTimeout数值发动给Server。ZooKeeper Server有两个配置项：

minSessionTimeout 单位毫秒。默认2倍tickTime

maxSessionTimeout 单位毫秒。默认20倍tickTime

（tickTime也是一个配置项。是Server内部控制时间逻辑的最小时间单位）

如果客户端发来的sessionTimeout超过min-max这个范围，server会自动截取为min或max，然后为这个Client新建一个Session对象。Session对象包含sessionId、timeout、tickTime三个属性。其中sessionId是Server端维护的一个原子自增long型（8字节）整数；启动时Leader将其初始化为1个字节的leader Server Id+当前时间的后5个字节+2个字节的0；这个可以保证在leader切换中，sessionId的唯一性（只要leader两次切换为同一个Server的时间间隔中session建立数不超过(
2的16次方)*间隔毫秒数。。。不可能达到的数值）。

ZK Server端维护如下3个Map结构，Session创建后相关数据分别放入这三个Map中：

Map<Long[sessionId], Session> sessionsById

Map<Long[sessionId], Integer> sessionsWithTimeout

Map<Long[tickTime], SessionSet> sessionSets

其中sessionsById简单用来存放Session对象及校验sessionId是否过期。sessionsWithTimeout用来维护session的持久化：数据会写入snapshot，在Server重启时会从snapshot恢复到sessionsWithTimeout，从而能够维持跨重启的session状态。

Session对象的tickTime属性表示session的过期时间。sessionSets这个Map会以过期时间为key，将所有过期时间相同的session收集为一个集合。Server每次接到Client的一个请求或者心跳时，会根据当前时间和其sessionTimeout重新计算过期时间并更新Session对象和sessionSets。计算出的过期时间点会向上取整为ZKServer的属性tickTime的整数倍。Server启动时会启动一个独立的线程负责将大于当前时间的所有tickTime对应的Session全部清除关闭。

Leader收到连接请求后，会发起一个createSession的Proposal，如果表决成功，最终所有的Server都会在其内存中建立同样的Session，并作同样的过期管理。等表决通过后，与客户端建立连接的Server为这个session生成一个password，连同sessionId，sessionTimeOut一起返回给客户端（ConnectResponse）。客户端如果需要重连Server，可以新建一个ZooKeeper对象，将上一个成功连接的ZooKeeper 对象的sessionId和password传给Server

ZooKeeper zk = new ZooKeeper(serverList, sessionTimeout, watcher, sessionId,passwd);

ZKServer会根据sessionId和password为同一个client恢复session，如果还没有过期的话。

Server生成password的算法比较有意思：

new Random(sessionId ^ superSecret).nextBytes(byte[] passwd)

superSecret是一个固定的常量。Server不保存password，每次在返回client的ConnectRequest应答时计算生成。在客户端重连时再重新计算，与传入的password作比较。因为Random相同的seed随机生成的序列是完全相同的！

Client发送完ConnectRequest包，会紧接着发送authInfo包(OpCode.auth)和setWatches 包OpCode.setWatches；authInfo列表由ZooKeeper的addAuthInfo()方法添加，用来进行自定义的认证和授权。

最后当zookeeper.disableAutoWatchReset为false时，若建立连接时ZooKeeper注册的Watcher不为空，那么会通过setWatches告诉ZKServer重新注册这些Watcher。这个用来在Client自动切换ZKServer或重练时，尚未触发的Watcher能够带到新的Server上

以上是连接初始化的时候做的事情。

关于ACL

之前看到很多例子里

zk.create(“/test”, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

中Ids.OPEN_ACL_UNSAFE的地方用 Ids.CREATOR_ALL_ACL，在zookeeper-3.3.3上面跑直接就挂了，报下面的错：

org.apache.zookeeper.KeeperException$InvalidACLException: KeeperErrorCode = InvalidACL for /test

at org.apache.zookeeper.KeeperException.create(KeeperException.java:112)

at org.apache.zookeeper.KeeperException.create(KeeperException.java:42)

at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:637)

是因为3.3.3的ACL进行了细微的调整。先来看下ACL的数据结构：

每一个znode节点上都可以设置一个访问控制列表，数据结构为List

ACL
+--perms int （allow What）
+--id Id
+--scheme String （Who）
+--id String      （How）

一个ACL对象就是一个Id和permission对，用来表示哪个/哪些范围的Id（Who）在通过了怎样的鉴权（How）之后，就允许进行那些操作（What）：Who How What；permission（What）就是一个int表示的位码，每一位代表一个对应操作的允许状态。类似unix的文件权限，不同的是共有5种操作：CREATE、READ、WRITE、DELETE、ADMIN(对应更改ACL的权限)；Id由scheme（Who）和一个具体的字符串鉴权表达式id(How)构成，用来描述哪个/哪些范围的Id应该怎样被鉴权。Scheme事实上是所使用的鉴权插件的标识。id的具体格式和语义由scheme对应的鉴权实现决定。不管是内置还是自定义的鉴权插件都要实现AuthenticationProvider接口(以下简称AP)。自定义的鉴权插件由zookeeper.authProvider开头的系统属性指定其类名，例如：

authProvider.1=com.f.MyAuth

authProvider.2=com.f.MyAuth2

AP接口的getScheme()方法定义了其对应的scheme

客户端与Server建立连接时，会将ZooKeeper.addAuthInfo()方法添加的每个authInfo都发送给ZKServer。

void addAuthInfo(String scheme, byte auth[])

addAuthInfo 方法本身也会直接将authInfo发送给ZKServer。ZKServer接受到authInfo请求后，首先根据scheme找到对应的AP，然后调用其handleAuthentication()方法将auth数据传入。对应的AP将auth数据解析为一个Id，将其加入连接上绑定的authInfo列表（List）中。Server在接入客户端连接时，首先会自动在连接上加上一个默认的scheme为ip的authIndo：authInfo.add(new Id(“ip”, client-ip));

鉴权时调用AP的matches()方法判断进行该操作的当前连接上绑定的authInfo是否与所操作的znode的ACL列表匹配。

ZK有4个内置的scheme：

· world 只有一个唯一的id：anyone；表示任何人都可以做对应的操作。这个scheme没有对应的鉴权实现。只要一个znode的ACL list中包含有这个scheme的Id，其对应的操作就运行执行

· auth 没有对应的id，或者只有一个空串””id。这个scheme没有对应的鉴权实现。语义是当前连接绑定的适合做创建者鉴权的autoInfo (通过调用autoInfo的scheme对应的AP的isAuthenticated()得知)都拥有对应的权限。遇到这个auth后，Server会根据当前连接绑定的符合要求的autoInfo生成ACL加入到所操作znode的acl列表中。

· digest 使用username:password格式的字符串生成MD5 hash 作为ACL ID。 具体格式为：username:base64 encoded SHA1 password digest.对应内置的鉴权插件：DigestAuthenticationProvider

· ip 用IP通配符匹配客户端ip。对应内置鉴权插件IPAuthenticationProvider

只有两类API会改变Znode的ACL列表：一个是create()，一个是setACL()。所以这两个方法都要求传入一个List。Server接到这两种更新请求后，会判断指定的每一个ACL中，scheme对应的AuthenticationProvider是否存在，如果存在，调用其isValid(String)方法判断对应的id表达式是否合法。。。具体参见PrepRequestProcessor.fixupACL()方法。上文的那个报错是因为CREATOR_ALL_ACL只包含一个ACL : Perms.ALL,
Id(“auth”, “”)，而auth要求将连接上适合做创建者鉴权的autoInfo都加入节点的acl中，而此时连接上只有一个默认加入的Id(“ip”, client-ip)，其对应的IPAuthenticationProvider的isAuthenticated()是返回false的，表示不用来鉴权node的创建者。

tbd：具体例子

关于Watcher

先来看一下ZooKeeper的API: 读API包括exists，getData，getChildren四种

Stat exists(String path, Watcher watcher)
Stat exists(String path, boolean watch)
void exists(String path, Watcher watcher, StatCallback cb, Object ctx)
void exists(String path, boolean watch  , StatCallback cb, Object ctx)
byte[] getData(String path, Watcher watcher, Stat stat)
byte[] getData(String path, boolean watch  , Stat stat)
void   getData(String path, Watcher watcher, DataCallback cb, Object ctx)
void   getData(String path, boolean watch  , DataCallback cb, Object ctx)
List<String> getChildren(String path, Watcher watcher)
List<String> getChildren(String path, boolean watch  )
void  getChildren(String path, Watcher watcher, ChildrenCallback cb, Object ctx)
void  getChildren(String path, boolean watch  , ChildrenCallback cb, Object ctx)
List<String> getChildren(String path, Watcher watcher, Stat stat)
List<String> getChildren(String path, boolean watch  , Stat stat)
void getChildren(String path, Watcher watcher, Children2Callback cb, Object ctx)
void getChildren(String path, boolean watch  , Children2Callback cb, Object ctx)

每一种按同步还是异步，添加指定watcher还是默认watcher又分为4种。默认watcher是只在ZooKeeper zk = new ZooKeeper(serverList, sessionTimeout, watcher)中指定的watch。如果包含boolean watch的读方法传入true则将默认watcher注册为所关注事件的watch。如果传入false则不注册任何watch

写API包括create、delete、setData、setACL四种，每一种根据同步还是异步又分为两种：

String create(String path, byte data[], List<ACL> acl, CreateMode createMode)
void   create(String path, byte data[], List<ACL> acl, CreateMode createMode, StringCallback cb, Object ctx)
void delete(String path, int version)
void delete(String path, int version, VoidCallback cb, Object ctx)
Stat setData(String path, byte data[], int version)
void setData(String path, byte data[], int version, StatCallback cb, Object ctx)
Stat setACL(String path, List<ACL> acl, int version)
void setACL(String path, List<ACL> acl, int version, StatCallback cb, Object ctx)

一个读写交互，或者说pub/sub的简单描述如下图：

更详细一点：

可见Watcher机制的轻量性：通知的只是事件。Client和server端额外传输的只是个boolean值。对于读写api操作来说，path和eventType的信息本身就有了。只有在notify的时候才需要加上path、eventType的信息。内部存储上，Server端只维护一个Map(当然会根据watcher的类型分为两个)，key为path，value为本身以及存在的连接对象。所以存储上也不是负担。不会随着watcher的增加无限制的增大

Watcher的一次性设计也大大的减轻了服务器的负担和风险。假设watcher不是一次性，那么在更新很频繁的时候，大量的通知要不要丢弃？精简？并发怎么处理？都是一个问题。一次性的话，这些问题就都丢给了Client端。并且Client端事实上并不需要每次密集更新都去处理。

如果一个znode上大量Client都注册了watcher，那么触发的时候是串行的。这里可能会有一些延迟。

关于Log文件和snapshot

Follower/Leader每接收到一个PROPOSAL消息之后，都会写入log文件。log文件的在配置项dataLogDir指定的目录下面。文件名为log.+第一条记录对应的zxid

[linxuan@test036081 conf]$ ls /usr/zkdataLogDir/version-2/

log.100000001 log.200000001

ZooKeeper在每次写入log文件时会做检查，当文件剩余大小不足4k的时候，默认会一次性预扩展64M大小。这个预扩展大小可以通过系统属性zookeeper.preAllocSize或配置参数preAllocSize指定，单位为K；

会为每条记录计算checksum，放在实际数据前面

每写1000条log做一次flush(调用BufferedOutputStream.flush()和FileChannel.force(false))。这个次数直到3.3.3都是硬编码的，无法配置

每当log写到一定数目时，ZooKeeper会将当前数据的快照输出为一个snapshot文件：

      randRoll = Random.nextInt(snapCount/2);
if (logCount > (snapCount / 2 + randRoll)) {
rollLog();
take_a_snapshot_in_a_new_started_thread();
}

这个randRoll是一个随机数，为了避免几台Zk Server在同一时间都做snapshot

输出快照的log数目阀值snapCount可以通过zookeeper.snapCount系统属性设置，默认是100000条。输出snapshot文件的操作在新创建的单独线程里进行。任一时刻只会有一个snapshot线程。Snapshot文件在配置项dataDir指定的目录下面生成，命名格式为snapshot.+最后一个更新的zxid。

如指定dataDir=/home/linxuan/zookeeper-3.3.3/data，则snapshot文件为：

[linxuan@test036081 version-2]$ ls /home/linxuan/zookeeper-3.3.3/data/version-2

snapshot.0 snapshot.100000002

每个snapshot文件也都会写入全部数据的一个checksum。

ZK在每次启动snapshot线程前都会将当前的log文件刷出，在下次写入时创建一个新的log文件。不管当前的log文件有没有写满。旧的log文件句柄会在下一次commit（也就是flush的时候）再顺便关闭。

所以这种机制下，log文件会有一定的空间浪费，大多情况下会没有写满就换到下一个文件了。可以通过调整preAllocSize和snapCount两个参数来减少这种浪费。但是定时自动删除没用的log文件还是必须的，只保留最新的即可。

为了保证消息的安全，排队的消息在没有flush到log文件之前不会提交到下一个环节。而为了提高log文件写入的效率，又必须做批量flush。所以更新消息实际上也是和批量flushlog文件的操作一起，批量提交到下一个协议环节的。当请求比较少时（包括读请求），每个更新会很快刷出，即使没有写够1000条。当请求压力很大时，才会一直等堆积到1000条才刷出log文件，同时送出消息到下一个环节。这里的实现比较细致，实质上是在压力大时，不光是写log，连同消息处理都做了一个批量操作。具体实现细节在SyncRequestProcessor中

Client和ZK集群的完整交互

ZK整体上来说，通过单线程和大量的队列来达到消息在集群内完成一致性协议的情况下，仍然能保证全局顺序。下面是一个线程和queue的全景图：


这个图中，除了个别的之外，每个节点都要么代表一个Thread，要么代表一个queue

其他

ZKServer内部通过大量的queue来处理消息，保证顺序。这些queue的大小本身都不设上限。有一个配置属性globalOutstandingLimit用来指定Server的最大请求堆积数。ZKServer在读入消息时如果发觉内部的全局消息计数大于这个值，就会直接关闭当前连接上的读取来保护服务端。（取消与当前Client的Nio连接上的读取事件注册）

　　

　　

这段时间来，也在和公司里的一些同学交流使用zk的心得，整理了一些常见的zookeeper问题。这个页面的目标是解答一些zk常见的使用问题，同时也让大家明确zk不能干什么。页面会一直更新。

客户端

1.
客户端对ServerList的轮询机制是什么

随机，客户端在初始化( new ZooKeeper(String connectString, int sessionTimeout, Watcher watcher) )的过程中，将所有Server保存在一个List中，然后随机打散，形成一个环。之后从0号位开始一个一个使用。

两个注意点：1. Server地址能够重复配置，这样能够弥补客户端无法设置Server权重的缺陷，但是也会加大风险。（比如: 192.168.1.1:2181,192.168.1.1:2181,192.168.1.2:2181). 2. 如果客户端在进行Server切换过程中耗时过长，那么将会收到SESSION_EXPIRED. 这也是上面第1点中的加大风险之处。

2.客户端如何正确处理CONNECTIONLOSS(连接断开)
和 SESSIONEXPIRED(Session 过期)两类连接异常

在ZooKeeper中，服务器和客户端之间维持的是一个长连接，在 SESSION_TIMEOUT 时间内，服务器会确定客户端是否正常连接(客户端会定时向服务器发送heart_beat),服务器重置下次SESSION_TIMEOUT时间。因此，在正常情况下，Session一直有效，并且zk集群所有机器上都保存这个Session信息。在出现问题情况下，客户端与服务器之间连接断了（客户端所连接的那台zk机器挂了，或是其它原因的网络闪断），这个时候客户端会主动在地址列表（初始化的时候传入构造方法的那个参数connectString）中选择新的地址进行连接。

好了，上面基本就是服务器与客户端之间维持长连接的过程了。在这个过程中，用户可能会看到两类客异常CONNECTIONLOSS(连接断开) 和SESSIONEXPIRED(Session 过期)。

CONNECTIONLOSS发生在上面红色文字部分，应用在进行操作A时，发生了CONNECTIONLOSS，此时用户不需要关心我的会话是否可用，应用所要做的就是等待客户端帮我们自动连接上新的zk机器，一旦成功连接上新的zk机器后，确认刚刚的操作A是否执行成功了。

SESSIONEXPIRED发生在上面蓝色文字部分，这个通常是zk客户端与服务器的连接断了，试图连接上新的zk机器，这个过程如果耗时过长，超过 SESSION_TIMEOUT 后还没有成功连接上服务器，那么服务器认为这个session已经结束了（服务器无法确认是因为其它异常原因还是客户端主动结束会话），开始清除和这个会话有关的信息，包括这个会话创建的临时节点和注册的Watcher。在这之后，客户端重新连接上了服务器在，但是很不幸，服务器会告诉客户端SESSIONEXPIRED。此时客户端要做的事情就看应用的复杂情况了，总之，要重新实例zookeeper对象，重新操作所有临时数据（包括临时节点和注册Watcher）。

3.
不同的客户端对同一个节点是否能获取相同的数据

4.
一个客户端修改了某个节点的数据，其它客户端能够马上获取到这个最新数据吗

ZooKeeper不能确保任何客户端能够获取（即Read Request）到一样的数据，除非客户端自己要求：方法是客户端在获取数据之前调用org.apache.zookeeper.AsyncCallback.VoidCallback, java.lang.Object) sync.

通常情况下（这里所说的通常情况满足：1. 对获取的数据是否是最新版本不敏感，2. 一个客户端修改了数据，其它客户端需要不需要立即能够获取最新），可以不关心这点。

在其它情况下，最清晰的场景是这样：ZK客户端A对 /my_test 的内容从 v1->v2, 但是ZK客户端B对 /my_test 的内容获取，依然得到的是 v1. 请注意，这个是实际存在的现象，当然延时很短。解决的方法是客户端B先调用 sync(), 再调用 getData().

5.
ZK为什么不提供一个永久性的Watcher注册机制

不支持用持久Watcher的原因很简单，ZK无法保证性能。
6.
使用watch需要注意的几点

a. Watches通知是一次性的，必须重复注册.

b. 发生CONNECTIONLOSS之后，只要在session_timeout之内再次连接上（即不发生SESSIONEXPIRED），那么这个连接注册的watches依然在。

c. 节点数据的版本变化会触发NodeDataChanged，注意，这里特意说明了是版本变化。存在这样的情况，只要成功执行了setData()方法，无论内容是否和之前一致，都会触发NodeDataChanged。

d. 对某个节点注册了watch，但是节点被删除了，那么注册在这个节点上的watches都会被移除。

e. 同一个zk客户端对某一个节点注册相同的watch，只会收到一次通知。即

for( int i = 0; i < 3; i++ ){
zk.getData( path, true, null );
zk.getChildren( path, true );
}

7.我能否收到每次节点变化的通知

如果节点数据的更新频率很高的话，不能。

原因在于：当一次数据修改，通知客户端，客户端再次注册watch，在这个过程中，可能数据已经发生了许多次数据修改，因此，千万不要做这样的测试：”数据被修改了n次，一定会收到n次通知”来测试server是否正常工作。（我曾经就做过这样的傻事，发现Server一直工作不正常？其实不是）。即使你使用了GitHub上这个客户端也一样。

8.能为临时节点创建子节点吗

不能。

9.
是否可以拒绝单个IP对ZK的访问,操作

ZK本身不提供这样的功能，它仅仅提供了对单个IP的连接数的限制。你可以通过修改iptables来实现对单个ip的限制，当然，你也可以通过这样的方式来解决。

10.
在getChildren(String
path, boolean watch)是注册了对节点子节点的变化，那么子节点的子节点变化能通知吗

不能

11.创建的临时节点什么时候会被删除，是连接一断就删除吗？延时是多少？

连接断了之后，ZK不会马上移除临时数据，只有当SESSIONEXPIRED之后，才会把这个会话建立的临时数据移除。因此，用户需要谨慎设置Session_TimeOut

服务器