問題背景
現在的應用程序架構中，很多服務都是多副本運行，從而保證服務的穩定性。一個服務實例掛了，其他服務依舊可以接收請求。但是服務的多副本運行隨之也會引來一些分布式問題，比如某個接口的處理邏輯是這樣的：接收到請求后，先查詢 DB 看是否有相關的數據，如果沒有則插入數據，如果有則更新數據。在這種場景下如果相同的 N 個請求并發發到后端服務實例，就會出現重復插入數據的情況：
解決方案
針對上面問題，一般的解決方案是使用分布式鎖來解決。同一個進程內的話用本進程內的鎖即可解決，但是服務多實例部署的話是分布式的，各自進程獨立，這種情況下可以設置一個全局獲取鎖的地方，各個進程都可以通過某種方式獲取這個全局鎖，獲得到鎖后就可以執行相關業務邏輯代碼，沒有拿到鎖則跳過不執行，這個全局鎖就是我們所說的分布式鎖。分布式鎖一般有三種實現方式：1. 數據庫樂觀鎖；2. 基于Redis的分布式鎖；3. 基于ZooKeeper的分布式鎖。
我們這里介紹如何基于 Redis 的分布式鎖來解決分布式并發問題：Redis 充當獲取全局鎖的地方，每個實例在接收到請求的時候首先從 Redis 獲取鎖，獲取到鎖后執行業務邏輯代碼，沒爭搶到鎖則放棄執行。
主要實現原理：
Redis 鎖主要利用 Redis 的 setnx 命令：
加鎖命令：SETNX key value，當鍵不存在時，對鍵進行設置操作并返回成功，否則返回失敗。KEY 是鎖的唯一標識，一般按業務來決定命名。Value 一般用 UUID 標識，確保鎖不被誤解。
解鎖命令：DEL key，通過刪除鍵值對釋放鎖，以便其他線程可以通過 SETNX 命令來獲取鎖。
鎖超時：EXPIRE key timeout, 設置 key 的超時時間，以保證即使鎖沒有被顯式釋放，鎖也可以在一定時間后自動釋放，避免資源被永遠鎖住。
可靠性：
為了確保分布式鎖可用，我們至少要確保鎖的實現同時滿足以下四個條件：
- 互斥性。在任意時刻，保證只有一臺機器的一個線程可以持有鎖；
- 不會發生死鎖。即使有一個客戶端在持有鎖的期間崩潰而沒有主動解鎖，也能保證后續其他客戶端能加鎖；
- 具備非阻塞性。一旦獲取不到鎖就立刻返回加鎖失敗；
- 加鎖和解鎖必須是同一個客戶端，客戶端自己不能把別人加的鎖給解了；
SpringBoot 集成使用 Redis 分布式鎖
寫了一個 RedisLock 工具類，用于業務邏輯執行前加鎖和業務邏輯執行完解鎖操作。這里的加鎖操作可能實現的不是很完善，有加鎖和鎖過期兩個操作原子性問題，如果 SpringBoot 版本是2.x的話是可以用注釋中的代碼在加鎖的時候同時設置鎖過期時間，如果 SpringBoot 版本是2.x以下的話建議使用 Lua 腳本來確保操作的原子性，這里為了簡單就先這樣寫：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
import org.springframework.beans.factory.annotation.Autowired;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.data.redis.core.script.DefaultRedisScript;import org.springframework.stereotype.Component;import java.util.ArrayList;import java.util.List;import java.util.concurrent.TimeUnit;/** * @description: Redis分布式鎖實現工具類 * @author: qianghaohao * @time: 2021/7/19 */@Componentpublic class RedisLock { @Autowired StringRedisTemplate redisTemplate; /** * 獲取鎖 * * @param lockKey 鎖 * @param identity 身份標識（保證鎖不會被其他人釋放） * @param expireTime 鎖的過期時間（單位：秒） * @return */ public boolean lock(String lockKey, String identity, long expireTime) { // 由于我們目前 springboot 版本比較低，1.5.9，因此還不支持下面這種寫法 // return redisTemplate.opsForValue().setIfAbsent(lockKey, identity, expireTime, TimeUnit.SECONDS); if (redisTemplate.opsForValue().setIfAbsent(lockKey, identity)) { redisTemplate.expire(lockKey, expireTime, TimeUnit.SECONDS); return true; } return false; } /** * 釋放鎖 * * @param lockKey 鎖 * @param identity 身份標識（保證鎖不會被其他人釋放） * @return */ public boolean releaseLock(String lockKey, String identity) { String luaScript = "if " + " redis.call('get', KEYS[1]) == ARGV[1] " + "then " + " return redis.call('del', KEYS[1]) " + "else " + " return 0 " + "end"; DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(); redisScript.setResultType(Boolean.class); redisScript.setScriptText(luaScript); List<String> keys = new ArrayList<>(); keys.add(lockKey); Object result = redisTemplate.execute(redisScript, keys, identity); return (boolean) result; }} |
使用示例
這里只貼出關鍵的使用代碼，注意：鎖的 key 根據自己的業務邏輯命名，能唯一標示同一個請求即可。value 這里設置為 UUID，為了確保釋放鎖的時候能正確釋放（只釋放自己加的鎖）。
|
1
2
|
@Autowiredprivate RedisLock redisLock; // redis 分布式鎖 |
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
String redisLockKey = String.format("%s:docker-image:%s", REDIS_LOCK_PREFIX, imageVo.getImageRepository()); String redisLockValue = UUID.randomUUID().toString(); try { if (!redisLock.lock(redisLockKey, redisLockValue, REDIS_LOCK_TIMEOUT)) { logger.info("redisLockKey [" + redisLockKey + "] 已存在，不執行鏡像插入和更新"); result.setMessage("新建鏡像頻繁，稍后重試，鎖占用"); return result; } ... // 執行業務邏輯catch (Execpion e) { ... // 異常處理} finally { // 釋放鎖 if (!redisLock.releaseLock(redisLockKey, redisLockValue)) { logger.error("釋放redis鎖 [" + redisLockKey + "] 失敗); } else { logger.error("釋放redis鎖 [" + redisLockKey + "] 成功"); } } |
參考文檔
https://www.jianshu.com/p/6c2f85e2c586
https://xiaomi-info.github.io/2019/12/17/redis-distributed-lock/
到此這篇關于SpringBoot 使用 Redis 分布式鎖解決并發問題的文章就介紹到這了,更多相關SpringBoot Redis 分布式鎖內容請搜索服務器之家以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持服務器之家！
原文鏈接：https://blog.csdn.net/qianghaohao/article/details/119061642
