在 Redis 上實現的分佈式鎖【轉載】

　　因爲近排很忙，忙各類事情，還有工做上的項目，已經超過一個月沒寫博客了，確實有點慚愧啊，沒能天天或者至少每週堅持寫一篇博客。這一個月裏面接觸到不少新知識，同時也遇到不少技術上的難點，在這我將對每個有用的技術點作一個小小的分析理解和總結。天天去學會總結，纔會有進步。

　　本次對我在工做上的項目中用到的技術---在redis上實現分佈式鎖，進行一個分析和總結。

　　先了解下什麼時分佈式鎖，在百度上是這麼定義的：

　　　　分佈式鎖是控制分佈式系統之間同步訪問共享資源的一種方式。在分佈式系統中，經常須要協調他們的動做。若是不一樣的系統或是同一個系統的不一樣主機之間共享了一個或一組資源，那麼訪問這些資源的時候，每每須要互斥來防止彼此干擾來保證一致性，在這種狀況下，便須要使用到分佈式鎖。

　　簡單的理解就是：分佈式鎖是一個在不少環境中很是有用的原語，它是不一樣的系統或是同一個系統的不一樣主機之間互斥操做共享資源的有效方法。

　　背景：

　　　　在不少互聯網產品應用中，有些場景須要加鎖處理，好比：秒殺，全局遞增ID，樓層生成等等。大部分是解決方案基於DB實現的，Redis爲單進程單線程模式，採用隊列模式將併發訪問變成串行訪問，且多客戶端對Redis的鏈接並不存在競爭關係。

　　咱們的項目：

　　咱們如今的項目中，任務隊列用到分佈式鎖的狀況比較多，在將業務邏輯中能夠異步處理的操做放入隊列，在其餘線程中處理後出隊，此時隊列中使用了分佈式鎖，保證入隊和出隊的一致性。關於redis隊列這塊的邏輯分析，我將在下一次對其進行總結，此處先略過。

 

　　接下來對redis實現的分佈式鎖的邏輯代碼進行詳細的分析和理解：

　　一、爲避免特殊緣由致使鎖沒法釋放, 在加鎖成功後, 鎖會被賦予一個生存時間(經過 lock 方法的參數設置或者使用默認值), 超出生存時間鎖將被自動釋放.
　　二、鎖的生存時間默認比較短(秒級, 具體見 lock 方法), 所以若須要長時間加鎖, 能夠經過 expire 方法延長鎖的生存時間爲適當的時間. 好比在循環內調用 expire
　　三、系統級的鎖當進程不管由於任何緣由出現crash，操做系統會本身回收鎖，因此不會出現資源丟失。
　　四、但分佈式鎖不一樣。若一次性設置很長的時間，一旦因爲各類緣由進程 crash 或其餘異常致使 unlock 未被調用，則該鎖在剩下的時間就變成了垃圾鎖，致使其餘進程或進程重啓後沒法進入加鎖區域。

<?php

require_once 'RedisFactory.php';

/**
* 在 Redis 上實現的分佈式鎖
*/
class RedisLock {
    //單例模式
    private static $_instance = null;
    public static function instance() {
        if(self::$_instance == null) {
            self::$_instance = new RedisLock();
        }
        return self::$_instance;
    }

    //redis對象變量
    private $redis;
    //存放被鎖的標誌名的數組
    private $lockedNames = array();

    public function __construct() {
        //獲取一個 RedisString 實例
        $this->redis = RedisFactory::instance()->getString();
    }

    /** 
    * 加鎖
    *
    * @param string 鎖的標識名
    * @param int 獲取鎖失敗時的等待超時時間(秒), 在此時間以內會一直嘗試獲取鎖直到超時. 爲 0 表示失敗後直接返回不等待
    * @param int 當前鎖的最大生存時間(秒), 必須大於 0 . 若是超過生存時間後鎖仍未被釋放, 則系統會自動將其強制釋放
    * @param int 獲取鎖失敗後掛起再試的時間間隔(微秒)
    */
    public function lock($name, $timeout = 0, $expire = 15, $waitIntervalUs = 100000) {
        if(empty($name)) return false;

        $timeout = (int)$timeout;
        $expire = max((int)$expire, 5);
        $now = microtime(true);
        $timeoutAt = $now + $timeout;
        $expireAt = $now + $expire;

        $redisKey = "Lock:$name";
        while(true) {
            $result = $this->redis->setnx($redisKey, (string)$expireAt);
            if($result !== false) {
                //對$redisKey設置生存時間
                $this->redis->expire($redisKey, $expire);
                //將最大生存時刻記錄在一個數組裏面
                $this->lockedNames[$name] = $expireAt;
                return true;
            }

            //以秒爲單位，返回$redisKey 的剩餘生存時間
            $ttl = $this->redis->ttl($redisKey);
            // TTL 小於 0 表示 key 上沒有設置生存時間(key 不會不存在, 由於前面 setnx 會自動建立)
            // 若是出現這種狀況, 那就是進程在某個實例 setnx 成功後 crash 致使緊跟着的 expire 沒有被調用. 這時能夠直接設置 expire 並把鎖納爲己用
            if($ttl < 0) {
                $this->redis->set($redisKey, (string)$expireAt, $expire);
                $this->lockedNames[$name] = $expireAt;
                return true;
            }

            // 設置了不等待或者已超時
            if($timeout <= 0 || microtime(true) > $timeoutAt) break;

            // 掛起一段時間再試
            usleep($waitIntervalUs);
        }

        return false;
    }

    /**
    * 給當前鎖增長指定的生存時間(秒), 必須大於 0
    *
    * @param string 鎖的標識名
    * @param int 生存時間(秒), 必須大於 0
    */
    public function expire($name, $expire) {
        if($this->isLocking($name)) {
            if($this->redis->expire("Lock:$name", max($expire, 1))) {
                return true;
            }
        }
        return false;
    }

    /**
    * 判斷當前是否擁有指定名稱的鎖
    *
    * @param mixed $name
    */
    public function isLocking($name) {
        if(isset($this->lockedNames[$name])) {
            return (string)$this->lockedNames[$name] == (string)$this->redis->get("Lock:$name");
        }
        return false;
    }

    /**
    * 釋放鎖
    *
    * @param string 鎖的標識名
    */
    public function unlock($name) {
        if($this->isLocking($name)) {
            if($this->redis->deleteKey("Lock:$name")) {
                unset($this->lockedNames[$name]);
                return true;
            }
        }
        return false;
    }

    /** 釋放當前已經獲取到的全部鎖 */
    public function unlockAll() {
        $allSuccess = true;
        foreach($this->lockedNames as $name => $item) {
            if(false === $this->unlock($name)) {
                $allSuccess = false;
            }
        }
        return $allSuccess;
    }
}

　　此類不少代碼都寫上了註釋，只要認真理解下，就很容易懂得如何在redis實現分佈式鎖了。

　　另外，我在網上找到另外一篇關於redis實現分佈式鎖的文章，我感受挺不錯的，推薦給你們：

　　網址：　http://www.oschina.net/translate/redis-distlock

　　結合我所總結的和我推薦的文章作對比，基本上能理解清楚是如何在redis實現分佈式鎖的了。

　　若是此博文中有哪裏講得讓人難以理解，歡迎留言交流，如有講解錯的地方歡迎指出。