Zookeeper接口kazoo實(shí)例解析

更新時(shí)間：2018年01月22日 09:50:09 作者：run4life

這篇文章主要介紹了Zookeeper接口kazoo實(shí)例解析，簡(jiǎn)單介紹了Zookeeper接口接口的開(kāi)發(fā)，然后分享了相關(guān)實(shí)例，小編覺(jué)得還是挺不錯(cuò)的，具有一定借鑒價(jià)值，需要的朋友可以參考下

本文主要研究的是Zookeeper接口kazoo的相關(guān)內(nèi)容，具體介紹如下。

zookeeper的開(kāi)發(fā)接口以前主要以java和c為主，隨著python項(xiàng)目越來(lái)越多的使用zookeeper作為分布式集群實(shí)現(xiàn)，python的zookeeper接口也出現(xiàn)了很多，現(xiàn)在主流的純python的zookeeper接口是kazoo。因此如何使用kazoo開(kāi)發(fā)基于python的分布式程序是必須掌握的。

1.安裝kazoo

yum install python-pip
pip install kazoo

安裝過(guò)程中會(huì)出現(xiàn)一些python依賴(lài)包未安裝的情況，安裝即可。

2.運(yùn)行kazoo基礎(chǔ)例子kazoo_basic.py

import time
from kazoo.client import KazooClient
from kazoo.client import KazooState
def main():
 zk=KazooClient(hosts='127.0.0.1:2182')
 zk.start()
&nbsp;
 @zk.add_listener
 def my_listener(state):
 if state == KazooState.LOST:
  print("LOST")
 elif state == KazooState.SUSPENDED:
  print("SUSPENDED")
 else:
  print("Connected")
 #Creating Nodes
 # Ensure a path, create if necessary
 zk.ensure_path("/my/favorite")
 # Create a node with data
 zk.create("/my/favorite/node", b"")
 zk.create("/my/favorite/node/a", b"A")
 #Reading Data
 # Determine if a node exists
 if zk.exists("/my/favorite"):
 print("/my/favorite is existed")
 @zk.ChildrenWatch("/my/favorite/node")
 def watch_children(children):
 print("Children are now: %s" % children)
 # Above function called immediately, and from then on
 @zk.DataWatch("/my/favorite/node")
 def watch_node(data, stat):
 print("Version: %s, data: %s" % (stat.version, data.decode("utf-8")))
 # Print the version of a node and its data
 data, stat = zk.get("/my/favorite/node")
 print("Version: %s, data: %s" % (stat.version, data.decode("utf-8")))
 # List the children
 children = zk.get_children("/my/favorite/node")
 print("There are %s children with names %s" % (len(children), children))
 #Updating Data
 zk.set("/my/favorite", b"some data")
 #Deleting Nodes
 zk.delete("/my/favorite/node/a")
 #Transactions
 transaction = zk.transaction()
 transaction.check('/my/favorite/node', version=-1)
 transaction.create('/my/favorite/node/b', b"B")
 results = transaction.commit()
 print ("Transaction results is %s" % results)
 zk.delete("/my/favorite/node/b")
 zk.delete("/my", recursive=True)
 time.sleep(2)
 zk.stop()
if __name__ == "__main__":
 try:
 main()
 except Exception, ex:
 print "Ocurred Exception: %s" % str(ex)
 quit()

運(yùn)行結(jié)果：

Children are now: [u'a']
Version: 0, data:&nbsp;
Version: 0, data:&nbsp;
There are 1 children with names [u'a']
Children are now: []
Transaction results is [True, u'/my/favorite/node/b']
Children are now: [u'b']
Children are now: []
No handlers could be found for logger "kazoo.recipe.watchers"
LOST

以上程序運(yùn)行了基本kazoo接口命令，包括創(chuàng)建刪除加watcher等操作，通過(guò)調(diào)試并對(duì)比zookeeper服務(wù)節(jié)點(diǎn)znode目錄結(jié)構(gòu)的變化，就可以理解具體的操作結(jié)果。

3.運(yùn)行通過(guò)kazoo實(shí)現(xiàn)的分布式鎖程序kazoo_lock.py

import logging, os, time
from kazoo.client import KazooClient
from kazoo.client import KazooState
from kazoo.recipe.lock import Lock

class ZooKeeperLock():
 def __init__(self, hosts, id_str, lock_name, logger=None, timeout=1):
 self.hosts = hosts
 self.id_str = id_str
 self.zk_client = None
 self.timeout = timeout
 self.logger = logger
 self.name = lock_name
 self.lock_handle = None
 self.create_lock()
 def create_lock(self):
 try:
 self.zk_client = KazooClient(hosts=self.hosts, logger=self.logger, timeout=self.timeout)
 self.zk_client.start(timeout=self.timeout)
 except Exception, ex:
 self.init_ret = False
 self.err_str = "Create KazooClient failed! Exception: %s" % str(ex)
 logging.error(self.err_str)
 return
 try:
 lock_path = os.path.join("/", "locks", self.name)
 self.lock_handle = Lock(self.zk_client, lock_path)
 except Exception, ex:
 self.init_ret = False
 self.err_str = "Create lock failed! Exception: %s" % str(ex)
 logging.error(self.err_str)
 return
 def destroy_lock(self):
 #self.release()
 if self.zk_client != None:
 self.zk_client.stop()
 self.zk_client = None
 def acquire(self, blocking=True, timeout=None):
 if self.lock_handle == None:
 return None
 try:
 return self.lock_handle.acquire(blocking=blocking, timeout=timeout)
 except Exception, ex:
 self.err_str = "Acquire lock failed! Exception: %s" % str(ex)
 logging.error(self.err_str)
 return None
 def release(self):
 if self.lock_handle == None:
 return None
 return self.lock_handle.release()
 def __del__(self):
 self.destroy_lock()

def main():
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
 sh = logging.StreamHandler()
 formatter = logging.Formatter('%(asctime)s -%(module)s:%(filename)s-L%(lineno)d-%(levelname)s: %(message)s')
 sh.setFormatter(formatter)
 logger.addHandler(sh)
 zookeeper_hosts = "127.0.0.1:2182"
 lock_name = "test"
 lock = ZooKeeperLock(zookeeper_hosts, "myid is 1", lock_name, logger=logger)
 ret = lock.acquire()
 if not ret:
 logging.info("Can't get lock! Ret: %s", ret)
 return
 logging.info("Get lock! Do something! Sleep 10 secs!")
 for i in range(1, 11):
 time.sleep(1)
 print str(i)
 lock.release()

if __name__ == "__main__":
 try:
 main()
 except Exception, ex:
 print "Ocurred Exception: %s" % str(ex)
 quit()

將該測(cè)試文件copy到多個(gè)服務(wù)器，同時(shí)運(yùn)行，就可以看到分布式鎖的效果了。

總結(jié)

以上就是本文關(guān)于Zookeeper接口kazoo實(shí)例解析的全部?jī)?nèi)容，希望對(duì)大家有所幫助。感興趣的朋友可以繼續(xù)參閱本站其他相關(guān)專(zhuān)題，如有不足之處，歡迎留言指出。感謝朋友們對(duì)本站的支持！

您可能感興趣的文章: