분산잠금 : 진입부터 매몰까지 5건

오늘 여러분과 공유하고 싶은 것은 분산 잠금입니다. 이 기사에서는 5가지 사례, 다이어그램, 소스 코드 분석 등을 사용하여 분석합니다.

공통 동기화, 잠금 등의 잠금은 단일 JVM이 구현되었습니다. 분산 시나리오에서는 어떻게 해야 합니까? 이때 분산 잠금이 나타났습니다. JVM的实现的，如果分布式场景下怎么办呢？这时候分布式锁就出现了。

关于分布式的实现方案，在业界流行的有三种：

1、基于数据库

2、基于Redis

3、基于Zookeeper

另外，还有使用etcd、consul

분산 구현 솔루션과 관련하여 업계에서 널리 사용되는 세 가지 솔루션이 있습니다. 🎜🎜1. 데이터베이스 기반 🎜🎜2 Redis🎜🎜3. Zookeeper🎜🎜또한 etcd, consul. 🎜

개발에서 가장 일반적으로 사용되는 것은 Redis 및 Zookeeper에는 두 가지 해결 방법이 있는데, 두 해결 방법 중 가장 복잡하고 문제를 일으킬 가능성이 가장 높은 해결 방법은 Redis 구현 계획이므로 오늘은 Redis 구현 솔루션. Redis和Zookeeper两种方案，并且两种方案中最复杂的，最容易出问题的就是Redis的实现方案，所以，我们今天就来把Redis实现方案都聊聊。

本文主要内容

分布式锁场景

估计部分朋友还不太清楚分布式的使用场景，下面我简单罗列三种：

案例1

如下代码模拟了下单减库存的场景，我们分析下在高并发场景下会存在什么问题

@RestController
public class IndexController {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 模拟下单减库存的场景
     * @return
     */
    @RequestMapping(value = "/duduct_stock")
    public String deductStock(){
        // 从redis 中拿当前库存的值
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock",realStock + "");
            System.out.println("扣减成功，剩余库存：" + realStock);
        }else{
            System.out.println("扣减失败，库存不足");
        }
        return "end";
    }
}

假设在Redis

이 글의 주요 내용

분산 잠금 시나리오 < / span>

일부 친구들은 분산 사용 시나리오에 대해 명확하지 않은 것으로 추정됩니다. 아래에 세 가지 유형을 간략하게 나열하겠습니다.

🎜

사례 1

🎜다음 코드는 재고를 줄이기 위해 주문하는 시나리오를 시뮬레이션합니다. 높은 동시성 시나리오에서 어떤 문제가 발생할지 분석해 보겠습니다.🎜

int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));

🎜Redis의 초기 주식 가치는 100입니다. 🎜🎜이제 5개의 클라이언트가 동시에 이 인터페이스를 요청하고 있으며 동시 실행이 있을 수 있습니다🎜

int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));

这行代码，获取到的值都为100，紧跟着判断大于0后都进行-1操作，最后设置到redis 中的值都为99。但正常执行完成后redis中的值应为 95。

案例2-使用synchronized 实现单机锁

在遇到案例1的问题后，大部分人的第一反应都会想到加锁来控制事务的原子性，如下代码所示：

@RequestMapping(value = "/duduct_stock")
public String deductStock(){
    synchronized (this){
        // 从redis 中拿当前库存的值
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock",realStock + "");
            System.out.println("扣减成功，剩余库存：" + realStock);
        }else{
            System.out.println("扣减失败，库存不足");
        }
    }
    return "end";
}

现在当有多个请求访问该接口时，同一时刻只有一个请求可进入方法体中进行库存的扣减，其余请求等候。

但我们都知道，synchronized 锁是属于JVM级别的，也就是我们俗称的“单机锁”。但现在基本大部分公司使用的都是集群部署，现在我们思考下以上代码在集群部署的情况下还能保证库存数据的一致性吗？

答案是不能，如上图所示，请求经Nginx分发后，可能存在多个服务同时从Redis中获取库存数据，此时只加synchronized （单机锁）是无效的，并发越高，出现问题的几率就越大。

案例3-使用SETNX实现分布式锁

setnx：将 key 的值设为 value，当且仅当 key 不存在。

若给定 key 已经存在，则 setnx 不做任何动作。

使用setnx实现简单的分布式锁：

/**
 * 模拟下单减库存的场景
 * @return
 */
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
    String lockKey = "product_001";
    // 使用 setnx 添加分布式锁
    // 返回 true 代表之前redis中没有key为 lockKey 的值，并已进行成功设置
    // 返回 false 代表之前redis中已经存在 lockKey 这个key了
    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "wangcp");
    if(!result){
        // 代表已经加锁了
        return "error_code";
    }

    // 从redis 中拿当前库存的值
    int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
    if(stock > 0){
        int realStock = stock - 1;
        stringRedisTemplate.opsForValue().set("stock",realStock + "");
        System.out.println("扣减成功，剩余库存：" + realStock);
    }else{
        System.out.println("扣减失败，库存不足");
    }

    // 释放锁
    stringRedisTemplate.delete(lockKey);
    return "end";
}

我们知道 Redis 是单线程执行，现在再看案例2中的流程图时，哪怕高并发场景下多个请求都执行到了setnx的代码，redis会根据请求的先后顺序进行排列，只有排列在队头的请求才能设置成功。其它请求只能返回“error_code”。

当setnx设置成功后，可执行业务代码对库存扣减，执行完成后对锁进行释放。

我们再来思考下以上代码已经完美实现分布式锁了吗？能够支撑高并发场景吗？答案并不是，上面的代码还是存在很多问题的，离真正的分布式锁还差的很远。

我们分析一下，上面的代码存在的问题：

死锁：假如第一个请求在setnx加锁完成后，执行业务代码时出现了异常，那释放锁的代码就无法执行，后面所有的请求也都无法进行操作了。

针对死锁的问题，我们对代码再次进行优化，添加try-finally，在finally中添加释放锁代码，这样无论如何都会执行释放锁代码，如下所示：

/**
     * 模拟下单减库存的场景
     * @return
     */
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
    String lockKey = "product_001";

    try{
        // 使用 setnx 添加分布式锁
        // 返回 true 代表之前redis中没有key为 lockKey 的值，并已进行成功设置
        // 返回 false 代表之前redis中已经存在 lockKey 这个key了
        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "wangcp");
        if(!result){
            // 代表已经加锁了
            return "error_code";
        }
        // 从redis 中拿当前库存的值
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock",realStock + "");
            System.out.println("扣减成功，剩余库存：" + realStock);
        }else{
            System.out.println("扣减失败，库存不足");
        }
    }finally {
        // 释放锁
        stringRedisTemplate.delete(lockKey);
    }

    return "end";
}

经过改进后的代码是否还存在问题呢？我们思考正常执行的情况下应该是没有问题，但我们假设请求在执行到业务代码时服务突然宕机了，或者正巧你的运维同事重新发版，粗暴的 kill -9 掉了呢，那代码还能执行 finally 吗？

案例4-加入过期时间

针对想到的问题，对代码再次进行优化，加入过期时间，这样即便出现了上述的问题，在时间到期后锁也会自动释放掉，不会出现“死锁”的情况。

@RequestMapping(value = "/duduct_stock")
public String deductStock(){
    String lockKey = "product_001";

    try{
        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,"wangcp",10,TimeUnit.SECONDS);
        if(!result){
            // 代表已经加锁了
            return "error_code";
        }
        // 从redis 中拿当前库存的值
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock",realStock + "");
            System.out.println("扣减成功，剩余库存：" + realStock);
        }else{
            System.out.println("扣减失败，库存不足");
        }
    }finally {
        // 释放锁
        stringRedisTemplate.delete(lockKey);
    }

    return "end";
}

现在我们再思考一下，给锁加入过期时间后就可以了吗？就可以完美运行不出问题了吗？

超时时间设置的10s真的合适吗？如果不合适设置多少秒合适呢？如下图所示

Pictures

동시에 세 개의 요청이 있다고 가정해 보겠습니다.

• 요청 1은 먼저 잠긴 후 15초 동안 실행되어야 하지만 실행 10초 후에 잠금이 만료되고 해제됩니다.
• 요청 2가 들어오면 잠긴 후 실행됩니다. 요청 2가 5초 동안 실행되면 요청 1이 실행되고 잠금이 해제되지만 이때 요청 2의 잠금이 해제됩니다.
• 요청 2가 5초 동안 실행되면 요청 3이 실행되기 시작하지만, 요청 2가 3초 동안 실행되면 요청 3의 잠금이 해제됩니다.

지금 3개의 요청을 시뮬레이션하면 문제를 확인할 수 있습니다. 실제로 높은 동시성 시나리오에서는 잠금이 "항상 유효하지 않음" 또는 "영구 유효하지 않음"에 직면할 수 있습니다.

그럼 구체적인 문제는 어디에 있나요? 요약하면 다음과 같습니다.

• 1. 잠금 해제 요청이 있을 때 해제된 잠금은 자신의 것이 아닙니다.
• 2. 시간 초과가 너무 짧고, 잠금 해제되기 전에 코드가 자동으로 해제됩니다.

문제를 고려하여 해당 솔루션을 생각합니다.

• 문제 1의 경우 고유 ID를 잠금 값으로 사용하여 요청이 들어올 때 고유 ID를 생성하는 것을 생각합니다. 해제할 때 먼저 획득 비교를 수행하고 비교가 동일할 때 진행하여 다른 요청 잠금을 해제하는 문제를 해결할 수 있습니다.
• 2번 질문에 대한 답변으로, 유효기간을 지속적으로 연장하는 것이 정말 적절한가요? 설정이 짧으면 시간이 지나면서 자동 해제되는 문제가 있고, 설정이 길면 "교착 상태"가 더 이상 발생하지 않지만 종료 후 일정 시간 동안 잠금이 해제되지 않습니다. 이 문제를 해결하는 방법은 무엇입니까?

사례 5-Redisson 분산 잠금

Spring Boot集成RedissonSteps

引入依赖

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.6.5</version>
</dependency>

初始化客户端

@Bean
public RedissonClient redisson(){
    // 单机模式
    Config config = new Config();
    config.useSingleServer().setAddress("redis://192.168.3.170:6379").setDatabase(0);
    return Redisson.create(config);
}

Redisson实现分布式锁

@RestController
public class IndexController {

    @Autowired
    private RedissonClient redisson;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 模拟下单减库存的场景
     * @return
     */
    @RequestMapping(value = "/duduct_stock")
    public String deductStock(){
        String lockKey = "product_001";
        // 1.获取锁对象
        RLock redissonLock = redisson.getLock(lockKey);
        try{
            // 2.加锁
            redissonLock.lock();  // 等价于 setIfAbsent(lockKey,"wangcp",10,TimeUnit.SECONDS);
            // 从redis 中拿当前库存的值
            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            if(stock > 0){
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock",realStock + "");
                System.out.println("扣减成功，剩余库存：" + realStock);
            }else{
                System.out.println("扣减失败，库存不足");
            }
        }finally {
            // 3.释放锁
            redissonLock.unlock();
        }
        return "end";
    }
}

Redisson 分布式锁实现原理图

图片

Redisson 底层源码分析

我们点击lock()方法，查看源码，最终看到以下代码

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        internalLockLeaseTime = unit.toMillis(leaseTime);

        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                  "if (redis.call(&#39;exists&#39;, KEYS[1]) == 0) then " +
                      "redis.call(&#39;hset&#39;, KEYS[1], ARGV[2], 1); " +
                      "redis.call(&#39;pexpire&#39;, KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
                  "if (redis.call(&#39;hexists&#39;, KEYS[1], ARGV[2]) == 1) then " +
                      "redis.call(&#39;hincrby&#39;, KEYS[1], ARGV[2], 1); " +
                      "redis.call(&#39;pexpire&#39;, KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
                  "return redis.call(&#39;pttl&#39;, KEYS[1]);",
                    Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}

没错，加锁最终执行的就是这段lua 脚本语言。

if (redis.call('exists', KEYS[1]) == 0) then 
    redis.call('hset', KEYS[1], ARGV[2], 1); 
    redis.call('pexpire', KEYS[1], ARGV[1]); 
    return nil; 
end;

脚本的主要逻辑为：

• exists 判断 key 是否存在
• 当判断不存在则设置 key
• 然后给设置的key追加过期时间

这样来看其实和我们前面案例中的实现方法好像没什么区别，但实际上并不是。

这段lua脚本命令在Redis中执行时，会被当成一条命令来执行，能够保证原子性，故要不都成功，要不都失败。

我们在源码中看到Redssion的许多方法实现中很多都用到了lua脚本，这样能够极大的保证命令执行的原子性。

下面是Redisson锁自动“续命”源码：

private void scheduleExpirationRenewal(final long threadId) {
    if (expirationRenewalMap.containsKey(getEntryName())) {
        return;
    }

    Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
        @Override
        public void run(Timeout timeout) throws Exception {

            RFuture<Boolean> future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                                                                     "if (redis.call(&#39;hexists&#39;, KEYS[1], ARGV[2]) == 1) then " +
                                                                     "redis.call(&#39;pexpire&#39;, KEYS[1], ARGV[1]); " +
                                                                     "return 1; " +
                                                                     "end; " +
                                                                     "return 0;",
                                                                     Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));

            future.addListener(new FutureListener<Boolean>() {
                @Override
                public void operationComplete(Future<Boolean> future) throws Exception {
                    expirationRenewalMap.remove(getEntryName());
                    if (!future.isSuccess()) {
                        log.error("Can&#39;t update lock " + getName() + " expiration", future.cause());
                        return;
                    }

                    if (future.getNow()) {
                        // reschedule itself
                        scheduleExpirationRenewal(threadId);
                    }
                }
            });
        }
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);

    if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) {
        task.cancel();
    }
}

这段代码是在加锁后开启一个守护线程进行监听。Redisson超时时间默认设置30s，线程每10s调用一次判断锁还是否存在，如果存在则延长锁的超时时间。

现在，我们再回过头来看看案例5中的加锁代码与原理图，其实完善到这种程度已经可以满足很多公司的使用了，并且很多公司也确实是这样用的。但我们再思考下是否还存在问题呢？例如以下场景：

• 우리 모두 알고 있듯이Redis 실제 배포 및 사용에서는 클러스터에 배포됩니다. 동시성이 높은 시나리오에서는 잠급니다. 마스터 노드에 키를 쓴 후 마스터가 슬레이브 노드에 동기화되기 전에 충돌이 발생합니다. 원본 슬레이브 노드가 선출된 후 새로운 마스터 노드가 됩니다. 이때 잠금 실패 문제가 발생할 수 있습니다.
Redis在实际部署使用时都是集群部署的，那在高并发场景下我们加锁，当把key写入到master节点后，master还未同步到slave节点时master宕机了，原有的slave节点经过选举变为了新的master节点，此时可能就会出现锁失效问题。

通过分布式锁的实现机制我们知道，高并发场景下只有加锁成功的请求可以继续处理业务逻辑。那就出现了大伙都来加锁，但有且仅有一个加锁成功了，剩余的都在等待。其实分布式锁与高并发在语义上就是相违背的，我们的请求虽然都是并发，但Redis帮我们把请求进行了排队执行，也就是把我们的并行转为了串行。串行执行的代码肯定不存在并发问题了，但是程序的性能肯定也会因此受到影响。

针对这些问题，我们再次思考解决方案

• 在思考解决方案时我们首先想到CAP原则（一致性、可用性、分区容错性），那么现在的Redis就是满足AP(可用性、分区容错性)，如果想要解决该问题我们就需要寻找满足CP(一致性、分区容错性)的分布式系统。首先想到的就是Zookeeper，Zookeeper🎜분산 잠금 구현 메커니즘을 통해 동시성이 높은 시나리오에서는 성공적으로 잠긴 요청만 비즈니스 로직을 계속 처리할 수 있다는 것을 알고 있습니다. 그런 다음 모두가 잠그기 위해 왔지만 단 한 번의 잠금만 성공하고 나머지는 기다리고 있습니다. 실제로 분산 잠금과 높은 동시성은 의미상 모순됩니다. 요청은 모두 동시이지만 Redis는 실행 요청을 대기열에 추가하는 데 도움이 됩니다. 이는 병렬 처리를 직렬화로 변환하는 것을 의미합니다. 직렬로 실행되는 코드에서는 동시성 문제가 전혀 발생하지 않지만 프로그램 성능에는 확실히 영향을 미칩니다.

이러한 문제에 대응하여 우리는 해결책을 다시 생각합니다

🎜🎜🎜솔루션을 생각할 때 우리는 먼저 다음을 생각합니다. CAP 원칙(일관성, 가용성, 파티션 허용 오차), 현재 Redis는 AP(가용성, 파티션 허용오차)를 만족합니다. 이 문제를 해결하려면 CP (일관성, 파티션 내결함성) 분산 시스템. 가장 먼저 떠오르는 것은 사육사 , Zookeeper의 클러스터 간 데이터 동기화 메커니즘은 마스터 노드가 데이터를 수신하면 성공 피드백을 클라이언트에 즉시 반환하지 않고 먼저 하위 노드와 데이터를 동기화한 다음 노드의 절반 이상이 동기화를 완료한 후에만 클라이언트에 알립니다.그리고 Zookeeper的Zab协议（Zookeeper原子广播）重新选举的主节点一定是已经同步成功的。

那么问题来了，Redisson与Zookeeper分布式锁我们如何选择呢？答案是如果并发量没有那么高，可以用Zookeeper来做分布式锁，但是它的并发能力远远不如Redis。如果你对并发要求比较高的话，那就用Redis，偶尔出现的主从架构锁失效的问题其实是可以容忍的。

关于第二个提升性能的问题，我们可以参考ConcurrentHashMap的锁分段技术的思想，例如我们代码的库存量当前为1000，那我们可以分为10段，每段100，然后对每段分别加锁，这样就可以同时执行10个请求的加锁与处理，当然有要求的同学还可以继续细分。但其实Redis的Qps已经达到10W+에 따르면 마스터 노드가 다운되면 특별히 높은 동시성이 없는 시나리오에서는 완전히 충분합니다.

위 내용은 분산잠금 : 진입부터 매몰까지 5건의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!