一、前言

????redis 現在已經成為系統緩存的必備組件，針對緩存讀取更新操作，通常我們希望當緩存過期之后能夠只有一個請求去更新緩存，其它請求依然使用舊的數據。這就需要用到鎖，因為應用服務多數以集群方式部署，因此這里的鎖就必需要是分布式鎖才能符合需求。

?

二、spring-boot 引入 redis

在 pom 文件中加入如下依賴，spring-boot 的自動注冊功能會幫我們準備好，我們直接使用 StringRedisTemplate 就可以了。

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId> </dependency>

?

三、redis 分布式鎖實現

/*** @author koma <komazhang@foxmail.com>* @date 2018-09-19 11:24*/ @Slf4j @Service public class CacheService {private static final Long RELEASE_SUCCESS = 1L;private static final String LOCK_SUCCESS = "OK";private static final String SET_IF_NOT_EXIST = "NX";private static final String SET_WITH_EXPIRE_TIME = "EX";private static final String RELEASE_LOCK_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";@Autowiredprivate StringRedisTemplate redisTemplate;/*** 該加鎖方法僅針對單實例 Redis 可實現分布式加鎖* 對于 Redis 集群則無法使用** 支持重復，線程安全** @param lockKey 加鎖鍵* @param clientId 加鎖客戶端唯一標識(采用UUID)* @param seconds 鎖過期時間* @return*/public Boolean tryLock(String lockKey, String clientId, long seconds) {redisTemplate.opsForValue().set();return redisTemplate.execute((RedisCallback<Boolean>) redisConnection -> {Jedis jedis = (Jedis) redisConnection.getNativeConnection();String result = jedis.set(lockKey, clientId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, seconds);if (LOCK_SUCCESS.equals(result)) {return Boolean.TRUE;}return Boolean.FALSE;});}/*** 與 tryLock 相對應，用作釋放鎖** @param lockKey* @param clientId* @return*/public Boolean releaseLock(String lockKey, String clientId) {return redisTemplate.execute((RedisCallback<Boolean>) redisConnection -> {Jedis jedis = (Jedis) redisConnection.getNativeConnection();Object result = jedis.eval(RELEASE_LOCK_SCRIPT, Collections.singletonList(lockKey),Collections.singletonList(clientId));if (RELEASE_SUCCESS.equals(result)) {return Boolean.TRUE;}return Boolean.FALSE;});} }

?

? ? 上述代碼實現，僅對 redis 單實例架構有效，當面對 redis 集群時就無效了。但是一般情況下，我們的 redis 架構多數會做成“主備”模式，然后再通過 redis 哨兵實現主從切換，這種模式下我們的應用服務器直接面向主機，也可看成是單實例，因此上述代碼實現也有效。但是當在主機宕機，從機被升級為主機的一瞬間的時候，如果恰好在這一刻，由于 redis 主從復制的異步性，導致從機中數據沒有即時同步，那么上述代碼依然會無效，導致同一資源有可能會產生兩把鎖，違背了分布式鎖的原則。

?

redis 單實例架構示意圖

?

????為什么上面的代碼可以實現分布式鎖，根本原因在于 redis 對 set 命令中的 NX 選項和對 lua 腳本的執行都是原子的，因此當多個客戶端去爭搶執行上鎖或解鎖代碼時，最終只會有一個客戶端執行成功。同時 set 命令還可以指定key的有效期，這樣即使當前客戶端奔潰，過一段時間鎖也會被 redis 自動釋放，這就給了其它客戶端獲取鎖的機會。

?

????上述代碼不能使用 spring-boot 提供的?redisTemplate.opsForValue().set()?命令是因為 spring-boot 對 jedis 的封裝中沒有返回 set 命令的返回值，這就導致上層沒有辦法判斷 set 執行的結果，因此需要通過 execute 方法調用 RedisCallback 去拿到底層的 Jedis 對象，來直接調用 set 命令。這個問題主要是在 spring-data-redis 的封裝上，了解即可。?

?

?四、分布式鎖的原則

• 獨享: 即互斥屬性，在同一時刻，一個資源只能有一把鎖被一個客戶端持有

• 無死鎖: 當持有鎖的客戶端奔潰后，鎖仍然可以被其它客戶端獲取

• 容錯性: 當部分節點失活之后，其余節點客戶端依然可以獲取和釋放鎖

• 統一性: 即釋放鎖的客戶端只能由獲取鎖的客戶端釋放

?

五、一類常見錯誤實現和推薦使用方式

if (redisTemplate.opsForValue().setIfAbsent(lockKey, clientId)) {//這里存在宕機風險，導致設置有效期失敗redisTemplate.expire(lockKey, seconds, TimeUnit.SECONDS); }

????這是一種典型的錯誤實現，在早期的 redis 分布式鎖實踐中我們經常可以看到類似的實現，其中 spring-boot 中的?setIfAbsent 方法在底層調用的是 redis 的 setNx 命令，該命令和 set 命令的 NX 選項具有同樣的功能，但是 setNx 命令不能夠設置 key 的有效期，這也是為什么我們會在獲取到鎖之后馬上去設置鎖的有效期，但是恰好這里卻隱藏著風險，因為這一整個操作并非是原子的。

?

if (clientId.equals(redisTemplate.opsForValue().get(lockKey))) {redisTemplate.delete(lockKey); }

????對于解鎖代碼，也存在同樣的風險，因為在執行 delete 的時候，lockKey 現在可能已經被另外一個客戶端持有了，那么這里直接刪除就是刪除了其它客戶端的鎖，導致的最終結果就是真正應該持有鎖的客戶端在沒有完全執行完之后，鎖又被另外的客戶端持有了，這樣一個資源就產生了兩把鎖，同樣違背了分布式鎖的原則。

?

????推薦的使用方式是，當 redis 的架構如上圖所示一樣是單實例模式時，如果存在主備且可以忍受小概率的鎖出錯，那么就可以直接使用上述代碼，當然最嚴謹的方式還是使用官方的?Redlock?算法實現。其中 Java 包推薦使用?redisson。

?

六、參考資料

• Redis分布式鎖

原文：https://blog.51cto.com/13975879/2177400

總結

以上是生活随笔為你收集整理的spring-boot 中实现标准 redis 分布式锁的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。