이 글에서는 Redis를 사용하여 안전하고 안정적인 분산 잠금을 구현하는 방법을 소개하고 분산 잠금 구현에 대한 주요 요소와 일반적인 오해를 설명합니다. 도움이 필요한 친구들이 모두 참고할 수 있기를 바랍니다.
동시 시나리오에서 여러 프로세스 또는 스레드가 리소스를 공유하는 경우 리소스에 대한 액세스를 상호 배제해야 합니다. 독립형 시스템에서는 Java 동시성 패키지의 API, 동기화 키워드 등을 사용하여 문제를 해결할 수 있지만 분산 시스템에서는 이러한 방법을 더 이상 적용할 수 없으며 분산 잠금을 직접 구현해야 합니다. .
일반적인 분산 잠금 구현 솔루션에는 데이터베이스 기반, Redis 기반, Zookeeper 기반 등이 포함됩니다. Redis 주제의 일부로 이 기사에서는 Redis 기반 분산 잠금 구현에 대해 설명합니다. [관련 권장사항: Redis 동영상 튜토리얼]
분석 및 구현
문제 분석
분산 잠금과 JVM 내장 잠금은 동일한 목적을 가집니다. 즉, 애플리케이션이 예상대로 액세스하거나 작동할 수 있도록 허용하는 것입니다. 공유 리소스는 여러 스레드가 동일한 리소스에서 동시에 작동하는 것을 방지하여 혼란스럽고 제어할 수 없는 시스템 작동을 초래합니다. 제품 재고 공제 및 쿠폰 공제와 같은 시나리오에서 자주 사용됩니다.
이론적으로 잠금의 보안과 효율성을 보장하려면 분산 잠금은 최소한 다음 조건을 충족해야 합니다.
- 상호 배타성: 동시에 하나의 스레드만 잠금을 얻을 수 있습니다.
- 교착 상태 없음: 스레드가 잠금을 획득한 후에는 해제가 보장되어야 합니다. 스레드가 잠금을 획득한 후 애플리케이션이 다운되더라도 제한된 시간 내에 해제될 수 있습니다.
- 잠금 및 잠금 해제는 동일한 스레드여야 합니다.
구현 측면에서 배포 잠금은 대략 세 단계로 나뉩니다.
a-리소스 운영 권한 획득
- b-리소스 운영 권한 해제
- Java의 내장 잠금이든 분산 잠금이든, 어떤 분산 구현 방식을 사용하든 상관없이 a와 c의 두 단계를 중심으로 진행됩니다. Redis는 다음과 같은 이유로 자연스럽게 분산 잠금을 구현하는 데 적합합니다.
-
Redis는 명령 처리 단계에서 단일 스레드 처리를 사용하므로 동시에 하나의 스레드만 동일한 키를 처리할 수 있으므로 다중 스레드 경쟁 조건이 없습니다. 문제.
SET 키 값 NX PX 밀리초 명령은 키가 존재하지 않을 때 만료 시간이 있는 키를 추가하여 보안 잠금을 지원합니다.
- Lua 스크립트와 DEL 명령은 안전한 잠금 해제를 위한 안정적인 지원을 제공합니다.
SET key value NX PX milliseconds命令在不存在key的情况下添加具有过期时间的key，为安全加锁提供支持。
- Lua脚本和DEL命令为安全解锁提供可靠支撑。
代码实现
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
<version>${your-spring-boot-version}</version>
</dependency>로그인 후 복사
在application.properties增加以下内容，单机版Redis实例。
spring.redis.database=0
spring.redis.host=localhost
spring.redis.port=6379
로그인 후 복사
@Configuration
public class RedisConfig {
// 自己定义了一个 RedisTemplate
@Bean
@SuppressWarnings("all")
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory)
throws UnknownHostException {
// 我们为了自己开发方便，一般直接使用 <String, Object>
RedisTemplate<String, Object> template = new RedisTemplate<String,
Object>();
template.setConnectionFactory(factory);
// Json序列化配置
Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper om = new ObjectMapper();
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jackson2JsonRedisSerializer.setObjectMapper(om);
// String 的序列化
StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
// key采用String的序列化方式
template.setKeySerializer(stringRedisSerializer);
// hash的key也采用String的序列化方式
template.setHashKeySerializer(stringRedisSerializer);
// value序列化方式采用jackson
template.setValueSerializer(jackson2JsonRedisSerializer);
// hash的value序列化方式采用jackson
template.setHashValueSerializer(jackson2JsonRedisSerializer);
template.afterPropertiesSet();
return template;
}
}로그인 후 복사
@Service
public class RedisLock {
@Resource
private RedisTemplate<String, Object> redisTemplate;
/**
* 加锁，最多等待maxWait毫秒
*
* @param lockKey 锁定key
* @param lockValue 锁定value
* @param timeout 锁定时长（毫秒）
* @param maxWait 加锁等待时间（毫秒）
* @return true-成功，false-失败
*/
public boolean tryAcquire(String lockKey, String lockValue, int timeout, long maxWait) {
long start = System.currentTimeMillis();
while (true) {
// 尝试加锁
Boolean ret = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, timeout, TimeUnit.MILLISECONDS);
if (!ObjectUtils.isEmpty(ret) && ret) {
return true;
}
// 计算已经等待的时间
long now = System.currentTimeMillis();
if (now - start > maxWait) {
return false;
}
try {
Thread.sleep(200);
} catch (Exception ex) {
return false;
}
}
}
/**
* 释放锁
*
* @param lockKey 锁定key
* @param lockValue 锁定value
* @return true-成功，false-失败
*/
public boolean releaseLock(String lockKey, String lockValue) {
// lua脚本
String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>(script, Long.class);
Long result = redisTemplate.opsForValue().getOperations().execute(redisScript, Collections.singletonList(lockKey), lockValue);
return result != null && result > 0L;
}
}로그인 후 복사
@SpringBootTest
class RedisDistLockDemoApplicationTests {
@Resource
private RedisLock redisLock;
@Test
public void testLock() {
redisLock.tryAcquire("abcd", "abcd", 5 * 60 * 1000, 5 * 1000);
redisLock.releaseLock("abcd", "abcd");
}
}로그인 후 복사
安全隐患
可能很多同学（也包括我）在日常工作中都是使用上面的实现方式，看似是稳妥的：
- 使用
set命令NX、PX选项进行加锁，保证了加锁互斥，避免了死锁； - 使用lua脚本解锁，防止解除其他线程的锁；
- 加锁、解锁命令都是原子操作；
其实以上实现的稳妥有个前提条件：单机版Redis、开启AOF持久化方式并设置appendfsync=always코드 구현
Maven 종속성
@Test
public void testRedLock() throws InterruptedException {
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379");
final RedissonClient client = Redisson.create(config);
// 获取锁实例
final RLock lock = client.getLock("test-lock");
// 加锁
lock.lock(60 * 1000, TimeUnit.MILLISECONDS);
try {
// 假装做些什么事情
Thread.sleep(50 * 1000);
} catch (Exception ex) {
ex.printStackTrace();
} finally {
//解锁
lock.unlock();
}
}로그인 후 복사
로그인 후 복사
구성 파일
독립형 Redis 인스턴스인 application.properties에 다음 콘텐츠를 추가하세요. rrreee
RedisConfig
rrreee
RedisLock
rrreee테스트 사례
rrreee
안전 위험
아마도 많은 학생들(저 포함)이 이 작업을 하고 있을 것입니다. 그들의 일상 업무 위의 구현 방법은 모두에서 사용되며 안전해 보입니다.
🎜🎜
set 명령
NX 및
PX</code 사용 > 잠금 옵션, 잠금 상호 배제가 보장되고 교착 상태가 방지됩니다. 🎜🎜다른 스레드의 잠금 해제를 방지하려면 잠금 해제에 lua 스크립트를 사용하세요. 🎜🎜잠금 및 잠금 해제 명령은 모두 원자적 작업입니다. 위 구현이 안정적이기 위한 전제 조건: Redis의 독립형 버전, AOF 지속성 모드를 활성화하고 <code>appendfsync=always를 설정합니다. 🎜🎜근데 센트리 모드와 클러스터 모드에서는 문제가 있을 수 있는데 왜 그럴까요? 🎜🎜센티넬 모드와 클러스터 모드는 마스터-슬레이브 아키텍처를 기반으로 하며 명령 전파를 통해 마스터와 슬레이브 간에 데이터 동기화가 이루어지며 명령 전파는 비동기식입니다. 🎜🎜그래서 마스터 노드 데이터가 성공적으로 기록되었으나 슬레이브 노드에 통보되기 전에 마스터 노드가 다운될 가능성이 있습니다. 🎜🎜Failover를 통해 슬레이브 노드가 새로운 마스터 노드로 승격되면 다른 스레드가 성공적으로 다시 잠금할 수 있는 기회를 가지게 되어 분산 잠금의 상호 배제 조건이 충족되지 않게 됩니다. 🎜🎜🎜공식 RedLock🎜🎜🎜🎜클러스터 모드에서는 클러스터의 모든 노드가 안정적으로 실행되고 장애 조치가 발생하지 않으면 보안이 보장됩니다. 그러나 어떤 시스템도 100% 안정성을 보장할 수 없으며 Redis 기반의 분산 잠금은 내결함성을 고려해야 합니다. 🎜🎜🎜마스터-슬레이브 동기화는 비동기 복제 원칙을 기반으로 하기 때문에 센트리 모드와 클러스터 모드는 본질적으로 이 조건을 충족할 수 없습니다. 이러한 이유로 Redis 작성자는 특별히 솔루션-RedLock(Redis Distribute Lock)을 제안했습니다. 🎜🎜🎜디자인 아이디어🎜🎜🎜 공식 문서에 따르면 RedLock의 디자인 아이디어가 소개되어 있습니다. 🎜🎜먼저 환경 요구 사항에 대해 이야기해 보겠습니다. N(N>=3)개의 독립적으로 배포된 Redis 인스턴스가 필요합니다. 서로 마스터-슬레이브 복제, 장애 조치 및 기타 기술이 필요하지 않습니다. 🎜🎜잠금을 획득하기 위해 클라이언트는 다음 프로세스를 따릅니다. 🎜
- 获取当前时间（毫秒）作为开始时间start；
- 使用相同的key和随机value，按顺序向所有N个节点发起获取锁的请求。当向每个实例设置锁时，客户端会使用一个过期时间（小于锁的自动释放时间）。比如锁的自动释放时间是10秒，这个超时时间应该是5-50毫秒。这是为了防止客户端在一个已经宕机的实例浪费太多时间：如果Redis实例宕机，客户端尽快处理下一个实例。
- 客户端计算加锁消耗的时间cost（cost=start-now）。只有客户端在半数以上实例加锁成功，并且整个耗时小于整个有效时间（ttl），才能认为当前客户端加锁成功。
- 如果客户端加锁成功，那么整个锁的真正有效时间应该是：validTime=ttl-cost。
- 如果客户端加锁失败（可能是获取锁成功实例数未过半，也可能是耗时超过ttl），那么客户端应该向所有实例尝试解锁（即使刚刚客户端认为加锁失败）。
RedLock的设计思路延续了Redis内部多种场景的投票方案，通过多个实例分别加锁解决竞态问题，虽然加锁消耗了时间，但是消除了主从机制下的安全问题。
代码实现
官方推荐Java实现为Redisson，它具备可重入特性，按照RedLock进行实现，支持独立实例模式、集群模式、主从模式、哨兵模式等；API比较简单，上手容易。示例如下（直接通过测试用例）：
@Test
public void testRedLock() throws InterruptedException {
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379");
final RedissonClient client = Redisson.create(config);
// 获取锁实例
final RLock lock = client.getLock("test-lock");
// 加锁
lock.lock(60 * 1000, TimeUnit.MILLISECONDS);
try {
// 假装做些什么事情
Thread.sleep(50 * 1000);
} catch (Exception ex) {
ex.printStackTrace();
} finally {
//解锁
lock.unlock();
}
}로그인 후 복사
로그인 후 복사
Redisson封装的非常好，我们可以像使用Java内置的锁一样去使用，代码简洁的不能再少了。关于Redisson源码的分析，网上有很多文章大家可以找找看。
全文总结
分布式锁是我们研发过程中常用的的一种解决并发问题的方式，Redis是只是一种实现方式。
关键的是要弄清楚加锁、解锁背后的原理，以及实现分布式锁需要解决的核心问题，同时考虑我们所采用的中间件有什么特性可以支撑。了解这些后，实现起来就不是什么问题了。
学习了RedLock的思想，我们是不是也可以在自己的应用程序内实现了分布式锁呢？欢迎沟通！
更多编程相关知识，请访问：编程入门！！
위 내용은 Redis를 사용하여 안전하고 안정적인 분산 잠금 구현의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!
![[웹 프런트엔드] Node.js 빠른 시작](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
