您好,登錄后才能下訂單哦!
本篇內(nèi)容主要講解“Redis分布式鎖實(shí)現(xiàn)的方法是什么”,感興趣的朋友不妨來看看。本文介紹的方法操作簡(jiǎn)單快捷,實(shí)用性強(qiáng)。下面就讓小編來帶大家學(xué)習(xí)“Redis分布式鎖實(shí)現(xiàn)的方法是什么”吧!
分布式鎖是 滿足分布式系統(tǒng)或集群模式下多進(jìn)程可見并且互斥的鎖。
基于Redis實(shí)現(xiàn)分布式鎖:
互斥:確保只能有一個(gè)線程獲取鎖;
非阻塞:嘗試獲取鎖,成功返回true,失敗返回false;
添加鎖過期時(shí)間,避免服務(wù)宕機(jī)引起死鎖。
SET lock thread1 NX EX 10
手動(dòng)釋放;DEL key1
超時(shí)釋放,獲取鎖時(shí)添加一個(gè)超時(shí)鎖;
package com.guor.utils; import org.springframework.data.redis.core.StringRedisTemplate; import java.util.concurrent.TimeUnit; public class RedisLock implements ILock{ private String name; private StringRedisTemplate stringRedisTemplate; public RedisLock(String name, StringRedisTemplate stringRedisTemplate) { this.name = name; this.stringRedisTemplate = stringRedisTemplate; } private static final String KEY_PREFIX = "lock:"; @Override public boolean tryLock(long timeout) { // 獲取線程唯一標(biāo)識(shí) long threadId = Thread.currentThread().getId(); // 獲取鎖 Boolean success = stringRedisTemplate.opsForValue() .setIfAbsent(KEY_PREFIX + name, threadId+"", timeout, TimeUnit.SECONDS); // 防止拆箱的空指針異常 return Boolean.TRUE.equals(success); } @Override public void unlock() { stringRedisTemplate.delete(KEY_PREFIX + name); } }
如果線程1獲取鎖,但線程1發(fā)生了阻塞,導(dǎo)致Redis超時(shí)釋放鎖;
此時(shí),線程2嘗試獲取鎖,成功,并執(zhí)行業(yè)務(wù);
此時(shí),線程1重新開始執(zhí)行任務(wù),并執(zhí)行完畢,執(zhí)行釋放鎖(即刪除鎖);
但是,線程1刪除的鎖,和線程2的鎖是同一把鎖,這就是分布式鎖誤刪問題
;
在釋放鎖時(shí),釋放線程自己的分布式鎖,就可以解決這個(gè)問題。
package com.guor.utils; import cn.hutool.core.lang.UUID; import org.springframework.data.redis.core.StringRedisTemplate; import java.util.concurrent.TimeUnit; public class RedisLock implements ILock{ private String name; private StringRedisTemplate stringRedisTemplate; public RedisLock(String name, StringRedisTemplate stringRedisTemplate) { this.name = name; this.stringRedisTemplate = stringRedisTemplate; } private static final String KEY_PREFIX = "lock:"; private static final String UUID_PREFIX = UUID.randomUUID().toString(true) + "-"; @Override public boolean tryLock(long timeout) { // 獲取線程唯一標(biāo)識(shí) String threadId = UUID_PREFIX + Thread.currentThread().getId(); // 獲取鎖 Boolean success = stringRedisTemplate.opsForValue() .setIfAbsent(KEY_PREFIX + name, threadId, timeout, TimeUnit.SECONDS); // 防止拆箱的空指針異常 return Boolean.TRUE.equals(success); } @Override public void unlock() { // 獲取線程唯一標(biāo)識(shí) String threadId = UUID_PREFIX + Thread.currentThread().getId(); // 獲取鎖中的標(biāo)識(shí) String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name); // 判斷標(biāo)示是否一致 if(threadId.equals(id)) { // 釋放鎖 stringRedisTemplate.delete(KEY_PREFIX + name); } } }
SETNX
實(shí)現(xiàn)的分布式鎖存在下面幾個(gè)問題同一個(gè)線程無法多次獲取同一把鎖。
獲取鎖只嘗試一次就返回false,沒有重試機(jī)制。
鎖的超時(shí)釋放雖然可以避免死鎖,但如果業(yè)務(wù)執(zhí)行耗時(shí)較長(zhǎng),也會(huì)導(dǎo)致鎖釋放,存在安全隱患。
如果Redis是集群部署的,主從同步存在延遲,當(dāng)主機(jī)宕機(jī)時(shí),此時(shí)會(huì)選一個(gè)從作為主機(jī),但是此時(shí)的從沒有鎖標(biāo)識(shí),此時(shí),其它線程可能會(huì)獲取到鎖,導(dǎo)致安全問題。
Redisson是一個(gè)在Redis的基礎(chǔ)上實(shí)現(xiàn)的Java駐內(nèi)存數(shù)據(jù)網(wǎng)格。它不僅提供了一系列的分布式的Java常用對(duì)象,還提供了許多分布式服務(wù),其中包含各種分布式鎖的實(shí)現(xiàn)。
<!--redisson--> <dependency> <groupId>org.redisson</groupId> <artifactId>redisson</artifactId> <version>3.13.6</version> </dependency>
package com.guor.config; import org.redisson.Redisson; import org.redisson.api.RedissonClient; import org.redisson.config.Config; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RedissonConfig { @Bean public RedissonClient redissonClient(){ // 配置 Config config = new Config(); /** * 單點(diǎn)地址useSingleServer,集群地址useClusterServers */ config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456"); // 創(chuàng)建RedissonClient對(duì)象 return Redisson.create(config); } }
package com.guor; import lombok.extern.slf4j.Slf4j; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.redisson.api.RLock; import org.redisson.api.RedissonClient; import org.springframework.boot.test.context.SpringBootTest; import javax.annotation.Resource; import java.util.concurrent.TimeUnit; @Slf4j @SpringBootTest class RedissonTest { @Resource private RedissonClient redissonClient; private RLock lock; @BeforeEach void setUp() { // 獲取指定名稱的鎖 lock = redissonClient.getLock("nezha"); } @Test void test() throws InterruptedException { // 嘗試獲取鎖 boolean isLock = lock.tryLock(1L, TimeUnit.SECONDS); if (!isLock) { log.error("獲取鎖失敗"); return; } try { log.info("哪吒最帥,哈哈哈"); } finally { // 釋放鎖 lock.unlock(); } } }
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException { // 最大等待時(shí)間 long time = unit.toMillis(waitTime); long current = System.currentTimeMillis(); long threadId = Thread.currentThread().getId(); Long ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId); if (ttl == null) { return true; } else { // 剩余等待時(shí)間 = 最大等待時(shí)間 - 獲取鎖失敗消耗的時(shí)間 time -= System.currentTimeMillis() - current; if (time <= 0L) {// 獲取鎖失敗 this.acquireFailed(waitTime, unit, threadId); return false; } else { // 再次嘗試獲取鎖 current = System.currentTimeMillis(); // subscribe訂閱其它釋放鎖的信號(hào) RFuture<RedissonLockEntry> subscribeFuture = this.subscribe(threadId); // 當(dāng)Future在等待指定時(shí)間time內(nèi)完成時(shí),返回true if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) { if (!subscribeFuture.cancel(false)) { subscribeFuture.onComplete((res, e) -> { if (e == null) { // 取消訂閱 this.unsubscribe(subscribeFuture, threadId); } }); } this.acquireFailed(waitTime, unit, threadId); return false;// 獲取鎖失敗 } else { try { // 剩余等待時(shí)間 = 剩余等待時(shí)間 - 獲取鎖失敗消耗的時(shí)間 time -= System.currentTimeMillis() - current; if (time <= 0L) { this.acquireFailed(waitTime, unit, threadId); boolean var20 = false; return var20; } else { boolean var16; do { long currentTime = System.currentTimeMillis(); // 重試獲取鎖 ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId); if (ttl == null) { var16 = true; return var16; } // 再次失敗了,再看一下剩余時(shí)間 time -= System.currentTimeMillis() - currentTime; if (time <= 0L) { this.acquireFailed(waitTime, unit, threadId); var16 = false; return var16; } // 再重試獲取鎖 currentTime = System.currentTimeMillis(); if (ttl >= 0L && ttl < time) { // 通過信號(hào)量的方式嘗試獲取信號(hào),如果等待時(shí)間內(nèi),依然沒有結(jié)果,會(huì)返回false ((RedissonLockEntry)subscribeFuture.getNow()).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } else { ((RedissonLockEntry)subscribeFuture.getNow()).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS); } time -= System.currentTimeMillis() - currentTime; } while(time > 0L); this.acquireFailed(waitTime, unit, threadId); var16 = false; return var16; } } finally { this.unsubscribe(subscribeFuture, threadId); } } } } }
private void scheduleExpirationRenewal(long threadId) { RedissonLock.ExpirationEntry entry = new RedissonLock.ExpirationEntry(); // this.getEntryName():鎖的名字,一個(gè)鎖對(duì)應(yīng)一個(gè)entry // putIfAbsent:如果不存在,將鎖和entry放到map里 RedissonLock.ExpirationEntry oldEntry = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.putIfAbsent(this.getEntryName(), entry); if (oldEntry != null) { // 同一個(gè)線程多次獲取鎖,相當(dāng)于重入 oldEntry.addThreadId(threadId); } else { // 如果是第一次 entry.addThreadId(threadId); // 更新有效期 this.renewExpiration(); } }
更新有效期,遞歸調(diào)用更新有效期,永不過期
private void renewExpiration() { // 從map中得到當(dāng)前鎖的entry RedissonLock.ExpirationEntry ee = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName()); if (ee != null) { // 開啟延時(shí)任務(wù) Timeout task = this.commandExecutor.getConnectionManager().newTimeout(new TimerTask() { public void run(Timeout timeout) throws Exception { RedissonLock.ExpirationEntry ent = (RedissonLock.ExpirationEntry)RedissonLock.EXPIRATION_RENEWAL_MAP.get(RedissonLock.this.getEntryName()); if (ent != null) { // 取出線程id Long threadId = ent.getFirstThreadId(); if (threadId != null) { // 刷新有效期 RFuture<Boolean> future = RedissonLock.this.renewExpirationAsync(threadId); future.onComplete((res, e) -> { if (e != null) { RedissonLock.log.error("Can't update lock " + RedissonLock.this.getName() + " expiration", e); } else { if (res) { // 遞歸調(diào)用更新有效期,永不過期 RedissonLock.this.renewExpiration(); } } }); } } } }, this.internalLockLeaseTime / 3L, TimeUnit.MILLISECONDS);// 10S ee.setTimeout(task); } }
protected RFuture<Boolean> renewExpirationAsync(long threadId) { return this.evalWriteAsync(this.getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, // 判斷當(dāng)前線程的鎖是否是當(dāng)前線程 "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then // 更新有效期 redis.call('pexpire', KEYS[1], ARGV[1]); return 1; end; return 0;", Collections.singletonList(this.getName()), this.internalLockLeaseTime, this.getLockName(threadId)); }
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) { // 鎖釋放時(shí)間 this.internalLockLeaseTime = unit.toMillis(leaseTime); return this.evalWriteAsync(this.getName(), LongCodec.INSTANCE, command, // 判斷鎖成功 "if (redis.call('exists', KEYS[1]) == 0) then redis.call('hincrby', KEYS[1], ARGV[2], 1); // 如果不存在,記錄鎖標(biāo)識(shí),次數(shù)+1 redis.call('pexpire', KEYS[1], ARGV[1]); // 設(shè)置鎖有效期 return nil; // 相當(dāng)于Java的null end; if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then redis.call('hincrby', KEYS[1], ARGV[2], 1); // 如果存在,判斷鎖標(biāo)識(shí)是否是自己的,次數(shù)+1 redis.call('pexpire', KEYS[1], ARGV[1]); // 設(shè)置鎖有效期 return nil; end; // 判斷鎖失敗,pttl:指定鎖剩余有效期,單位毫秒,KEYS[1]:鎖的名稱 return redis.call('pttl', KEYS[1]);", Collections.singletonList(this.getName()), this.internalLockLeaseTime, this.getLockName(threadId)); }
public RFuture<Void> unlockAsync(long threadId) { RPromise<Void> result = new RedissonPromise(); RFuture<Boolean> future = this.unlockInnerAsync(threadId); future.onComplete((opStatus, e) -> { // 取消更新任務(wù) this.cancelExpirationRenewal(threadId); if (e != null) { result.tryFailure(e); } else if (opStatus == null) { IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: " + this.id + " thread-id: " + threadId); result.tryFailure(cause); } else { result.trySuccess((Object)null); } }); return result; }
void cancelExpirationRenewal(Long threadId) { // 從map中取出當(dāng)前鎖的定時(shí)任務(wù)entry RedissonLock.ExpirationEntry task = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName()); if (task != null) { if (threadId != null) { task.removeThreadId(threadId); } // 刪除定時(shí)任務(wù) if (threadId == null || task.hasNoThreads()) { Timeout timeout = task.getTimeout(); if (timeout != null) { timeout.cancel(); } EXPIRATION_RENEWAL_MAP.remove(this.getEntryName()); } } }
到此,相信大家對(duì)“Redis分布式鎖實(shí)現(xiàn)的方法是什么”有了更深的了解,不妨來實(shí)際操作一番吧!這里是億速云網(wǎng)站,更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢,關(guān)注我們,繼續(xù)學(xué)習(xí)!
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。