1. 분산 잠금이란 무엇인가요?
분산 잠금은 분산 시스템이나 클러스터 모드에서 여러 프로세스를 볼 수 있고 상호 배타적이게 하는 잠금입니다.
Redis 기반 분산 잠금 구현:
1. 잠금 획득
서비스 중단 시간으로 인한 교착 상태를 방지하려면 잠금 만료 시간을 추가하세요.
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 실수로 삭제되는 문제;
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 중국어 웹사이트의 기타 관련 기사를 참조하세요!
![[웹 프런트엔드] Node.js 빠른 시작](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
