Redis實現分佈式鎖

時間 2020-07-06

標籤 redis 實現分佈式欄目 Redis 简体版

原文原文鏈接

分佈式鎖的實現方式有不少，本篇文章講述一下使用Redis實現分佈式鎖。網上有不少使用Redis實現分佈式鎖的代碼，可是這些代碼或多或少都有問題。這篇文章會寫一個實現，同時標明一些注意點。php

場景

爲了便於闡述，這裏假設一個遊戲場景，用戶A有開山斧一把，價值500元寶，用戶B有800元寶，想買A的開山斧，這些數據都存在Redis中。須要編寫代碼成功的實現該筆交易。html

問題

Redis實現分佈式鎖，須要考慮以下問題：程序員

持有鎖的進程由於操做時間過長而致使鎖被自動釋放，但進程自己並不知曉這一點，甚至還可能會錯誤地釋放掉了其餘進程持有的鎖。
一個持有鎖並打算執行長時間操做的進程已經崩潰，但其餘想要獲取鎖的進程不知道哪一個進程持有着鎖，也沒法檢測出持有鎖的進程已經崩潰，只能白白地浪費時間等待鎖被釋放。
在一個進程持有的鎖過時以後，其餘多個進程同時嘗試去獲取鎖，而且都得到了鎖。

三個特性

實現一個最低保障的分佈式鎖，須要具有三個特性redis

安全屬性（Safety property）: 獨享（相互排斥）。在任意一個時刻，只有一個客戶端持有鎖。
活性A(Liveness property A): 無死鎖。即使持有鎖的客戶端崩潰（crashed)或者網絡被分裂（gets partitioned)，鎖仍然能夠被獲取。
活性B(Liveness property B): 容錯。只要大部分Redis節點都活着，客戶端就能夠獲取和釋放鎖.

命令

使用Redis實現分佈式鎖，通常使用SETNX或者SET命令，SETNX不能同時設置過時時間，若是使用的版本大於等於2.6.12，可使用SET命令，可使用這個命令原子性的實現SETNX和EXPIRE的功能，下面是兩個命令的簡介算法

SETNX

命令格式：SETNX key value緩存

時間複雜度：O(1)安全

說明：將key設置值爲value，若是key不存在，這種狀況下等同SET命令。當key存在時，什麼也不作。SETNX是」SET if Not eXists」的簡寫。網絡

返回值框架

1 若是key被設置了
0 若是key沒有被設置

SET

命令格式：SET key value [EX seconds] [PX milliseconds] [NX|XX]編輯器

時間複雜度：O(1)

說明：將鍵key設定爲指定的「字符串」值。若是 key 已經保存了一個值，那麼這個操做會直接覆蓋原來的值，而且忽略原始類型。當set命令執行成功以後，以前設置的過時時間都將失效。

選項

從2.6.12版本開始，redis爲SET命令增長了一系列選項:

EX seconds – 設置鍵key的過時時間，單位時秒
PX milliseconds – 設置鍵key的過時時間，單位是毫秒
NX – 只有鍵key不存在的時候纔會設置key的值
XX – 只有鍵key存在的時候纔會設置key的值

實現

此處使用SETNX實現，畢竟有的公司Redis版本可能較低，使用SETNX能夠實現，SET更加沒有問題。

代碼以下：

<?php

function uuid($prefix = '') {
    $chars = md5(uniqid(mt_rand(), true));
    $uuid  = substr($chars, 0, 8) . '-';
    $uuid .= substr($chars, 8, 4) . '-';
    $uuid .= substr($chars, 12, 4) . '-';
    $uuid .= substr($chars, 16, 4) . '-';
    $uuid .= substr($chars, 20, 12);
    $ret = $prefix . $uuid;
    return strtoupper($ret);
}

function acquireLock($redis,$lockName, $acquireTime = 10, $lockTime = 10) {
    $lockKey    = 'lock:' + $lockName;
    $identifier = uuid('identify');
    $end        = time() + $acquireTime;
    while (time() < $end) {
        if ($redis->setnx($lockKey, $identifier)) {
            $redis->expire($lockKey, $lockTime);
            return $identifier;
        } elseif ($redis->ttl($lockKey) == -1) {
            $redis->expire($lockKey, $lockTime);
        }
        usleep(1000);
    }
    return false;
}


function process(){
    $redis      = new Redis();
    $lockName = 'market';
    //1.獲取鎖
    $locked = acquireLock($redis,$lockName);
    if($locked === false){
        return false;
    }
    //2.進行交易
    //判斷A和B是否知足交易條件
    //使用管道，對A和B進行操做

    //3.釋放鎖
    $releaseRes = releaseLock($redis,$lockName,$locked);
    if($releaseRes === false){
        return false;
    }
}

function releaseLock($redis,$lockName,$identifier){
    $lockKey    = 'lock:' + $lockName;
    $redis->watch($lockKey);
    if($redis->get($lockKey) === $identifier){
        $redis->multi();
        $redis->del($lockKey);
        $redis->exec();
        return true;
    }
    $redis->unwatch();
    return false;
}
複製代碼

說明：

獲取鎖：
- 建立惟一的$identifier，這個值用於刪除key的時候，判斷是否爲當前客戶端獲取的鎖，以避免刪除其它客戶端的鎖
- while循環用於在一段時間內不停的獲取鎖
- 若是可以獲取鎖便獲取，同時設置超時時間，防止線程運行時崩潰，鎖永遠沒法釋放
- 若是沒能成功獲取到鎖，檢查當前鎖的過時時間，若是未設置過時時間，進行設置，防止其餘線程得到鎖後當即崩潰，沒有設置過時時間
處理業務
- 須要先判斷交易雙方是否都知足條件，由於鎖定的是整個市場，因此一旦得到鎖，交易雙方的狀態都不會再進行改變
- 使用管道能保證整個交易能像事務同樣被處理，並且性能會比用redis的事務更好
釋放鎖
- 使用watch監控鎖，一旦key被變動，刪除key的事務不會被執行
- 須要判斷key的值是否和本線程記錄的$identifier同樣，只有一致才能進行刪除
- 使用事務來作刪除key的操做，使用事務的緣由是防止中途該鎖被別的線程獲取
- 若是失敗，記得unwatch
其餘問題
- 可重入問題：可重入指的是，線程能夠再次獲取到鎖。實現方法比較簡單，只須要在acquireLock的時候，傳入identifier，判斷當前鎖的identifier和傳入的是否一致，若是一致則能夠進行操做
- 線程未執行完畢，鎖的超時時間已過，其餘線程獲取到鎖：解決該問題的一個方案是，當獲取到鎖後，在超時時間通過一半的時候檢查鎖是否存在或者被修改，若是沒有變化且線程正常運行，則延長超時時間