ZooKeeper之分布式锁（Python版）

============================================================================
原创作品，允许转载。转载时请务必以超链接形式标明原始出处、以及本声明。

请注明转自：http://yunjianfei.iteye.com/blog/


============================================================================


前言
　　在做分布式系统开发的时候，分布式锁可以说是必需的一个组件。最近做了一些调研和尝试，经过对比，基于ZooKeeper的分布式锁还是很不错的。
　　参照了IBM的一个帖子：https://www.ibm.com/developerworks/cn/opensource/os-cn-zookeeper/

其中有一段话描述了ZooKeeper的共享锁（即分布式锁）实现，如下：

共享锁在同一个进程中很容易实现，但是在跨进程或者在不同 Server 之间就不好实现了。Zookeeper 却很容易实现这个功能，实现方式也是需要获得锁的 Server 创建一个 EPHEMERAL_SEQUENTIAL 目录节点，然后调用 getChildren方法获取当前的目录节点列表中最小的目录节点是不是就是自己创建的目录节点，如果正是自己创建的，那么它就获得了这个锁，如果不是那么它就调用 exists(String path, boolean watch) 方法并监控 Zookeeper 上目录节点列表的变化，一直到自己创建的节点是列表中最小编号的目录节点，从而获得锁，释放锁很简单，只要删除前面它自己所创建的目录节点就行了。

 通过这段话，大概可以明白其原理。下面我主要写一下基于Python的分布式锁实现。

实现
　　Google了一下，有个叫Kazoo的python开源包很好的实现了对ZooKeeper的支持。

Kazoo is a Python library designed to make working with Zookeeper a more hassle-free experience that is less prone to errors.
　　链接如下：https://kazoo.readthedocs.org/en/latest/
　　GitHub地址: https://github.com/python-zk/kazoo
　　首先，我们去GitHub，下载其源码包。解压缩之后，进行安装

python setup.py install
　　OK，准备工作完成，一切尽在代码中：
　　文件名：zk_lock.py

#!/usr/bin/env python2.7
# -*- coding:utf-8 -*-
#
#   Author  :   yunjianfei
#   E-mail  :   yunjianfei1987@gmail.com
#   Date    :   2014/12/09
#   Desc    :
#
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 = "192.168.10.2:2181, 192.168.10.3:2181, 192.168.10.4:2181"
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()

　　测试的时候，只需要改一下“zookeeper_hosts ”这个参数，改为你自己的ZooKeeper的server地址即可.
　　将该测试文件copy到多个服务器，同时运行，就可以看到分布式锁的效果了。