造了一個 Redis 分佈鎖的輪子，沒想到還學到這麼多東西！！！

時間 2020-08-05

標籤一個 redis 分佈輪子沒想到學到東西欄目 Redis 简体版

原文原文鏈接

書接上文

上篇文章「MySQL 可重複讀，差點就讓我背上了一個 P0 事故！」發佈以後，收到不少小夥伴們的留言，從中又學習到不少，總結一下。html

上篇文章可能舉得例子有點不恰當，致使有些小夥伴沒看懂爲何餘額會變負。redis

此次咱們舉得實際一點，仍是上篇文章 account 表，假設 id=1，balance=1000，不過此次咱們扣款 1000，兩個事務的時序圖以下:算法

此次使用兩個命令窗口真實執行一把：spring

注意事務 2，③處查詢到 id=1，balance=1000，可是實際上因爲此時事務 1 已經提交，最新結果如②處所示 id=1，balance=900。sql

原本 Java 代碼層會作一層餘額判斷：數據庫

if (balance - amount < 0) {
  throw new XXException("餘額不足，扣減失敗");
}

可是此時因爲 ③ 處使用快照讀，讀到是個舊值，未讀到最新值，致使這層校驗失效，從而代碼繼續往下運行，執行了數據更新。數組

更新語句又採用以下寫法:緩存

UPDATE account set balance=balance-1000 WHERE id =1;

這條更新語句又必須是在這條記錄的最新值的基礎作更新，更新語句執行結束，這條記錄就變成了 id=1，balance=-1000。服務器

以前有朋友疑惑 t12 更新以後，再次進行快照讀，結果會是多少。網絡

上圖執行結果 ④ 能夠看到結果爲 id=1,balance=-1000,能夠看到已經查詢最新的結果記錄。

這行數據最新版本因爲是事務 2 本身更新的，自身事務更新永遠對本身可見。

另外此次問題上本質上由於 Java 層與數據庫層數據不一致致使，有的朋友留言提出，能夠在更新餘額時加一層判斷：

UPDATE account set balance=balance-1000 WHERE id =1 and balance>0;

而後更新完成，Java 層判斷更新有效行數是否大於 0。這種作法確實能規避這個問題。

最後這位朋友留言總結的挺好，粘貼一下：

手擼分佈式鎖

如今切回正文，這篇文章原本是準備寫下 Mysql 查詢左匹配的問題，可是還沒研究出來。那就先寫下最近在鼓搗一個東西，使用 Redis 實現可重入分佈鎖。

看到這裏，有的朋友可能會提出來使用 redisson 不香嗎，爲何還要本身實現？

哎，redisson 真的很香，可是現有項目中沒辦法使用，只好本身手擼一個可重入的分佈式鎖了。

雖然用不了 redisson，可是我能夠研究其源碼，最後實現的可重入分佈鎖參考了 redisson 實現方式。

分佈式鎖

分佈式鎖特性就要在於排他性，同一時間內多個調用方加鎖競爭，只能有一個調用方加鎖成功。

Redis 因爲內部單線程的執行，內部按照請求前後順序執行，沒有併發衝突，因此只會有一個調用方纔會成功獲取鎖。

並且 Redis 基於內存操做，加解鎖速度性能高，另外咱們還可使用集羣部署加強 Redis 可用性。

加鎖

使用 Redis 實現一個簡單的分佈式鎖，很是簡單，能夠直接使用 SETNX 命令。

SETNX 是『SET if Not eXists』，若是不存在，纔會設置，使用方法以下：

不過直接使用 SETNX 有一個缺陷，咱們沒辦法對其設置過時時間，若是加鎖客戶端宕機了，這就致使這把鎖獲取不了了。

有的同窗可能會提出，執行 SETNX 以後，再執行 EXPIRE 命令，主動設置過時時間，僞碼以下：

var result = setnx lock "client"
if(result==1){
    // 有效期 30 s
    expire lock 30
}

不過這樣仍是存在缺陷，加鎖代碼並不能原子執行，若是調用加鎖語句，還沒來得及設置過時時間，應用就宕機了，仍是會存在鎖過時不了的問題。

不過這個問題在 Redis 2.6.12 版本就能夠被完美解決。這個版本加強了 SET 命令，能夠經過帶上 NX,EX 命令原子執行加鎖操做，解決上述問題。參數含義以下：

EX second ：設置鍵的過時時間，單位爲秒
NX 當鍵不存在時，進行設置操做，等同與 SETNX 操做

使用 SET 命令實現分佈式鎖只須要一行代碼：

SET lock_name anystring NX EX lock_time

解鎖

解鎖相比加鎖過程，就顯得很是簡單，只要調用 DEL 命令刪除鎖便可:

DEL lock_name

不過這種方式卻存在一個缺陷，可能會發生錯解鎖問題。

假設應用 1 加鎖成功，鎖超時時間爲 30s。因爲應用 1 業務邏輯執行時間過長，30 s 以後，鎖過時自動釋放。

這時應用 2 接着加鎖，加鎖成功，執行業務邏輯。這個期間，應用 1 終於執行結束，使用 DEL 成功釋放鎖。

這樣就致使了應用 1 錯誤釋放應用 2 的鎖，另外鎖被釋放以後，其餘應用可能再次加鎖成功，這就可能致使業務重複執行。

爲了使鎖不被錯誤釋放，咱們須要在加鎖時設置隨機字符串，好比 UUID。

SET lock_name uuid NX EX lock_time

釋放鎖時，須要提早獲取當前鎖存儲的值，而後與加鎖時的 uuid 作比較，僞代碼以下：

var value= get lock_name
if value == uuid
 // 釋放鎖成功
else
 // 釋放鎖失敗

上述代碼咱們不能經過 Java 代碼運行,由於沒法保證上述代碼原子化執行。

幸虧 Redis 2.6.0 增長執行 Lua 腳本的功能，lua 代碼能夠運行在 Redis 服務器的上下文中，而且整個操做將會被當成一個總體執行，中間不會被其餘命令插入。

這就保證了腳本將會以原子性的方式執行，當某個腳本正在運行的時候，不會有其餘腳本或 Redis 命令被執行。在其餘的別的客戶端看來，執行腳本的效果，要麼是不可見的，要麼就是已完成的。

EVAL 與 EVALSHA

EVAL

Redis 可使用 EVAL 執行 LUA 腳本，而咱們能夠在 LUA 腳本中執行判斷求值邏輯。EVAL 執行方式以下：

EVAL script numkeys key [key ...] arg [arg ...]

numkeys 參數用於鍵名參數，即後面 key 數組的個數。

key [key ...] 表明須要在腳本中用到的全部 Redis key，在 Lua 腳本使用使用數組的方式訪問 key，相似以下 KEYS[1] ， KEYS[2]。注意 Lua 數組起始位置與 Java 不一樣，Lua 數組是從 1 開始。

命令最後，是一些附加參數，能夠用來當作 Redis Key 值存儲的 Value 值，使用方式如 KEYS 變量同樣，相似以下：ARGV[1] 、 ARGV[2] 。

用一個簡單例子運行一下 EVAL 命令：

eval "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2],ARGV[3]}" 2 key1 key2 first second third

運行效果以下：

能夠看到 KEYS 與 ARGVS內部數組能夠不一致。

在 Lua 腳本可使用下面兩個函數執行 Redis 命令：

redis.call()
redis.pcall()

兩個函數做用法與做用徹底一致，只不過對於錯誤的處理方式不一致，感興趣的小夥伴能夠具體點擊如下連接，查看錯誤處理一章。

http://doc.redisfans.com/script/eval.html

下面咱們統一在 Lua 腳本中使用 redis.call()，執行如下命令：

eval "return redis.call('set',KEYS[1],ARGV[1])" 1 foo 樓下小黑哥

運行效果以下：

EVALSHA

EVAL 命令每次執行時都須要發送 Lua 腳本，可是 Redis 並不會每次都會從新編譯腳本。

當 Redis 第一次收到 Lua 腳本時，首先將會對 Lua 腳本進行 sha1 獲取簽名值，而後內部將會對其緩存起來。後續執行時，直接經過 sha1 計算事後簽名值查找已經編譯過的腳本，加快執行速度。

雖然 Redis 內部已經優化執行的速度，可是每次都須要發送腳本，仍是有網絡傳輸的成本，若是腳本很大，這其中花在網絡傳輸的時間就會相應的增長。

因此 Redis 又實現了 EVALSHA 命令，原理與 EVAL 一致。只不過 EVALSHA 只須要傳入腳本通過 sha1計算事後的簽名值便可，這樣大大的減小了傳輸的字節大小，減小了網絡耗時。

EVALSHA命令以下：

evalsha c686f316aaf1eb01d5a4de1b0b63cd233010e63d 1 foo 樓下小黑哥

運行效果以下：

「

SCRIPT FLUSH 命令用來清除全部 Lua 腳本緩存。

能夠看到，若是以前未執行過 EVAL命令，直接執行 EVALSHA 將會報錯。

優化執行 EVAL

咱們能夠結合使用 EVAL 與 EVALSHA，優化程序。下面就不寫僞碼了，以 Jedis 爲例，優化代碼以下：

//鏈接本地的 Redis 服務
Jedis jedis = new Jedis("localhost", 6379);
jedis.auth("1234qwer");

System.out.println("服務正在運行: " + jedis.ping());

String lua_script = "return redis.call('set',KEYS[1],ARGV[1])";
String lua_sha1 = DigestUtils.sha1DigestAsHex(lua_script);

try {
    Object evalsha = jedis.evalsha(lua_sha1, Lists.newArrayList("foo"), Lists.newArrayList("樓下小黑哥"));
} catch (Exception e) {
    Throwable current = e;
    while (current != null) {
        String exMessage = current.getMessage();
        // 包含 NOSCRIPT，表明該 lua 腳本從未被執行，須要先執行 eval 命令
        if (exMessage != null && exMessage.contains("NOSCRIPT")) {
            Object eval = jedis.eval(lua_script, Lists.newArrayList("foo"), Lists.newArrayList("樓下小黑哥"));
            break;
        }

    }
}
String foo = jedis.get("foo");
System.out.println(foo);

上面的代碼看起來仍是很複雜吧，不過這是使用原生 jedis 的狀況下。若是咱們使用 Spring Boot 的話，那就沒這麼麻煩了。Spring 組件執行的 Eval 方法內部就包含上述代碼的邏輯。

不過須要注意的是，若是 Spring-Boot 使用 Jedis 做爲鏈接客戶端,而且使用Redis Cluster 集羣模式，須要使用 2.1.9 以上版本的spring-boot-starter-data-redis,否則執行過程當中將會拋出：

org.springframework.dao.InvalidDataAccessApiUsageException: EvalSha is not supported in cluster environment.

詳細狀況能夠參考這個修復的 IssueAdd support for scripting commands with Jedis Cluster

優化分佈式鎖

講完 Redis 執行 LUA 腳本的相關命令，咱們來看下如何優化上面的分佈式鎖，使其沒法釋放其餘應用加的鎖。

「

如下代碼基於 spring-boot 2.2.7.RELEASE 版本，Redis 底層鏈接使用 Jedis。

加鎖的 Redis 命令以下：

SET lock_name uuid NX EX lock_time

加鎖代碼以下：

/**
 * 非阻塞式加鎖，若鎖存在，直接返回
 *
 * @param lockName  鎖名稱
 * @param request   惟一標識，防止其餘應用/線程解鎖，可使用 UUID 生成
 * @param leaseTime 超時時間
 * @param unit      時間單位
 * @return
 */
public Boolean tryLock(String lockName, String request, long leaseTime, TimeUnit unit) {
    // 注意該方法是在 spring-boot-starter-data-redis 2.1 版本新增長的，如果以前版本 能夠執行下面的方法
    return stringRedisTemplate.opsForValue().setIfAbsent(lockName, request, leaseTime, unit);
}

因爲setIfAbsent方法是在 spring-boot-starter-data-redis 2.1 版本新增長，以前版本沒法設置超時時間。若是使用以前的版本的，須要以下方法：

/**
 * 適用於 spring-boot-starter-data-redis 2.1 以前的版本
 *
 * @param lockName
 * @param request
 * @param leaseTime
 * @param unit
 * @return
 */
public Boolean doOldTryLock(String lockName, String request, long leaseTime, TimeUnit unit) {
    Boolean result = stringRedisTemplate.execute((RedisCallback<Boolean>) connection -> {
        RedisSerializer valueSerializer = stringRedisTemplate.getValueSerializer();
        RedisSerializer keySerializer = stringRedisTemplate.getKeySerializer();

        Boolean innerResult = connection.set(keySerializer.serialize(lockName),
                valueSerializer.serialize(request),
                Expiration.from(leaseTime, unit),
                RedisStringCommands.SetOption.SET_IF_ABSENT
        );
        return innerResult;
    });
    return result;
}

解鎖須要使用 Lua 腳本:

-- 解鎖代碼
-- 首先判斷傳入的惟一標識是否與現有標識一致
-- 若是一致，釋放這個鎖，不然直接返回
if redis.call('get', KEYS[1]) == ARGV[1] then
   return redis.call('del', KEYS[1])
else
   return 0
end

這段腳本將會判斷傳入的惟一標識是否與 Redis 存儲的標示一致，若是一直，釋放該鎖，不然馬上返回。

釋放鎖的方法以下：

/**
 * 解鎖
 * 若是傳入應用標識與以前加鎖一致，解鎖成功
 * 不然直接返回
 * @param lockName 鎖
 * @param request 惟一標識
 * @return
 */
public Boolean unlock(String lockName, String request) {
    DefaultRedisScript<Boolean> unlockScript = new DefaultRedisScript<>();
    unlockScript.setLocation(new ClassPathResource("simple_unlock.lua"));
    unlockScript.setResultType(Boolean.class);
    return stringRedisTemplate.execute(unlockScript, Lists.newArrayList(lockName), request);
}