zookeeper学习记录三(session,watcher,persit机制)

背景

继续前面的zookeeper学习的专题，这次主要是结合项目中遇到的一些问题，进一步学习了下zookeeper的一些内部机制。

 

针对以下几个问题：

1. zk是否可以保证watcher事件不丢失？

2. zk的EPHEMERAL节点的自动过期时间？ 

3. zk的如何保证节点数据不丢失？

 

如果你已经非常清楚这以上的几个问题，看官们可以不用往下看了。 

persit机制

zookeeper中的persit机制主要是通过本地disk进行持久化，在本地disk上会有个memory数据对象保持同步。

 

持久化实现：

ZKDatabase

 

• DataTree (内存树)
  
• FileTxnSnapLog (disk持久化)
• committedLog (FileTxnSnapLog的一份内存数据cache，默认存储500条变更记录)

DataTree(内存树)

zookeeper本身的数据结构就是一个树结构

数据模型（DataTree）:

 

• DataNode (1:n)
• data WatchManager (1:1，处理node节点的CRUD的变更事件，发送Watcher事件)
• child WatchManager (1:1,  处理node子节点的变更事件，发送Watcher事件)
• sessions (ephemerals)

DataNode模型：

• parent 
• data byte[]
• acl(安全)
• stat(审计信息)
• children

整个实现相对比较简单，就是查找一个树节点后进行响应的操作

 

FileTxnSnapLog (disk持久化)

持久化数据分两类： 

 

• TxnLog (类似于mysql/oracle的binlog/redolog)
• SnapShot (DataTree的数据镜像)

刚开始最容易搞不清楚就是Txnlog和SnapShot的区别，SnapShot主要是定期对DataTree的数据做一个本地备份，TxnLog只是一些历史的版本变更日志(每次由写事件变化，就会写入到该日志中)。

下面看一下，zookeeper在新节点启动后，是否如何保证数据一致： 

1. 首先节点启动后，尝试读取本地的SnapShot log数据(zkDb.loadDataBase())，反序列化为DataTree对象，并获取last zxid。 
2. follower启动后会向leader发送自己的last zxid
3. leader收到zxid后，对比自己当前的ZKDatabase中的last zxid
   如果当前follower的zxid在内存committedLog中，直接将内存中的committedLog提取出来进行发送，否则将当前的DataTree直接发送给follower.(不再是发送变更记录)
   
4. 数据同步完成后，follower会开始接收request请求

一致性机制

整个zk集群在处理数据变更过程中，会是先append变更信息到Txnlog中(此时会触发take snap操作)，最后在FinalRequestProcessor中更新内存中的DataTree信息。

触发take snap的条件：

 

if (logCount > (snapCount / 2 + randRoll)) {
randRoll = r.nextInt(snapCount/2);

 

 snapCount可以通过jvm参数zookeeper.snapCount指定，默认为100000。 这里zookeeper很巧妙的加两个随机处理，避免zk机器在同一时间点进行take snap处理，影响性能。

session机制

zookeeper会为每个client分配一个session，类似于web服务器一样。针对session可以有保存一些关联数据，zookeeper里针对session的一些关联数据主要就是EPHEMERAL节点。

EPHEMERAL的翻译为短命的，技术上理解就是session关闭后，其节点即消失，和session保持相同的生命周期。

创建EPHEMERAL节点：

zookeeper.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);

我们能用EPHEMERAL节点做啥?

1. 分布式集群是否存活监控. (每个节点启动后，注册一个EPHEMERAL节点到zookeeper中，注册一个Watcher获取EPHEMERAL节点的存在情况，消失即可代表集群节点dead)
2. 分布式lock (每个锁竞争者，排队时都在zookeeper中注册一个EPHEMERAL节点，排队过程中有节点dead了，zookeeper可以自动将其剔除队列，避免出现deadlock)
3. 替代web服务器的session，实现一个集中式session, session数据中的setAttribute都创建为EPHEMERAL节点，session关闭后即可自动删除，不会造成java中的"内存泄漏"

不扯远，下面看一下zookeeper中是如何处理session的管理？ 因为用上了zookeeper，必然要考虑分布式常见的问题

比如：

1. zookeeper server挂了，对应的session是否会丢失?
2. zookeeper client发生了failover后(出现了Connection Loss异常)，对应的session是否会丢失?

解答：

1. 在服务端，zookeeper中session的存储是有进行持久化的, 具体可见perist机制的描述。 一个新节点启动后，会从leader中同步对应的session数据
2. 在客户端，zookeeper在每次出现failover后(出现了Connection Loss异常)，会重新带上sessionId,sessionPasswd发起一次链接请求。接收到该请求的server，会返回内存中的session信息

所以，session是有可靠性保证的。

session expired机制

zookeeper中session expired机制和node数据一致性的保证原理类似，对应的follower都是受控于leader。

 

1. follower接收到客户端链接请求，就会向leader发送一次createSession的操作请求，leader收到后进行广播通知给所有的follower/observer节点createSession
2. leader会通过内存版的(SessionTrackerImpl)，定期扫描过期的session，发送一次closeSession的请求给所有的客户端
3. 在2发送过程中，如果有follower接收到过期session的请求，会提交给leader进行仲裁，leader会直接返回session expired。

session expired几个参数:

1. 服务端: minSessionTimeout (默认值为：tickTime * 2) , maxSessionTimeout (默认值为 : tickTime * 20) , ticktime的默认值为3000ms。所以session范围为6s ~ 60s 
2. 客户端： sessionTimeout， 无默认值，创建实例时必填。

一个误区： 很多人会按照hadoop文档中的建议，创建zookeeper客户端时设置了sessionTimeout为90s，而没有改变server端的配置，默认是不会生效的。

原因： 客户端的zookeeper实例在创建连接时，将sessionTimeout参数发送给了服务端，服务端会根据对应的minSessionTimeout/maxSessionTimeout的设置，强制修改sessionTimeout参数，也就是修改为6s~60s返回的参数。所以服务端不一定会以客户端的sessionTImeout做为session expire管理的时间。

 

int minSessionTimeout = zk.getMinSessionTimeout();
        if (sessionTimeout < minSessionTimeout) {
            sessionTimeout = minSessionTimeout;
        }
        int maxSessionTimeout = zk.getMaxSessionTimeout();
        if (sessionTimeout > maxSessionTimeout) {
            sessionTimeout = maxSessionTimeout;
        }

Watcher机制

watcher是zookeeper实现分布式lock一个很重要的feature，在写分布式lock时一定要对其有所了解。

就会冒出如下问题：

 

1. 什么情况下，会触发什么类型的watcher?
2. watcher信息出现failover是否会丢失?
3. watcher信息出现session expired是否会丢失？

针对第一个问题，需要在使用中详细阅读下对应Zookeeper类中方法的javadoc，注意几个点：

• exists方法： 设置watcher时，如果对应服务端已经不存在node时，watcher是不会留在服务端，下次不会被触发。针对这种情况需要判断返回的stat == null来进行处理
• getChildren方法： 和exist一样，需要处理节点不存在时watcher不会被记录。 还有一个点，当前的父node发生delete变化时，也可以得到触发
• getData方法： 和exist一样，需要处理节点不存在时watcher不会被记录

要了解watcher是否会丢失，必须要清楚zookeeper整套watcher机制的实现：

• Watcher是一个本地jvm的callback，在和服务端交互过程中是不会进行传递的。只是会将是否有watcher的boolean变量传递给server端
• 在服务端，在FinalRequestProcessor处理对应的node操作时，会根据客户端传递的watcher变量，添加到对应的zkDataBase中进行持久化存储，同时将自己NIOServerCnxn做为一个Watcher callback，监听服务端事件变化
• leader通过投票通过了某次node变化请求后，通知给对应的follower，follower根据自己内存中的zkDataBase信息，发送notification信息给zookeeper 客户端
• zookeeper客户端接收到notification信息后，找到对应变化path的watcher列表，挨个进行触发回调。

整个流程图：

可能存在的问题：

1. client向连接的server提交了watcher事件后，对应的server还未来得及提交给leader就直接出现了jvm crash，这时对应的watcher事件会丢失。(理论上正常关闭zookeeper server，不会存在该问题，需要客户端进行重试处理)

2. client在发生一次failover时，可以自动对新的server的notification使用watcher set，可以通过设置jvm变量：zookeeper.disableAutoWatchReset进行，默认为false。如果为true，则不进行自动使用的行为)

3. client出现session expired时，需要重新创建一个zookeeper client实例，此时对应的watcher set也会丢失，需要自己编码做一些额外的处理

zookeeper异常处理

官方文档：http://wiki.apache. org/hadoop/ZooKeeper/FAQ

主要处理两个系统异常：

 

• KeeperException.ConnectionLossException (client与其中的一台server socket链接出现异常)
• KeeperException.SessionExpiredException (client的session超过sessionTimeout为进行任何操作)

ConnectionLossException可以通过重试进行处理，在ClientCnxn会根据你初始化ZooKeeper时传递的服务列表，自动尝试下一个server节点

SessionExpiredException不能通过重试进行解决，需要应用重新new Zookeeper()，创建一个新的客户端，包括重新初始化对应的watcher，EPHEMERAL节点等。

 

最后

思路可能写的有点乱，文中所有的内容均通过阅读源码所得，如有不对的地方，尽情拍砖。

评论

3 楼 seancheer 2018-02-26  

qianshangding 写道

首先节点启动后，尝试读取本地的SnapShot log数据(zkDb.loadDataBase())，反序列化为DataTree对象，并获取last zxid。

服务重启的时候，应该是TnxLog配合SnapShot 一起，完成数据刷新吧？
因为有些节点的变化在SnapShot 是没有的。

是的，从snapshot中取出最大的zxid后，会结合tnxLog中的值来刷新最大的zxid，并回放tnxLog中的操作

      //根据snapshot获取的最大的zxid来读取txnLog文件,txnLog的命名方式是前缀为log的文件
        TxnIterator itr = txnLog.read(dt.lastProcessedZxid+1);
        long highestZxid = dt.lastProcessedZxid; //从snapshot文件中获取到的当前节点最大的zxid
        TxnHeader hdr;
        try {
            while (true) {
                // iterator points to
                // the first valid txn when initialized
                hdr = itr.getHeader();
                if (hdr == null) {
                    //empty logs
                    return dt.lastProcessedZxid;
                }
                if (hdr.getZxid() < highestZxid && highestZxid != 0) {
                    LOG.error("{}(higestZxid) > {}(next log) for type {}",
                            new Object[] { highestZxid, hdr.getZxid(),
                                    hdr.getType() });
                } else {
                    highestZxid = hdr.getZxid(); // ATTENTION!!!  从这里可以看出来，结合txnLog，取最大的zxid
                }
              try {
                    processTransaction(hdr,dt,sessions, itr.getTxn());
                }

2 楼 qianshangding 2013-11-29  

首先节点启动后，尝试读取本地的SnapShot log数据(zkDb.loadDataBase())，反序列化为DataTree对象，并获取last zxid。

服务重启的时候，应该是TnxLog配合SnapShot 一起，完成数据刷新吧？
因为有些节点的变化在SnapShot 是没有的。

1 楼 hustgaoguiju 2013-04-23  

顶        顶