用分散式鎖解決併發問題
在系統中，當存在多個程序和執行緒可以改變某個共享資料時，就容易出現併發問題導致共享資料的不一致性。即多個程序同時獲取到了對資料的操作許可權並對資料進行了更新，很典型的場景就是線上銷售系統在售賣熱銷商品時遇到多個併發請求在同一時間提交訂單的情況則極有可能造成商品超賣的現象。只要訪問流量不錯的系統都有可能遭遇併發請求造成資料庫中資料重複寫入的情況。
針對程式塊被多個程序併發執行問題的解決方案是確保同一個時刻同一個程式塊只能有一個程序可執行，其他程序等待當前程序執行完成才能獲取程式塊的執行權對資料進行更新，以此類推將併發執行變為序列順序執行。為了讓獲取執行權的程序不被其他干擾，就需要設定一個所有程序都能讀取到的標記，當標記不存在時可以設定該標記，其餘後續程序發現已經有標記了則等待擁有標記的程序結束執行程式塊取消標記後再去嘗試設定標記。這個標記可以理解為鎖，設定標記的過程就是我們通常說的加鎖。
用redis 的 setnx、expire 方法做分散式鎖
- setnx()
setnx 的含義就是 SET if Not Exists，其主要有兩個引數 setnx(key, value)。該方法是原子的，如果 key 不存在，則設定當前 key 成功，返回 1；如果當前 key 已經存在，則設定當前 key 失敗，返回 0。
- expire()
expire 設定過期時間，要注意的是 setnx 命令不能設定 key 的超時時間，只能通過 expire() 來對 key 設定。
- 具體步驟
1、setnx(lockKey, 1) 如果返回 0，則說明佔位失敗；如果返回 1，則說明佔位成功
2、expire() 命令對 lockKey 設定超時時間，為的是避免死鎖問題。
3、執行完業務程式碼後，可以通過 delete 命令刪除 key。
這個方案其實是可以解決日常工作中的需求的，但從技術方案的探討上來說，可能還有一些可以完善的地方。比如，如果在第一步 setnx 執行成功後，在 expire() 命令執行成功前，發生了宕機的現象，那麼就依然會出現死鎖的問題，所以如果要對其進行完善的話，可以使用 redis 的 setnx()、get() 和 getset() 方法來實現分散式鎖。
用 redis 的 setnx()、get()、getset()方法做分散式鎖
這個方案的背景主要是在 setnx() 和 expire() 的方案上針對可能存在的死鎖問題，做了一些優化。
- getset()
這個命令主要有兩個引數 getset(key，newValue)。該方法是原子的，對 key 設定 newValue 這個值，並且返回 key 原來的舊值。假設 key 原來是不存在的，那麼多次執行這個命令，會出現下邊的效果：
- getset(key, "value1") 返回 null 此時 key 的值會被設定為 value1
- getset(key, "value2") 返回 value1 此時 key 的值會被設定為 value2
- 依次類推！
- 使用步驟
- setnx(lockKey, 當前時間+過期超時時間)，如果返回 1，則獲取鎖成功；如果返回 0 則沒有獲取到鎖，轉到步驟 2。
- get(lockKey) 獲取值，值是當前lockKey的過期時間用oldExpireTime代表 ，並將這個 oldExpireTime與當前的系統時間進行比較，如果早於當前系統時間，則認為這個鎖已經超時，可以允許別的請求重新獲取，轉向 步驟3，否則等待指定時間後返回步驟2重新開始判定。
- 計算 newExpireTime = 當前時間+過期超時時間，然後 getset(lockKey, newExpireTime) 會返回當前 lockKey 之前設定的舊值currentExpireTime。
- 判斷 currentExpireTime 與 oldExpireTime 是否相等，如果相等，說明當前程序getset 設定鎖成功，獲取到了鎖。如果不相等，說明這個鎖已經被別的程序獲取走了，那麼當前請求可以根據具體需求邏輯直接返回失敗，或者返回步驟2繼續重試。
- 在獲取到鎖之後，當前程序可以開始自己的業務處理，當處理完畢後，比較當前理時間和對鎖設定的超時時間，如果小於鎖設定的超時時間，則直接執行 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
演算法和redisson
redis分散式鎖元件。