分散式鎖是在分散式環境下（多個JVM程序）控制多個客戶端對某一資源的同步訪問的一種實現，與之相對應的是執行緒鎖，執行緒鎖控制的是同一個JVM程序內多個執行緒之間的同步。分散式鎖的一般實現方法是在應用伺服器之外通過一個共享的儲存伺服器儲存鎖資源，同一時刻只有一個客戶端能佔有鎖資源來完成。通常有基於Zookeeper，Redis，或資料庫三種實現形式。本文介紹基於Redis的實現方案。 ## 要求 基於Redis實現分散式鎖需要滿足如下幾點要求： 1. 在分散式叢集中，被分散式鎖控制的方法或程式碼段同一時刻只能被一個客戶端上面的一個執行緒執行，也就是互斥 2. 鎖資訊需要設定過期時間，避免一個執行緒長期佔有（比如在做解鎖操作前異常退出）而導致死鎖 3. 加鎖與解鎖必須一致，誰加的鎖，就由誰來解（或過期超時），一個客戶端不能解開另一個客戶端加的鎖 4. 加鎖與解鎖的過程必須保證原子性 ## 實現 ### 1. 加鎖實現 基於Redis的分散式鎖加鎖操作一般使用 `SETNX` 命令，其含義是“將 key 的值設為 value ，當且僅當 key 不存在。若給定的 key 已經存在，則 SETNX 不做任何動作”。 在 Spring Boot 中，可以使用 StringRedisTemplate 來實現，如下，一行程式碼即可實現加鎖過程。（下列程式碼給出兩種呼叫形式——立即返回加鎖結果與給定超時時間獲取加鎖結果） ```java /** * 嘗試獲取鎖（立即返回） * @param key 鎖的redis key * @param value 鎖的value * @param expire 過期時間/秒 * @return 是否獲取成功 */ public boolean lock(String key, String value, long expire) { return stringRedisTemplate.opsForValue().setIfAbsent(key, value, expire, TimeUnit.SECONDS); } /** * 嘗試獲取鎖，並至多等待timeout時長 * * @param key 鎖的redis key * @param value 鎖的value * @param expire 過期時間/秒 * @param timeout 超時時長 * @param unit 時間單位 * @return 是否獲取成功 */ public boolean lock(String key, String value, long expire, long timeout, TimeUnit unit) { long waitMillis = unit.toMillis(timeout); long waitAlready = 0; while (!stringRedisTemplate.opsForValue().setIfAbsent(key, value, expire, TimeUnit.SECONDS) && waitAlready < waitMillis) { try { Thread.sleep(waitMillisPer); } catch (InterruptedException e) { log.error("Interrupted when trying to get a lock. key: {}", key, e); } waitAlready += waitMillisPer; } if (waitAlready < waitMillis) { return true; } log.warn("<====== lock {} failed after waiting for {} ms", key, waitAlready); return false; } ``` 上述實現如何滿足前面提到的幾點要求： 1. 客戶端互斥： 可以將expire過期時間設定為大於同步程式碼的執行時間，比如同步程式碼塊執行時間為1s，則可將expire設定為3s或5s。避免同步程式碼執行過程中expire時間到，其它客戶端又可以獲取鎖執行同步程式碼塊。 2. 通過設定過期時間expire來避免某個客戶端長期佔有鎖。 3. 通過value來控制誰加的鎖，由誰解的邏輯，比如可以使用requestId作為value，requestId唯一標記一次請求。 4. setIfAbsent方法 底層通過呼叫 Redis 的 `SETNX` 命令，操作具備原子性。 **錯誤示例：** 網上有如下實現， ```java public boolean lock(String key, String value, long expire) { boolean result = stringRedisTemplate.opsForValue().setIfAbsent(key, value); if(result) { stringRedisTemplate.expire(key, expire, TimeUnit.SECONDS); } return result; } ``` 該實現的問題是如果在result為true，但還沒成功設定expire時，程式異常退出了，將導致該鎖一直被佔用而導致死鎖，不滿足第二點要求。 ### 2. 解鎖實現 解鎖也需要滿足前面所述的四個要求，實現程式碼如下： ```java private static final String RELEASE_LOCK_LUA_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end"; private static final Long RELEASE_LOCK_SUCCESS_RESULT = 1L; /** * 釋放鎖 * @param key 鎖的redis key * @param value 鎖的value */ public boolean unLock(String key, String value) { DefaultRed