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，准備工作完成，一切盡在代碼中：

 

#!/usr/bin/env python2.7
# -*- coding:utf-8 -*-
#
#   Author  :   yunjianfei
#   E-mail  :   [e-mail]
#   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到多個服務器，同時運行，就可以看到分布式鎖的效果了。