Python与ZooKeeper集群连接

由于项目的需要，需要学习Python客户端连接ZooKeeper集群，并实现创建临时节点、获得指定的路径下的信息、监听子节点变化的功能。

环境配置

ZooKeeper集群的安装可以参考http://blog.csdn.net/mrbcy/article/details/54767484

使用下面的命令安装kazoo

pip install kazoo

基本使用

这一部分可参考官方文档：http://kazoo.readthedocs.io/en/latest/basic_usage.htm

监听子节点变化

下面的代码实现了创建一个临时、顺序的节点，并且可以监听子节点的变化。

#-*- coding: utf-8 -*- import time from kazoo.client import KazooClient from kazoo.recipe.watchers import ChildrenWatch class ValidatorDetector: def __init__(self): self.zk = KazooClient(hosts='amaster:2181,anode1:2181,anode2:2181') self.validator_children_watcher = ChildrenWatch(client=self.zk,path='/mproxy/validators',func=self.validator_watcher_fun) self.zk.start() def validator_watcher_fun(self,children): print "The children now are:", children def create_node(self): self.zk.create('/mproxy/validators/validator',b'validator_huabei_1',ephemeral=True,sequence=True,makepath=True) def __del__(self): self.zk.close() if __name__ == '__main__': detector = ValidatorDetector() detector.create_node() time.sleep(10)

ZooKeeper原生提供了监听节点变化及值的变化的API。关于这一部分可以参考http://blog.csdn.net/mrbcy/article/details/54790758。但是这些API只能生效一次，一旦被触发过一次以后就不会再触发了，除非再次注册。而kazoo则在这个基础上封装了更上层的API，可以持续的触发。这就是上面的ChildrenWatch，除此之外kazoo还封装了一个DataWatch，用于监听数据的变化。下面我们也会用到。

kazoo还实现了自动续订功能，使得在会话过期后我们不需要再次初始化ZooKeeper客户端（这里可以参考http://blog.csdn.net/mrbcy/article/details/55062713），也是非常方便的。

注册验证器

有了上面的知识就可以做一个注册类和一个监测类了。

#-*- coding: utf-8 -*- import threading import time from kazoo.client import KazooClient from kazoo.protocol.states import KazooState class InfoKeeper(threading.Thread): def __init__(self,register): threading.Thread.__init__(self) self.register=register def run(self): time.sleep(0.25) if self.register.zk_node is None: print "create method has not been called" return check_result = self.register.zk.exists(self.register.validator_path) if check_result is None: # redo the regist print "redo the regist" self.register.regist() else: print "the path remain exists" class ValidatorRegister: def __init__(self): self.zk = KazooClient(hosts='amaster:2181,anode1:2181,anode2:2181') self.zk_node = None self.validator_path = '/mproxy/validators/' self.zk.add_listener(self.conn_state_watcher) self.zk.start() def __del__(self): self.zk.close() def regist(self): self.zk_node = self.zk.create(self.validator_path + 'validator',bytes('validator_huabei_1'),ephemeral=True,sequence=True,makepath=True) def close(self): self.zk.stop() self.zk.close() def conn_state_watcher(self, state): if state == KazooState.CONNECTED: print "Now connected" if self.zk_node is None: print "create method has not been called" return info_keeper = InfoKeeper(self) info_keeper.start() elif state == KazooState.LOST: print "Now lost" else: print "Now suspended"

监测类：

#-*- coding: utf-8 -*- import time from kazoo.client import KazooClient from kazoo.recipe.watchers import ChildrenWatch class ValidatorDetector: def __init__(self): self.validator_path = '/mproxy/validators/' self.zk = KazooClient(hosts='amaster:2181,anode1:2181,anode2:2181') self.validator_children_watcher = ChildrenWatch(client=self.zk,path=self.validator_path,func=self.validator_watcher_fun) self.zk.start() def validator_watcher_fun(self,children): for child in children: validator_name = self.zk.get(path=self.validator_path + str(child)) print validator_name[0] print "The children now are:", children def __del__(self): self.zk.close()

注册类这里稍微复杂了一点，做了一个在会话过期后重新注册的机制，如果会话过期，重新注册之前的注册信息。

监听子节点值的变化

嗯，这个需求仔细想过后可以通过监听子节点的变化来代替，所以暂时不实现了。