Redis를 사용하여 안전하고 안정적인 분산 잠금 구현

이 글에서는 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命令为安全解锁提供可靠支撑。

代码实现

• Maven依赖

<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

• RedisConfig

@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;
    }
}

• RedisLock

@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(&#39;get&#39;,KEYS[1]) == ARGV[1] then return redis.call(&#39;del&#39;,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사용 > 잠금 옵션, 잠금 상호 배제가 보장되고 교착 상태가 방지됩니다. 🎜🎜다른 스레드의 잠금 해제를 방지하려면 잠금 해제에 lua 스크립트를 사용하세요. 🎜🎜잠금 및 잠금 해제 명령은 모두 원자적 작업입니다. 위 구현이 안정적이기 위한 전제 조건: Redis의 독립형 버전, AOF 지속성 모드를 활성화하고 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 중국어 웹사이트의 기타 관련 기사를 참조하세요!