鎖在我們的日常開發可謂用得比較多。通常用來解決資源並發的問題。特別是多機集群情況下，資源爭搶的問題。但是，很多新手在鎖的處理上常常會犯一些問題。今天我們來深入理解鎖。
一、Redis 鎖錯誤使用之一
我曾經見過有的項目把查詢結果存儲到 Redis 當中時的偽代碼如下：
$redis = new \Redis('127.0.0.1', 6379);
$cacheKey = 'query_cache';
$result = $redis->get($cacheKey);
if ($result) { // 緩存有效則直接返回。
return $result;
} else { // 緩存失效則重新獲取並存儲到 Redis。
$mysqlResult = [];
$redis->set($cacheKey, json_encode($mysqlResult), 3600);
return $mysqlResult;
}
初看代碼並不會發現問題所在。通常情況下，當服務器資源壓力非常小的時候，這段代碼不會有任何問題。並且，真的可以提升服務器吞吐性能。
假如，這個位置的代碼出現了單點壓力呢？比如，這個功能是統計結果，查詢數據庫需要花 5s。而且，由於該功能比較常用，單位時間內達到了 1000 次/秒。
這時就會出現並發穿透問題。
1000 個請求同時到達這個程序位置，都去讀取緩存是否存在。假如此時緩存不存在。這 1000 個請求都會得到不存在的結果。並且都會執行到去數據庫取緩存結果的步驟。同時也會把結果重寫到 Redis。
那就導致了這一瞬間單點壓力導致穿透到數據庫，造成數據庫壓力瞬間到達峰值。如果我們的數據庫的性能處理不了這么大的壓力，就會導致數據庫服務器 CPU 直接爆滿。響應給前端的數據就會陷入停頓狀態。
所以，這段代碼是不正確的鎖使用。
二、Redis 鎖錯誤使用之二
在第一點中，我們發現了問題。於是，就有人想着去優化它。於是就有了下面的代碼：
$redis = new \Redis('127.0.0.1', 6379);
$lockKey = 'query_cache_lock'; // 鎖專用的 KEY。
$cacheKey = 'query_cache'; // 存儲查詢結果的 KEY。
$result = $redis->get($cacheKey);
if ($result) { // 緩存有效則直接返回。
return $result;
} else { // 緩存失效則重新獲取並存儲到 Redis。
if ($redis->setNx($lockKey) === false) {
throw new \Exception("服務器火爆，請稍候重試");
} else {
$mysqlResult = [];
$redis->set($cacheKey, json_encode($mysqlResult), 3600);
$redis->delete($lockKey); // 鎖用完了要解鎖。刪掉就是解鎖。
return $mysqlResult;
}
}
這段代碼就完全避免了第一點中的並發穿透的問題。但是，相對第一點，代碼也多增加了幾行。不過性能依然強勁。
即使如此，這段代碼依然存在三個問題：
1）並發越大，第一個取到鎖的請求能正常響應，后續的請求就會得到一個“服務器火爆，請稍候重試”的異常提示。
2）沒辦法對后續請求取鎖失效加一個等待時間。
3）如果代碼執行到 $redis->delete($lockKey) 之前程序異常了。那么鎖就不能正常釋放。后續的鎖也無法正常取到鎖了。
針對第 1） 點，這個是用戶體驗極差的。
針對第 2） 點，它是解決第一點的方案。
針對第 3） 點，它是我們必須解決的問題。否則，我們的分布式鎖將無法正常使用。
三、正確的分布式鎖
正常的分布式鎖要滿足以下幾點要求：
1）能解決並發時資源爭搶。這是最核心的需求。
2）鎖能正常添加與釋放。不能出現死鎖。
3）鎖能實現等待，否則不能最大保證用戶的體驗。
針對以上三點，得出 Redis 分布式鎖示例
class RedisMutexLock
{
/**
* 緩存 Redis 連接。
*
* @return void
*/
public static function getRedis()
{
// 這行代碼請根據自己項目替換為自己的獲取 Redis 連接。
return YCache::getRedisClient();
}
/**
* 獲得鎖,如果鎖被占用,阻塞,直到獲得鎖或者超時。
* -- 1、如果 $timeout 參數為 0,則立即返回鎖。
* -- 2、建議 timeout 設置為 0,避免 redis 因為阻塞導致性能下降。請根據實際需求進行設置。
*
* @param string $key 緩存KEY。
* @param int $timeout 取鎖超時時間。單位(秒)。等於0,如果當前鎖被占用,則立即返回失敗。如果大於0,則反復嘗試獲取鎖直到達到該超時時間。
* @param int $lockSecond 鎖定時間。單位(秒)。
* @param int $sleep 取鎖間隔時間。單位(微秒)。當鎖為占用狀態時。每隔多久嘗試去取鎖。默認 0.1 秒一次取鎖。
* @return bool 成功:true、失敗:false
*/
public static function lock($key, $timeout = 0, $lockSecond = 20, $sleep = 100000)
{
if (strlen($key) === 0) {
// 項目拋異常方法
YCore::exception(500, '緩存KEY沒有設置');
}
$start = self::getMicroTime();
$redis = self::getRedis();
do {
// [1] 鎖的 KEY 不存在時設置其值並把過期時間設置為指定的時間。鎖的值並不重要。重要的是利用 Redis 的特性。
$acquired = $redis->set("Lock:{$key}", 1, ['NX', 'EX' => $lockSecond]);
if ($acquired) {
break;
}
if ($timeout === 0) {
break;
}
usleep($sleep);
} while (!is_numeric($timeout) || (self::getMicroTime()) < ($start + ($timeout * 1000000)));
return $acquired ? true : false;
}
/**
* 釋放鎖
*
* @param mixed $key 被加鎖的KEY。
* @return void
*/
public static function release($key)
{
if (strlen($key) === 0) {
// 項目拋異常方法
YCore::exception(500, '緩存KEY沒有設置');
}
$redis = self::getRedis();
$redis->del("Lock:{$key}");
}
/**
* 獲取當前微秒。
*
* @return bigint
*/
protected static function getMicroTime()
{
return bcmul(microtime(true), 1000000);
}
}
以上是在項目中一些的用到的之處，大家可以更換為自己項目
有需要交流的小伙伴可以點擊這里加本人QQ：luke
