Redis 분산 잠금을 구현하는 방법은 무엇입니까?
- 王林앞으로
- 2023-05-28 17:58:271714검색
1. 분산 잠금이란 무엇인가요?
분산 잠금은 분산 시스템이나 클러스터 모드에서 여러 프로세스를 볼 수 있고 상호 배타적이게 하는 잠금입니다.
Redis 기반 분산 잠금 구현:
1. 잠금 획득
상호 배제: 단 하나의 스레드만 잠금을 획득할 수 있도록 보장
비 차단: 잠금 획득을 시도하고 성공하면 true를 반환합니다. , 실패하면 false
서비스 중단 시간으로 인한 교착 상태를 방지하려면 잠금 만료 시간을 추가하세요.
SET 잠금 스레드1 NX EX 10SET lock thread1 NX EX 10
2、释放锁
手动释放；
DEL key1超时释放，获取锁时添加一个超时锁；
二、代码实例
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) {
// 获取线程唯一标识
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发生了阻塞，导致Redis超时释放锁；
此时，线程2尝试获取锁，成功，并执行业务；
此时，线程1重新开始执行任务，并执行完毕，执行释放锁（即删除锁）；
但是，线程1删除的锁，和线程2的锁是同一把锁，这就是
分布式锁误删问题；
在释放锁时，释放线程自己的分布式锁，就可以解决这个问题。
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) {
// 获取线程唯一标识
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() {
// 获取线程唯一标识
String threadId = UUID_PREFIX + Thread.currentThread().getId();
// 获取锁中的标识
String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
// 判断标示是否一致
if(threadId.equals(id)) {
// 释放锁
stringRedisTemplate.delete(KEY_PREFIX + name);
}
}
}三、基于SETNX
2, 잠금 해제
DEL key1
시간 초과 해제, 획득 시 시간 초과 추가 lock Lock
2. 코드 예 <!--redisson-->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.13.6</version>
</dependency>
위 코드에는 실수로 잠금을 삭제하는 문제가 있습니다.
스레드 1이 잠금을 획득했지만 스레드가 1이 차단되어 Redis가 시간 초과 후 잠금을 해제합니다.
이때 스레드 2는 잠금 획득을 시도하고 성공하여 비즈니스를 실행합니다.
이때 스레드 1이 다시 시작됩니다. 작업을 완료하고 잠금을 해제합니다(즉, 잠금을 삭제합니다) ;
그러나 스레드 1이 삭제한 잠금은 스레드 2의 잠금과 동일한 잠금입니다. 이는 배포된 입니다. lock 실수로 삭제되는 문제;
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();
/**
* 单点地址useSingleServer，集群地址useClusterServers
*/
config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123456");
// 创建RedissonClient对象
return Redisson.create(config);
}
}
3. SETNX를 기반으로 구현된 분산 잠금에는 다음과 같은 문제가 있습니다1. 재진입이 불가능합니다동일한 스레드가 동일한 잠금을 여러 번 얻을 수 없습니다. 2. 재시도 없음 잠금 획득은 한 번만 시도하고 false를 반환합니다. 3. 시간 초과 해제 잠금 시간 초과 해제는 교착 상태를 피할 수 있지만, 비즈니스 실행에 오랜 시간이 걸리면 잠금 해제로 이어져 보안 위험이 발생할 수 있습니다. 🎜🎜4. 마스터-슬레이브 일관성 🎜🎜Redis가 클러스터에 배포되면 마스터-슬레이브 동기화가 지연될 수 있습니다. 호스트가 다운되면 슬레이브가 호스트로 선택됩니다. 이때 다른 스레드가 잠금을 획득하여 보안 문제가 발생할 수 있습니다. 🎜🎜4. Redisson은 분산 잠금을 구현합니다🎜🎜Redisson은 Redis 구현을 기반으로 하는 Java 메모리 데이터 그리드입니다. 일반적으로 사용되는 분산 Java 객체를 제공하는 것 외에도 다양한 분산 잠금 구현을 포함하여 많은 분산 서비스도 제공합니다. 🎜🎜1.pom🎜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();
}
}
}🎜2. 구성 클래스🎜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 = this.tryAcquire(waitTime, leaseTime, unit, threadId);
if (ttl == null) {
return true;
} else {
// 剩余等待时间 = 最大等待时间 - 获取锁失败消耗的时间
time -= System.currentTimeMillis() - current;
if (time <= 0L) {// 获取锁失败
this.acquireFailed(waitTime, unit, threadId);
return false;
} else {
// 再次尝试获取锁
current = System.currentTimeMillis();
// subscribe订阅其它释放锁的信号
RFuture<RedissonLockEntry> subscribeFuture = this.subscribe(threadId);
// 当Future在等待指定时间time内完成时，返回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 {
// 剩余等待时间 = 剩余等待时间 - 获取锁失败消耗的时间
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;
}
// 再次失败了，再看一下剩余时间
time -= System.currentTimeMillis() - currentTime;
if (time <= 0L) {
this.acquireFailed(waitTime, unit, threadId);
var16 = false;
return var16;
}
// 再重试获取锁
currentTime = System.currentTimeMillis();
if (ttl >= 0L && ttl < time) {
// 通过信号量的方式尝试获取信号，如果等待时间内，依然没有结果，会返回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);
}
}
}
}
}🎜3. 테스트 클래스🎜private void scheduleExpirationRenewal(long threadId) {
RedissonLock.ExpirationEntry entry = new RedissonLock.ExpirationEntry();
// this.getEntryName()：锁的名字，一个锁对应一个entry
// putIfAbsent：如果不存在，将锁和entry放到map里
RedissonLock.ExpirationEntry oldEntry = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.putIfAbsent(this.getEntryName(), entry);
if (oldEntry != null) {
// 同一个线程多次获取锁，相当于重入
oldEntry.addThreadId(threadId);
} else {
// 如果是第一次
entry.addThreadId(threadId);
// 更新有效期
this.renewExpiration();
}
}🎜 5. tryLock 소스 코드🎜🎜를 획득해 보세요🎜private void renewExpiration() {
// 从map中得到当前锁的entry
RedissonLock.ExpirationEntry ee = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName());
if (ee != null) {
// 开启延时任务
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) {
// 递归调用更新有效期，永不过期
RedissonLock.this.renewExpiration();
}
}
});
}
}
}
}, this.internalLockLeaseTime / 3L, TimeUnit.MILLISECONDS);// 10S
ee.setTimeout(task);
}
}🎜2. 잠금 기간🎜protected RFuture<Boolean> renewExpirationAsync(long threadId) {
return this.evalWriteAsync(this.getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
// 判断当前线程的锁是否是当前线程
"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) {
// 锁释放时间
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); // 如果不存在，记录锁标识，次数+1
redis.call('pexpire', KEYS[1], ARGV[1]); // 设置锁有效期
return nil; // 相当于Java的null
end;
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
redis.call('hincrby', KEYS[1], ARGV[2], 1); // 如果存在，判断锁标识是否是自己的，次数+1
redis.call('pexpire', KEYS[1], ARGV[1]); // 设置锁有效期
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) -> {
// 取消更新任务
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;
}🎜3. 잠금 해제 소스 코드를 해제합니다. 🎜🎜1. 업데이트 작업 취소🎜void cancelExpirationRenewal(Long threadId) {
// 从map中取出当前锁的定时任务entry
RedissonLock.ExpirationEntry task = (RedissonLock.ExpirationEntry)EXPIRATION_RENEWAL_MAP.get(this.getEntryName());
if (task != null) {
if (threadId != null) {
task.removeThreadId(threadId);
}
// 删除定时任务
if (threadId == null || task.hasNoThreads()) {
Timeout timeout = task.getTimeout();
if (timeout != null) {
timeout.cancel();
}
EXPIRATION_RENEWAL_MAP.remove(this.getEntryName());
}
}
}🎜2. 예약된 작업 삭제🎜rrreee위 내용은 Redis 분산 잠금을 구현하는 방법은 무엇입니까?의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!
성명：
이 기사는 yisu.com에서 복제됩니다. 침해가 있는 경우 admin@php.cn으로 문의하시기 바랍니다. 삭제