Python 官方文档:入门教程 => 点击学习
由于项目的需要,需要学习python客户端连接ZooKeeper集群,并实现创建临时节点、获得指定的路径下的信息、监听子节点变化的功能。 环境配置 ZooKeeper集群的安装可以参考Http://blog.csdn.net/mr
由于项目的需要,需要学习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()
注册类这里稍微复杂了一点,做了一个在会话过期后重新注册的机制,如果会话过期,重新注册之前的注册信息。
嗯,这个需求仔细想过后可以通过监听子节点的变化来代替,所以暂时不实现了。
--结束END--
本文标题: Python与ZooKeeper集群连接
本文链接: https://www.lsjlt.com/news/188873.html(转载时请注明来源链接)
有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
2024-03-01
2024-03-01
2024-03-01
2024-02-29
2024-02-29
2024-02-29
2024-02-29
2024-02-29
2024-02-29
2024-02-29
回答
回答
回答
回答
回答
回答
回答
回答
回答
回答
0