本文主要讲python支持zookeeper的接口库安装和使用。zk的python接口库有zkpython，还有kazoo，下面是zkpython，是基于zk的C库的python接口。

zkpython安装

前提是zookeeper安装包已经在/usr/local/zookeeper下

cd /usr/local/zookeeper/src/c./configuremakemake installwget --no-check-certificate http://pypi.python.org/packages/source/z/zkpython/zkpython-0.4.tar.gztar -zxvf zkpython-0.4.tar.gzcd zkpython-0.4sudo python setup.py install

zkpython应用

下面是网上一个zkpython的类，用的时候只要import进去就行
vim zkclient.py

#!/usr/bin/env python2.7# -*- coding: UTF-8 -*-import zookeeper, time, threadingfrom collections import namedtupleDEFAULT_TIMEOUT = 30000VERBOSE = TrueZOO_OPEN_ACL_UNSAFE = {"perms":0x1f, "scheme":"world", "id" :"anyone"}# Mapping of connection state values to human strings.STATE_NAME_MAPPING = { zookeeper.ASSOCIATING_STATE: "associating", zookeeper.AUTH_FAILED_STATE: "auth-failed", zookeeper.CONNECTED_STATE: "connected", zookeeper.CONNECTING_STATE: "connecting", zookeeper.EXPIRED_SESSION_STATE: "expired",}# Mapping of event type to human string.TYPE_NAME_MAPPING = { zookeeper.NOTWATCHING_EVENT: "not-watching", zookeeper.SESSION_EVENT: "session", zookeeper.CREATED_EVENT: "created", zookeeper.DELETED_EVENT: "deleted", zookeeper.CHANGED_EVENT: "changed", zookeeper.CHILD_EVENT: "child", }class ZKClientError(Exception): def __init__(self, value): self.value = value def __str__(self): return repr(self.value)class ClientEvent(namedtuple("ClientEvent", 'type, connection_state, path')): """ A client event is returned when a watch deferred fires. It denotes some event on the zookeeper client that the watch was requested on. """ @property def type_name(self): return TYPE_NAME_MAPPING[self.type] @property def state_name(self): return STATE_NAME_MAPPING[self.connection_state] def __repr__(self): return "<ClientEvent %s at %r state: %s>" % ( self.type_name, self.path, self.state_name)def watchmethod(func): def decorated(handle, atype, state, path): event = ClientEvent(atype, state, path) return func(event) return decoratedclass ZKClient(object): def __init__(self, servers, timeout=DEFAULT_TIMEOUT): self.timeout = timeout self.connected = False self.conn_cv = threading.Condition( ) self.handle = -1 self.conn_cv.acquire() if VERBOSE: print("Connecting to %s" % (servers)) start = time.time() self.handle = zookeeper.init(servers, self.connection_watcher, timeout) self.conn_cv.wait(timeout/1000) self.conn_cv.release() if not self.connected: raise ZKClientError("Unable to connect to %s" % (servers)) if VERBOSE: print("Connected in %d ms, handle is %d" % (int((time.time() - start) * 1000), self.handle)) def connection_watcher(self, h, type, state, path): self.handle = h self.conn_cv.acquire() self.connected = True self.conn_cv.notifyAll() self.conn_cv.release() def close(self): return zookeeper.close(self.handle) def create(self, path, data="", flags=0, acl=[ZOO_OPEN_ACL_UNSAFE]): start = time.time() result = zookeeper.create(self.handle, path, data, acl, flags) if VERBOSE: print("Node %s created in %d ms" % (path, int((time.time() - start) * 1000))) return result def delete(self, path, version=-1): start = time.time() result = zookeeper.delete(self.handle, path, version) if VERBOSE: print("Node %s deleted in %d ms" % (path, int((time.time() - start) * 1000))) return result def get(self, path, watcher=None): return zookeeper.get(self.handle, path, watcher) def exists(self, path, watcher=None): return zookeeper.exists(self.handle, path, watcher) def set(self, path, data="", version=-1): return zookeeper.set(self.handle, path, data, version) def set2(self, path, data="", version=-1): return zookeeper.set2(self.handle, path, data, version) def get_children(self, path, watcher=None): return zookeeper.get_children(self.handle, path, watcher) def async(self, path = "/"): return zookeeper.async(self.handle, path) def acreate(self, path, callback, data="", flags=0, acl=[ZOO_OPEN_ACL_UNSAFE]): result = zookeeper.acreate(self.handle, path, data, acl, flags, callback) return result def adelete(self, path, callback, version=-1): return zookeeper.adelete(self.handle, path, version, callback) def aget(self, path, callback, watcher=None): return zookeeper.aget(self.handle, path, watcher, callback) def aexists(self, path, callback, watcher=None): return zookeeper.aexists(self.handle, path, watcher, callback) def aset(self, path, callback, data="", version=-1): return zookeeper.aset(self.handle, path, data, version, callback)watch_count = 0"""Callable watcher that counts the number of notifications"""class CountingWatcher(object): def __init__(self): self.count = 0 global watch_count self.id = watch_count watch_count += 1 def waitForExpected(self, count, maxwait): """Wait up to maxwait for the specified count, return the count whether or not maxwait reached. Arguments: - `count`: expected count - `maxwait`: max milliseconds to wait """ waited = 0 while (waited < maxwait): if self.count >= count: return self.count time.sleep(1.0); waited += 1000 return self.count def __call__(self, handle, typ, state, path): self.count += 1 if VERBOSE: print("handle %d got watch for %s in watcher %d, count %d" % (handle, path, self.id, self.count))"""Callable watcher that counts the number of notificationsand verifies that the paths are sequential"""class SequentialCountingWatcher(CountingWatcher): def __init__(self, child_path): CountingWatcher.__init__(self) self.child_path = child_path def __call__(self, handle, typ, state, path): if not self.child_path(self.count) == path: raise ZKClientError("handle %d invalid path order %s" % (handle, path)) CountingWatcher.__call__(self, handle, typ, state, path)class Callback(object): def __init__(self): self.cv = threading.Condition() self.callback_flag = False self.rc = -1 def callback(self, handle, rc, handler): self.cv.acquire() self.callback_flag = True self.handle = handle self.rc = rc handler() self.cv.notify() self.cv.release() def waitForSuccess(self): while not self.callback_flag: self.cv.wait() self.cv.release() if not self.callback_flag == True: raise ZKClientError("asynchronous operation timed out on handle %d" % (self.handle)) if not self.rc == zookeeper.OK: raise ZKClientError( "asynchronous operation failed on handle %d with rc %d" % (self.handle, self.rc))class GetCallback(Callback): def __init__(self): Callback.__init__(self) def __call__(self, handle, rc, value, stat): def handler(): self.value = value self.stat = stat self.callback(handle, rc, handler)class SetCallback(Callback): def __init__(self): Callback.__init__(self) def __call__(self, handle, rc, stat): def handler(): self.stat = stat self.callback(handle, rc, handler)class ExistsCallback(SetCallback): passclass CreateCallback(Callback): def __init__(self): Callback.__init__(self) def __call__(self, handle, rc, path): def handler(): self.path = path self.callback(handle, rc, handler)class DeleteCallback(Callback): def __init__(self): Callback.__init__(self) def __call__(self, handle, rc): def handler(): pass self.callback(handle, rc, handler)

总结

以上就是本文关于zookeeper python接口实例详解的全部内容，希望对大家有所帮助。感兴趣的朋友可以继续参阅本站其他相关专题，如有不足之处，欢迎留言指出。感谢朋友们对本站的支持！