zookeeper 客户端编程

时间:2021-11-12 16:15:12

zookeeper是一个分布式的开源的分布式协调服务,用它可以来现同步服务,配置维护。zookeeper的稳定性也是可以保证的,笔者曾参与过的使用zookeeper的两个应用,一个是用zookeeper来做分布式锁,属于同步协调服务,另一个是配置维护,三台机器运行了一年多了,没有出现什么问题。

运用场景:
------------------------
同步服务:利用zookeeper可以使机器之间的同步如同concurrent包下的同步器同步线程一样,举两个互斥同步的例子

1.  应用系统部署在多台机器上,需要启动一个时间任务,而这个时间任务只能在一台机器上执行,否则会出现脏读脏写,zookeeper可以实现。
2.  应用系统部署在多台机器上,需要暴露一个服务,而这个服务同样不是全部的机器都有暴露,例如MASTER,而暴露这个服务的机器宕机后,其它的机器能自动的接替暴露这个服务,zookeeper可以实现。但是这协调中有一定的时间差问题。
------------------------
配置维护:服务集群需要使用一组配置,为了保证配置在集群中的一致性,以及对配置更改后的立即响应要求,使用zookeeper可以实现。hadoop的配置中心就是使用zookeeper来做管理的。
------------------------
zookeeper应用机器结构

zookeeper 客户端编程

客户端根据服务器地址随机的选择一个地址进行连接,如果连接出错会连接其他的地址,客户端一直连接这个地址知道这个服务器不可用。客户端发送的指令,所有修改数据的指令都由其连接的服务器发送给LEADER进行处理,查询指令则在其连接的机器上进行。
zookeeper数据组织形式

zookeeper 客户端编程

zookeeper上的数据结构类似于文件结构,每一个节点上都可以写入数据,可以创建叶子节点,也只能删除叶子节点同时可以对任意路径进行监听各种类型的事件,当该事件在该路径发生时,连接的服务器会发送NOTIFY包过来通知客户端进行注册的WATCHER的回调处理。

zookeeper上的数据操作主要有CREATE,DELETE,EXISTS,GETDATA,SETDATA,GETCHILDREN
-------------------
CREATE创建一个节点,并在这个节点处写入数据,创建节点有四种模式
1. PERSISTENT 创建持久化节点,对应机器关闭连接后节点/数据不会消失
2. PERSISTENT_SEQUENTIAL 如果PATH是以’/’结尾则以这个PATH作为父节点,创建一个子节点,其子节点名字是一个按先后顺序排列的数值;否则创建一个名字是 ‘/’后面字符加上先后顺序排列的数值字符串的节点,同样创建持久节点,例如:
path-------- PERSISTENT-------- PERSISTENT_SEQUENTIAL
/lock/------ error-----------------在/lock/下创建子节点: /lock/000001
/lock-------创建/lock节点-------创建节点 /lock000001
3. EPHEMERAL 创建瞬时节点,Zookeeper在感知连接机器宕机后会清除它创建的瞬节点
4. EPHEMERAL_SEQUENTIAL 穿件瞬时顺序节点,和PERSISTENT_SEQUENTIAL一样,区别在于它是瞬时的
-------------------
DELETE  -  删除一个节点
EXISTS  -  查看一个节点的状况,如果没有返回null,可注入watcher
GETDATA  -  获取一个节点的数据,可注入watcher
SETDATA  -  设置一个节点的数据
GETCHILDREN -  获取一个节点的子节点,可注入watcher
-------------------

Watcher种类
1. zookeeper实例化时注入的默认Watcher
2. dataWatchers 一个Map<string Set<Watcher>>数据结构,保存调用getData时 注入的Watcher或者调用exist时path指定的节点存在
3. existWatchers 一个Map<string Set<Watcher>>数据结构,保存调用exits时对应的path节点不存在条件下注入的Watcher
4. childWatchers一个Map<string Set<Watcher>>数据结构,保存调用getChildren 时注入的Watcher
默认Watcher主要是处理连接建立成功,连接断了-会重连,SESSION失效-这个需要重新实例zookeeper对象这些事件

其他的三个Watcher都是MAP<String, SET<Watcher>>结构的,MAP中的KEY是zookeeper数据路径,当调用exists,getData,getChildren函数时会需要传入一个Watcher,对应的path上的数据发生改变时,zookeeper服务器会发送NOTIFY包给客户端,客户端在收到NOTIFY包后根据包指明的path和事件类别回调有关Watcher的process函数。
所有需要系统的管理这些Watcher,故分成三个MAP来管理这些Watcher。
当调用getData时将注入的Watcher加入到dataWatchers.get(path)的SET里面,调用childWatchers时一样。需要重点注意的是当调用exists时,如果path不存在则加入到existWatchers.get(path)的SET里面,否则加入到dataWatchers.get(path)的SET里面,这样做的原因是为了保持和客户端接收到NOTIFY包时处理一致。

事件处理
NodeDataChanged & NodeCreated
回调dataWatchers.get(path) & existWatchers.get(path)得到的Watcher集合,同时会将path这一MAP映射从dataWatchers&existWatchers里面清除
---------
NodeChildrenChanged
回调childWatchers.get(path得到的Watcher集合,同时会将path这一MAP映射从childWatchers.get里面清除
--------- 
NodeDeleted
回调dataWatchers.get(path) & childWatchers.get(path)得到的Watcher集合,同时会将path这一MAP映射从dataWatchers& childWatchers里面清除,同时会将path这一MAP映射从existWatchers里面清除
上面提到过对于exists的Watcher,为了保证Watcher能被回调,如果节点原本不存在的话,那么这个节点就只会有NodeCreated事件,故加入到existWatchers能找到,如果存在的话就可能是数据修改和删除事件了,如果是数据修改同样可以监听到,存在是加入到dataWatchers里,同样可以监听到。
----------------------------------------------------
zookeeper客户端的一些细节问题

建立连接需要指定zookeeper集群的IP地址,SessionTimeout,默认的watcher。
在建立连接过程中根据集群的IP地址随机的选择一个地址,然后启动两个线程,一个做网络,一个处理事件回调的。
做网络线程SendThread建立网络连接是使用的非阻塞模式,当连接建立成功后,会把一个ConnectRequest包加入到发送队列中,其中包括了客户端自己可能制定的sessionId, sessionPasswd,和自己指定的sessionTimeout,如果需要验证信息的话,验证信息包也会加入到发送队列中。再到Selector上注册OP_READ|OP_WRITE事件;如果连接没建立
成功则等待Selector上OP_CONNECT事件,直到连接建立成功,进行上面提到的连接建立成功后的操作。
到目前为止ZooKeeper.States还是CONNECTING,接下来就是调用doIO()做读写操作,当前服务器是不会发送数据过来的,故没有数据可读,当时发送队列里面已经包含至少一个数据包了,故将其发送出去。
接着就是读取服务器的对连接包和验证信息的响应数据了,首先会读取响应数据里面的negotiatedSessionTimeout值,如果小于等于零,会做四件事,一是往事件队列里面添加Watcher.Event.KeeperState.Expired事件,再往事件队列里面添加标记线程死亡的对象,三是抛出SessionExpiredException异常,再就是ZooKeeper.States=CLOSED,三让做死循环逻辑的SendThread退到while语句处从而退出死循环,线程死掉。
如果negotiatedSessionTimeout是大于0的则进行本地逻辑数据的设置readTimeout = negotiatedSessionTimeout * 2 / 3;
connectTimeout = negotiatedSessionTimeout / serverAddrs.size();
接着zooKeeper.state=CONNECTED。连接才真正建立成功,到这个点zookeeper提供的操作才可以使用。
接着SendThread走其正常的逻辑
--------------
1. 如果由于异常导致先前的连接断开,会重新走上面提到的建立连接的过程
2. 进行连接空闲事件或者READ空闲的事件的判断,如果zooKeeper.state不是CONNECTED,则用上次做写操作的时间和当前时间的差值和connectTimeout作比较,如果大于的话会抛出SessionTimeoutException;如果上次读的时间和当前时间差值大于readTimeout 同样抛出SessionTimeoutException,这样通知上层其正在做重连操作,同时通知上层其正在等待响应的数据包和没有来得及发送的数据包。
3. 如果上次读的时间和当前时间差值大于readTimeout/2则会做发送PING数据包的操作。
4. 如果当前可读,则读取响应数据包,根据响应的数据类型进行处理,共分为四种,一种是对PING的响应;一种是对于先前提到的验证的响应包,如果验证失败和Expired一样会导致SendThread退出;一种是NOTIFY数据包,里面指定了路径和事件类型,根据上面提到的事件处理方式,往事件处理任务队列里添加这个事件给EventThread线程处理;一种是用户操作返回的响应数据包,对于这种数据包要核对这个包的ID号和等待响应的数据包的队列的第一个包的ID是否一致,如果一致则进行关于这个包的下一步的处理,如包含Watcher则加入到上面提到的Watcher管理数据结构里面,再从等待响应的数据包的队列里移除,对于zookeeper数据结构的操作分为同步和异步,如果是同步操作,提交操作数据包的线程还需要在此次唤醒,如果是异步则向事件处理任务队列里添加回调任务;如果ID不一致则说明发送了错误,则抛出IOException异常,导致重新走上面提到的建立连接的过程。
zookeeper对于当个客户端提交的任务是FIFO的,用来保证其[sequential consistency for single client]。
5. 如果当前发送任务队列有数据则做发送操作,在发送时会为每个数据分配全局唯一的ID号,在发送完成后将这个包移动到已经发送在等待回应的队列中去。
6. 做完IO后在回到1之前会进行是否退出的判断,如果zooKeeper.state是CLOSED|AUTH_FAILED会退出,否则回到1。
--------------
1. 在读取数据的过程中如果发送IO错误,或者逻辑错误而抛出的异常,都会被抓捕到,进行后续处理:
关闭当前的Socket的输入输出流和Socket,对于还在带发送队列里面和在已经发送出去的数据包,如果是同步的需要唤醒这些提交包的线程,如果是异步的则将通知任务失败的任务提交到EventThread的任务队列里面,再清除这两个队列。如果不是因为Expired或者验证失败则会通知默认的Watcher其重连,如果是因为这两个事件,这个zookeeper对象已经不可用了,不会再重连了,客户端在接受到这两个事件时需要重新实例化zookeeper对象。
2. Watcher在被回调一次后会再 Watcher管理器中删除,如果想继续监听这个事件,需要继续注入。
3. 注意临时数据的问题,如果客户端创建了一个临时数据,然后重启了机器,在短时间内,这个时间跟服务器配置的SESSIONTIMEOUT有关,是存在的,过了这个时间点就会消失,这个时间要注意。