redis和redission分布式鎖原理及區(qū)別說明
redis和redission分布式鎖原理及區(qū)別
我最近做租車項目，在處理分布式時用到分布式鎖，我發(fā)現(xiàn)很多同事都在網(wǎng)上找分布式鎖的資料，但是看的資料都不是很全，所以在這里我談談自己的分布式鎖理解。
結合我的其中某一業(yè)務需求：多個用戶在同一個區(qū)域內(nèi)發(fā)現(xiàn)只有一輛可租的車，最終結果肯定只有一位用戶租車成功，這就產(chǎn)生了多線程(多個用戶)搶同一資源的問題。
1、有的同伴想到了synchronized關鍵字鎖
暫且拋開性能問題，項目為了高可用，都會做集群部署，那么synchronized就失去了加鎖的意義，這里多嘴解釋一下：
2、有的小伙伴可能想到了樂觀鎖
沒錯?。酚^鎖可以解決的我的問題，但是在高并發(fā)的場景，頻繁的操作數(shù)據(jù)庫，數(shù)據(jù)庫的資源是很珍貴的，并且還存在性能的問題。但是我這里簡單說下樂觀鎖的使用：
- 我們在車的表中添加一個字段：version（int類型）（建議使用這個名稱，這樣別人看到就會直覺這是樂觀鎖字段，也可以使用別的名稱）
- 查詢出該車的數(shù)據(jù)，數(shù)據(jù)中就有version字段，假如version=1
select * from u_car where car_id = 10;
- 修改該車的狀態(tài)為鎖定
update u_car set status = 2，version = version +1 where car_id = 10 and version = 1
在修改的時候?qū)ersion作為參數(shù)，如果其他用戶鎖車，那么version已經(jīng)發(fā)生變化（version = version +1），所以version = 1不成立，修改失敗
樂觀鎖不是本次的終點，但還是簡單說下；
3、使用redis的分布式鎖
public boolean lock(String key, V v, int expireTime){
//獲取鎖
//在redis早期版本中，設置key和key的存活時間是分開的，設置key成功，但是設置存活時間時服務宕機，那么你的key就永遠不會過期，有BUG
//后來redis將加鎖和設置時間用同一個命令
//這里是重點，redis.setNx(key，value,time)方法是原子性的，設置key成功說明鎖車成功，如果失敗說明該車被別人租了
boolean b = false;
try {
b = redis.setNx(key, v, expireTime);
} catch (Exception e) {
log.error(e.getMessage(), e);
}
return b;
}
public boolean unlock(String key){
return redis.delete(key);
}
}
但是這樣寫還是存在BUG的，我的key設置了加鎖時間為5秒，但是我的業(yè)務邏輯5秒還沒有執(zhí)行完成，key過期了，那么其他用戶執(zhí)行redis.setNx(key, v, expireTime)時就成功了，將該車鎖定，又產(chǎn)生了搶資源；我們想一下，如果我能夠在業(yè)務邏輯沒有執(zhí)行完的時候，讓鎖過期后能夠延長鎖的時間，是不是就解決了上面的BUG；
實現(xiàn)這個鎖的延長，非要自己動手的話就得另啟一個線程來監(jiān)聽我們的業(yè)務線程，每隔1秒監(jiān)測當前業(yè)務線程是否執(zhí)行完成，如果沒有就獲取key的存活時間，時間小于一個閾值時，就自動給key設置N秒；當然，我們可以不用自己動手，redission已經(jīng)幫我們實現(xiàn)key的時間時間過期問題；
4、使用redission的分布式鎖
//引入依賴
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.10.6</version>
</dependency>
redisson支持單點、集群等模式，這里選擇單點的。
- application.yml配置好redis的連接：
spring:
redis:
host: 127.0.0.1
port: 6379
password:
- 配置redisson的客戶端bean
@Configuration
public class RedisConfig {
@Value("${spring.redis.host}")
private String host;
@Bean(name = {"redisTemplate", "stringRedisTemplate"})
public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory factory) {
StringRedisTemplate redisTemplate = new StringRedisTemplate();
redisTemplate.setConnectionFactory(factory);
return redisTemplate;
}
@Bean
public Redisson redisson() {
Config config = new Config();
config.useSingleServer().setAddress("redis://" + host + ":6379");
return (Redisson) Redisson.create(config);
}
}
- 加鎖使用
private Logger log = LoggerFactory.getLogger(getClass());
@Resource
private Redisson redisson;
//加鎖
public Boolean lock(String key,long waitTime,long leaseTime){
Boolean b = false;
try {
RLock rLock = redisson.getLock(key);
//說下參數(shù) waitTime：鎖的存活時間 leaseTime：鎖的延長時間 后面的參數(shù)是單位
b = rLock.tryLock(waitTime,leaseTime,TimeUnit.SECONDS);
} catch (Exception e) {
log.error(e.getMessage(), e);
}
}
return b;
}
//釋放鎖
public void unlock(String key){
try {
RLock rLock = redisson.getLock(key);
if(null!=lock){
lock.unlock();
lock.forceUnlock();
fileLog.info("unlock succesed");
}
} catch (Exception e) {
fileLog.error(e.getMessage(), e);
}
}
- 帶大家看下tryLock方法的實現(xiàn)源碼：
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
long time = unit.toMillis(waitTime);
long current = System.currentTimeMillis();
long threadId = Thread.currentThread().getId();
//嘗試獲取鎖，如果沒取到鎖，則獲取鎖的剩余超時時間
Long ttl = tryAcquire(leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
return true;
}
//如果waitTime已經(jīng)超時了，就返回false
time -= System.currentTimeMillis() - current;
if (time <= 0) {
acquireFailed(threadId);
return false;
}
current = System.currentTimeMillis();
RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {
if (!subscribeFuture.cancel(false)) {
subscribeFuture.onComplete((res, e) -> {
if (e == null) {
unsubscribe(subscribeFuture, threadId);
}
});
}
acquireFailed(threadId);
return false;
}
try {
time -= System.currentTimeMillis() - current;
if (time <= 0) {
acquireFailed(threadId);
return false;
}
//進入死循環(huán)，反復去調(diào)用tryAcquire嘗試獲取鎖，ttl為null時就是別的線程已經(jīng)unlock了
while (true) {
long currentTime = System.currentTimeMillis();
ttl = tryAcquire(leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
return true;
}
time -= System.currentTimeMillis() - currentTime;
if (time <= 0) {
acquireFailed(threadId);
return false;
}
// waiting for message
currentTime = System.currentTimeMillis();
if (ttl >= 0 && ttl < time) {
getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} else {
getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
}
time -= System.currentTimeMillis() - currentTime;
if (time <= 0) {
acquireFailed(threadId);
return false;
}
}
} finally {
unsubscribe(subscribeFuture, threadId);
}
// return get(tryLockAsync(waitTime, leaseTime, unit));
}
可以看到，其中主要的邏輯就是嘗試加鎖，成功了就返回true，失敗了就進入死循環(huán)反復去嘗試加鎖。中途還有一些超時的判斷。邏輯還是比較簡單的。
- 再看看tryAcquire方法
- 這個方法的調(diào)用棧也是比較多，之后會進入下面這個方法
上面的lua(俗稱膠水語言)腳本比較重要，主要是為了執(zhí)行命令的原子性解釋一下：
- KEYS[1]代表你的key
- ARGV[1]代表你的key的存活時間，默認存活30秒
- ARGV[2]代表的是請求加鎖的客戶端ID，后面的1則理解為加鎖的次數(shù)，簡單理解就是 如果該客戶端多次對key加鎖時，就會執(zhí)行hincrby原子加1命令
第一段if就是判斷你的key是否存在，如果不存在，就執(zhí)行redis call(hset key ARGV[2],1)加鎖和設置redis call(pexpire key ARGV[1])存活時間；
當?shù)诙€客戶來加鎖時，第一個if判斷已存在key，就執(zhí)行第二個if判斷key的hash是否存在客戶端2的ID，很明顯不是;
則進入到最后的return返回該key的剩余存活時間
當加鎖成功后會在后臺啟動一個watch dog（看門狗）線程，key的默認存活時間為30秒，則watch dog每隔10秒鐘就會檢查一下客戶端1是否還持有該鎖，如果持有，就會不斷的延長鎖key的存活時間
所以這里建議大家在設置key的存活時間時，最好大于10秒，延續(xù)時間也大于等于10秒
所以，總體流程應該是這樣的。
總結
以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關文章
為了保證我們線上服務的并發(fā)性和安全性，目前我們的服務一般拋棄了單體應用，采用的都是擴展性很強的分布式架構，這篇文章主要介紹了使用Redis實現(xiàn)分布式鎖的方法,需要的朋友可以參考下2022-06-06
本文主要介紹了深入理解Redis大key的危害及解決方案,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2025-01-01
這篇文章主要給大家介紹了關于Redis分布式鎖的正確實現(xiàn)方法，文中通過示例代碼介紹的非常詳細，對大家學習或者使用Redis具有一定的參考學習價值，需要的朋友們下面來一起學習學習吧2019-08-08
redis中使用bloomfilter的白名單功能解決緩存穿透問題
本文主要介紹了redis中使用bloomfilter的白名單功能解決緩存穿透問題，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2023-07-07
這篇文章主要介紹了Redis之ZipList壓縮列表的使用,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2025-06-06
這篇文章給大家介紹了如何利用Redis?lua實現(xiàn)高效的讀寫鎖,讀寫鎖的好處就是能幫助客戶讀到的數(shù)據(jù)一定是最新的,寫鎖是排他鎖,而讀鎖是一個共享鎖,需要的朋友可以參考下2024-01-01
session存放在服務器，關閉瀏覽器不會失效，本文主要介紹了redis實現(xiàn)分布式session的解決方案，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-03-03
本文主要介紹了redis中5種數(shù)據(jù)基礎查詢命令，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2023-04-04
今天為大家介紹下linux系統(tǒng)下redis的主從配置方法,Linux系統(tǒng)下的redis的主從配置方法非常簡單下面是具體的操作步驟2018-09-09
Key的過期機制是Redis保持高可用性的重要策略,過期策略分為惰性過期和定期過期,惰性過期在每次訪問key時檢查是否過期,定期過期則由serverCron方法定時清理過期key,本文就來詳細的介紹一下,感興趣的可以了解一下2024-09-09