用分佈式鎖解決併發問題

在系統中，當存在多個進程和線程能夠改變某個共享數據時，就容易出現併發問題致使共享數據的不一致性。即多個進程同時獲取到了對數據的操做權限並對數據進行了更新，很典型的場景就是在線銷售系統在售賣熱銷商品時遇到多個併發請求在同一時間提交訂單的狀況則極有可能形成商品超賣的現象。只要訪問流量不錯的系統都有可能遭遇併發請求形成數據庫中數據重複寫入的狀況。

針對程序塊被多個進程併發執行問題的解決方案是確保同一個時刻同一個程序塊只能有一個進程可執行，其餘進程等待當前進程執行完成才能獲取程序塊的執行權對數據進行更新，以此類推將併發執行變爲串行順序執行。爲了讓獲取執行權的進程不被其餘干擾，就須要設置一個全部進程都能讀取到的標記，當標記不存在時能夠設置該標記，其他後續進程發現已經有標記了則等待擁有標記的進程結束執行程序塊取消標記後再去嘗試設置標記。這個標記能夠理解爲鎖，設置標記的過程就是咱們一般說的加鎖。

用redis 的 setnx、expire 方法作分佈式鎖

• setnx()

setnx 的含義就是 SET if Not Exists，其主要有兩個參數 setnx(key, value)。該方法是原子的，若是 key 不存在，則設置當前 key 成功，返回 1；若是當前 key 已經存在，則設置當前 key 失敗，返回 0。

• expire()

expire 設置過時時間，要注意的是 setnx 命令不能設置 key 的超時時間，只能經過 expire() 來對 key 設置。

• 具體步驟

一、setnx(lockKey, 1) 若是返回 0，則說明佔位失敗；若是返回 1，則說明佔位成功

二、expire() 命令對 lockKey 設置超時時間，爲的是避免死鎖問題。

三、執行完業務代碼後，能夠經過 delete 命令刪除 key。

這個方案實際上是能夠解決平常工做中的需求的，但從技術方案的探討上來講，可能還有一些能夠完善的地方。好比，若是在第一步 setnx 執行成功後，在 expire() 命令執行成功前，發生了宕機的現象，那麼就依然會出現死鎖的問題，因此若是要對其進行完善的話，可使用 redis 的 setnx()、get() 和 getset() 方法來實現分佈式鎖。

用 redis 的 setnx()、get()、getset()方法作分佈式鎖

這個方案的背景主要是在 setnx() 和 expire() 的方案上針對可能存在的死鎖問題，作了一些優化。

• getset()

這個命令主要有兩個參數 getset(key，newValue)。該方法是原子的，對 key 設置 newValue 這個值，而且返回 key 原來的舊值。假設 key 原來是不存在的，那麼屢次執行這個命令，會出現下邊的效果：

1. getset(key, "value1") 返回 null 此時 key 的值會被設置爲 value1
2. getset(key, "value2") 返回 value1 此時 key 的值會被設置爲 value2
3. 依次類推！

• 使用步驟

1. setnx(lockKey, 當前時間+過時超時時間)，若是返回 1，則獲取鎖成功；若是返回 0 則沒有獲取到鎖，轉到步驟 2。
2. get(lockKey) 獲取值，值是當前lockKey的過時時間用oldExpireTime表明 ，並將這個 oldExpireTime與當前的系統時間進行比較，若是早於當前系統時間，則認爲這個鎖已經超時，能夠容許別的請求從新獲取，轉向 步驟3，不然等待指定時間後返回步驟2從新開始斷定。
3. 計算 newExpireTime = 當前時間+過時超時時間，而後 getset(lockKey, newExpireTime) 會返回當前 lockKey 以前設置的舊值currentExpireTime。
4. 判斷 currentExpireTime 與 oldExpireTime 是否相等，若是相等，說明當前進程getset 設置鎖成功，獲取到了鎖。若是不相等，說明這個鎖已經被別的進程獲取走了，那麼當前請求能夠根據具體需求邏輯直接返回失敗，或者返回步驟2繼續重試。
5. 在獲取到鎖以後，當前進程能夠開始本身的業務處理，當處理完畢後，比較當前理時間和對鎖設置的超時時間，若是小於鎖設置的超時時間，則直接執行 delete 釋放鎖；若是大於鎖設置的超時時間，鎖可能已由其餘進程得到，這時執行 delete釋放鎖的操做會致使把其餘進程已得到的鎖釋放掉。

下面是用PHP代碼實現的Redis分佈式鎖，關於Redis部分使用的是僞代碼，請根據本身的狀況用Redis鏈接對象替代其中的僞代碼。

/**
 * 獲取Redis分佈式鎖
 *
 * @param $lockKey
 * @return bool
 */
function getRedisDistributedLock(string $lockKey) : bool
{
    $lockTimeout = 2000;// 鎖的超時時間2000毫秒
    $now = intval(microtime(true) * 1000);
    $lockExpireTime = $now + $lockTimeout;
    $lockResult = Redis::setnx($lockKey, $lockExpireTime);

    if ($lockResult) {
        // 當前進程設置鎖成功
        return true;
    } else {
        $oldLockExpireTime = Redis::get($lockKey);
        if ($now > $oldLockExpireTime && $oldLockExpireTime == Redis::getset($lockKey, $lockExpireTime)) {
            return true;
        }
    }

    return false;
}

/**
 * 串行執行程序
 *
 * @param string $lockKey Key for lock
 * @param Closure $closure 得到鎖後進程要執行的閉包
 * @return mixed
 */
function serialProcessing(string $lockKey, Closure $closure)
{
    if (getRedisDistributedLock($lockKey)) {
        $result = $closure();
        $now = intval(microtime(true) * 1000);
        if ($now < Redis::get($lockKey)) {
            Redis::del($lockKey);   
        }
    } else {
        // 延遲200毫秒再執行
        usleep(200 * 1000);
        return serialProcessing($lockKey, $closure);
    }

    return $result;
}

上面serialProcessing方法裏當前進程設置鎖成功，獲取了代碼塊的執行權後就會執行閉包參數$closure裏的代碼塊，經過傳遞閉包給方法，讓咱們能夠在項目任何須要確保程序串行執行的地方使用serialProcessing方法給程序加分佈式鎖解決併發請求的問題。

上面代碼實現用面向過程的方式是爲了能簡單明瞭的描述怎麼設置分佈式鎖，讀者能夠針對本身的狀況執行設計實現代碼。針對於大型系統使用集羣Redis的狀況，設置分佈式鎖的步驟更復雜，有興趣的能夠查看Redlock 算法和redissonredis分佈式鎖組件。