Redis의 분산 잠금에 대해 자세히 이야기해 보겠습니다.

우리 모두는 분산 환경에서 분산 잠금을 사용해야 한다는 것을 알고 있습니다. 그렇다면 분산 잠금에는 어떤 특성이 필요합니까? 독립형 Redis를 잠그는 방법은 무엇입니까? Redis 클러스터 잠금의 함정은 무엇입니까? 걱정하지 마세요. Redis 분산 잠금의 베일을 단계별로 풀어보겠습니다.

분산 잠금의 특징

• 1. 배타성

어떤 상황에서도 하나의 스레드만 잠금을 유지할 수 있습니다.

• 2. 고가용성

Redis 클러스터 환경은 노드가 다운되어도 잠금을 획득하거나 해제하지 못할 수 없습니다. [관련 권장 사항: Redis 비디오 튜토리얼]

• 3. Anti-deadlock

에는 시간 초과 제어 메커니즘이나 실행 취소 작업이 있어야 합니다.

• 4. 아무것도 잡지 마세요

직접 잠그고 풀어보세요. 다른 사람이 추가한 잠금은 해제할 수 없습니다.

• 5. 재진입

동일한 스레드가 여러 번 잠길 수 있습니다.

단일 머신에서 Redis를 구현하는 방법

일반적으로 setnx+lua 스크립트를 사용하여 구현합니다.

코드를 직접 게시

package com.fandf.test.redis;

import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.RandomUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Collections;
import java.util.concurrent.TimeUnit;

/**
 * redis 单机锁
 *
 * @author fandongfeng
 * @date 2023/3/29 06:52
 */
@Slf4j
@Service
public class RedisLock {

    @Resource
    RedisTemplate<String, Object> redisTemplate;

    private static final String SELL_LOCK = "kill:";

    /**
     * 模拟秒杀
     *
     * @return 是否成功
     */
    public String kill() {

        String productId = "123";
        String key = SELL_LOCK + productId;
        //锁value,解锁时 用来判断当前锁是否是自己加的
        String value = IdUtil.fastSimpleUUID();
        //加锁 十秒钟过期 防死锁
        Boolean flag = redisTemplate.opsForValue().setIfAbsent(key, value, 10, TimeUnit.SECONDS);
        if (!flag) {
            return "加锁失败";
        }
        try {
            String productKey = "good123";
            //获取商品库存
            Integer stock = (Integer) redisTemplate.opsForValue().get(productKey);
            if (stock == null) {
                //模拟录入数据， 实际应该加载时从数据库读取
                redisTemplate.opsForValue().set(productKey, 100);
                stock = 100;
            }
            if (stock <= 0) {
                return "卖完了，下次早点来吧";
            }
            //扣减库存, 模拟随机卖出数量
            int randomInt = RandomUtil.randomInt(1, 10);
            redisTemplate.opsForValue().decrement(productKey, randomInt);
            // 修改db,可以丢到队列里慢慢处理
            return "成功卖出" + randomInt + "个，库存剩余" + redisTemplate.opsForValue().get(productKey) + "个";
        } finally {

//            //这种方法会存在删除别人加的锁的可能
//            redisTemplate.delete(key);

//            if(value.equals(redisTemplate.opsForValue().get(key))){
//                //因为if条件的判断和 delete不是原子性的，
//                //if条件判断成功后，恰好锁到期自己解锁
//                //此时别的线程如果持有锁了，就会把别人的锁删除掉
//                redisTemplate.delete(key);
//            }

            //使用lua脚本保证判断和删除的原子性
            String luaScript =
                    "if (redis.call(&#39;get&#39;,KEYS[1]) == ARGV[1]) then " +
                            "return redis.call(&#39;del&#39;,KEYS[1]) " +
                            "else " +
                            "return 0 " +
                            "end";
            redisTemplate.execute(new DefaultRedisScript<>(luaScript, Boolean.class), Collections.singletonList(key), value);
        }
    }


}

단위 테스트를 수행하고 수백 개의 스레드를 시뮬레이션하여 동시에 플래시 종료를 수행

package com.fandf.test.redis;

import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.parallel.Execution;
import org.springframework.boot.test.context.SpringBootTest;

import javax.annotation.Resource;

import static org.junit.jupiter.api.parallel.ExecutionMode.CONCURRENT;

/**
 * @Description:
 * @author: fandongfeng
 * @date: 2023-3-24 16:45
 */
@SpringBootTest
class SignServiceTest {

  
    @Resource
    RedisLock redisLock;


    @RepeatedTest(100)
    @Execution(CONCURRENT)
    public void redisLock() {
        String result = redisLock.kill();
        if("加锁失败".equals(result)) {

        }else {
            System.out.println(result);
        }
    }
}

3개의 스레드만 잠금을 잡았습니다

成功卖出5个，库存剩余95个
成功卖出8个，库存剩余87个
成功卖出7个，库存剩余80个

redis 잠금에 무슨 문제가 있나요?

일반적으로 2가지가 있습니다.

• 1.
• 2. 교착 상태를 방지하기 위해 잠금 시 만료 시간을 추가할 예정입니다. 대부분의 경우 기존 비즈니스에 대한 경험 평가를 기준으로 적용되지만, 프로그램 차단이나 예외가 발생할 경우 실행 시간이 오래 걸립니다. .오랜 시간이 지나면 잠금이 만료되면 자동으로 해제됩니다. 이때 다른 스레드가 Lock을 받아 로직을 실행하게 되면 문제가 발생할 수 있다.

그렇다면 이 두 가지 문제를 해결할 수 있는 방법은 없을까요? 네, Redisson에 대해 이야기해 봅시다

분산 잠금을 구현하는 Redisson

Redisson이 무엇인가요?

Redisson은 Redis를 기반으로 구현된 Java 인메모리 데이터 그리드(In-Memory Data Grid)입니다. 일련의 분산 공통 ​​Java 개체를 제공할 뿐만 아니라 많은 분산 서비스도 제공합니다. 여기에는 다음이 포함됩니다(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) Redisson은 Redis를 사용하는 가장 쉽고 편리한 방법을 제공합니다. Redisson의 목적은 Redis에서 사용자의 우려 사항 분리(Separation of Concern)를 촉진하여 사용자가 비즈니스 로직 처리에 더 집중할 수 있도록 하는 것입니다.

springboot는 Redisson을 통합합니다

통합은 두 단계만으로 매우 간단합니다

1. pom은 종속성을 도입합니다

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
</dependency>

2. application.yml은 redis 구성을 추가합니다

spring:
  application:
    name: test
  redis:
    host: 127.0.0.1
    port: 6379

사용하기도 매우 간단합니다. RedissonClient

package com.fandf.test.redis;

import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @author fandongfeng
 */
@Component
@Slf4j
public class RedissonTest {

    @Resource
    RedissonClient redissonClient;

    public void test() {
        RLock rLock = redissonClient.getLock("anyKey");
        //rLock.lock(10, TimeUnit.SECONDS);
        rLock.lock();
        try {
            // do something
        } catch (Exception e) {
            log.error("业务异常", e);
        } finally {
            rLock.unlock();
        }

    }
    
}

를 주입하기만 하면 됩니다. 레디송을 모르는 친구들은 질문을 하지 않을 수 없습니다.
잠금할 때 만료 시간을 추가해야 하는 거 아닌가요? 이로 인해 교착 상태가 발생합니까? 당신이 직접 가지고 있는지 여부를 잠금 해제하려면 판단이 필요하지 않습니까?
하하, 걱정하지 마세요. 리디송은 차근차근 공개하겠습니다.

Redisson lock() 소스 코드 추적

lock() 메서드를 단계별로 따라 소스 코드를 살펴보겠습니다. (로컬 Redisson 버전은 3.20.0입니다.)

//RedissonLock.class

@Override
public void lock() {
    try {
        lock(-1, null, false);
    } catch (InterruptedException e) {
        throw new IllegalStateException();
    }
}

잠금 보기(-1, null, false) ); method

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
        //获取当前线程id
        long threadId = Thread.currentThread().getId();
        //加锁代码块， 返回锁的失效时间
        Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
            return;
        }

        CompletableFuture<RedissonLockEntry> future = subscribe(threadId);
        pubSub.timeout(future);
        RedissonLockEntry entry;
        if (interruptibly) {
            entry = commandExecutor.getInterrupted(future);
        } else {
            entry = commandExecutor.get(future);
        }

        try {
            while (true) {
                ttl = tryAcquire(-1, leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    break;
                }

                // waiting for message
                if (ttl >= 0) {
                    try {
                        entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        if (interruptibly) {
                            throw e;
                        }
                        entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    }
                } else {
                    if (interruptibly) {
                        entry.getLatch().acquire();
                    } else {
                        entry.getLatch().acquireUninterruptibly();
                    }
                }
            }
        } finally {
            unsubscribe(entry, threadId);
        }
//        get(lockAsync(leaseTime, unit));
    }

us tryAcquire 메소드인 잠김 방식을 살펴보겠습니다

private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
    //真假加锁方法 tryAcquireAsync
    return get(tryAcquireAsync(waitTime, leaseTime, unit, threadId));
}

public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {
    super(commandExecutor, name);
    this.commandExecutor = commandExecutor;
    this.internalLockLeaseTime = commandExecutor.getServiceManager().getCfg().getLockWatchdogTimeout();
    this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub();
}

private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
    RFuture<Long> ttlRemainingFuture;
    if (leaseTime > 0) {
        ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    } else {
        //waitTime和leaseTime都是-1，所以走这里   
        //过期时间internalLockLeaseTime初始化的时候赋值commandExecutor.getServiceManager().getCfg().getLockWatchdogTimeout();
        //跟进去源码发现默认值是30秒， private long lockWatchdogTimeout = 30 * 1000;
        ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
                TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
    }
    CompletionStage<Long> s = handleNoSync(threadId, ttlRemainingFuture);
    ttlRemainingFuture = new CompletableFutureWrapper<>(s);
    //加锁成功，开启子线程进行续约
    CompletionStage<Long> f = ttlRemainingFuture.thenApply(ttlRemaining -> {
        // lock acquired
        if (ttlRemaining == null) {
            if (leaseTime > 0) {
                //如果指定了过期时间，则不续约
                internalLockLeaseTime = unit.toMillis(leaseTime);
            } else {
                //没指定过期时间，或者小于0，在这里实现锁自动续约
                scheduleExpirationRenewal(threadId);
            }
        }
        return ttlRemaining;
    });
    return new CompletableFutureWrapper<>(f);
}

위 코드에는 잠금 및 잠금 갱신 논리가 포함되어 있습니다. 먼저 잠금 코드를 살펴보겠습니다

<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
            "if ((redis.call(&#39;exists&#39;, KEYS[1]) == 0) " +
                        "or (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.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
}

Redisson은 Lua 스크립트를 사용하여 명령의 원자성을 보장합니다.
redis.call('hexists', KEYS[1], ARGV[2]) 키 값이 존재하는지 확인합니다.

Redis Hexists 명령은 해시 테이블의 지정된 필드가 존재하는지 확인하는 데 사용됩니다.
해시 테이블에 지정된 필드가 포함되어 있으면 1을 반환합니다. 해시 테이블에 지정된 필드가 없거나 키가 없으면 0이 반환됩니다.

127.0.0.1:6379> hexists 123 uuid
(integer) 0
127.0.0.1:6379> hincrby 123 uuid 1
(integer) 1
127.0.0.1:6379> hincrby 123 uuid 1
(integer) 2
127.0.0.1:6379> hincrby 123 uuid 1
(integer) 3
127.0.0.1:6379> hexists 123 uuid
(integer) 1
127.0.0.1:6379> hgetall 123
1) "uuid"
2) "3"
127.0.0.1:6379>

키가 존재하지 않거나 특정 필드가 이미 포함되어 있는 경우(즉, 잠긴 경우 재진입을 달성하기 위한 것임) 필드 값에 직접 1을 추가하세요.
이 필드의 값은 다음과 같습니다. getLockName(threadId) 메소드로 얻은 ARGV[2 ], 이 필드의 값을 살펴보겠습니다

    protected String getLockName(long threadId) {
        return id + ":" + threadId;
    }

    public RedissonBaseLock(CommandAsyncExecutor commandExecutor, String name) {
        super(commandExecutor, name);
        this.commandExecutor = commandExecutor;
        this.id = commandExecutor.getServiceManager().getId();
        this.internalLockLeaseTime = commandExecutor.getServiceManager().getCfg().getLockWatchdogTimeout();
        this.entryName = id + ":" + name;
    }

    //commandExecutor.getServiceManager() 的id默认值
    private final String id = UUID.randomUUID().toString();

여기서 이해하게 될 것입니다. 필드 이름은 uuid + : + threadId

다음으로 잠금 갱신을 살펴보겠습니다. code ScheduleExpirationRenewal(threadId);

protected void scheduleExpirationRenewal(long threadId) {
    ExpirationEntry entry = new ExpirationEntry();
    //判断该实例是否加过锁
    ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
    if (oldEntry != null) {
        //重入次数+1
        oldEntry.addThreadId(threadId);
    } else {
        //第一次加锁
        entry.addThreadId(threadId);
        try {
            //锁续约核心代码
            renewExpiration();
        } finally {
            if (Thread.currentThread().isInterrupted()) {
                //如果线程异常终止，则关闭锁续约线程
                cancelExpirationRenewal(threadId);
            }
        }
    }
}

renewExpiration() 메소드를 살펴보겠습니다

private void renewExpiration() {
    ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
    if (ee == null) {
        return;
    }
    //新建一个线程执行
    Timeout task = commandExecutor.getServiceManager().newTimeout(new TimerTask() {
        @Override
        public void run(Timeout timeout) throws Exception {
            ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
            if (ent == null) {
                return;
            }
            Long threadId = ent.getFirstThreadId();
            if (threadId == null) {
                return;
            }
            //设置锁过期时间为30秒
            CompletionStage<Boolean> future = renewExpirationAsync(threadId);
            future.whenComplete((res, e) -> {
                if (e != null) {
                    log.error("Can&#39;t update lock {} expiration", getRawName(), e);
                    EXPIRATION_RENEWAL_MAP.remove(getEntryName());
                    return;
                }
                //检查锁是还否存在
                if (res) {
                    // reschedule itself 10后调用自己
                    renewExpiration();
                } else {
                    //关闭续约
                    cancelExpirationRenewal(null);
                }
            });
        }
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
    //注意上行代码internalLockLeaseTime / 3，
    //internalLockLeaseTime默认30s，那么也就是10s检查一次
    ee.setTimeout(task);
}

//设置锁过期时间为internalLockLeaseTime  也就是30s  lua脚本保证原子性
protected CompletionStage<Boolean> renewExpirationAsync(long threadId) {
    return evalWriteAsync(getRawName(), 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.singletonList(getRawName()),
            internalLockLeaseTime, getLockName(threadId));
}

OK，分析到这里我们已经知道了，lock()，方法会默认加30秒过期时间，并且开启一个新线程，每隔10秒检查一下，锁是否释放，如果没释放，就将锁过期时间设置为30秒，如果锁已经释放，那么就将这个新线程也关掉。

我们写个测试类看看

package com.fandf.test.redis;

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;

/**
 * @Description:
 * @author: fandongfeng
 * @date: 2023-3-2416:45
 */
@SpringBootTest
class RedissonTest {

    @Resource
    private RedissonClient redisson;


    @Test
    public void watchDog() throws InterruptedException {
        RLock lock = redisson.getLock("123");
        lock.lock();
        Thread.sleep(1000000);
    }
}

查看锁的过期时间，及是否续约

127.0.0.1:6379> keys *
1) "123"
127.0.0.1:6379> ttl 123
(integer) 30
127.0.0.1:6379> ttl 123
(integer) 26
127.0.0.1:6379> ttl 123
(integer) 24
127.0.0.1:6379> ttl 123
(integer) 22
127.0.0.1:6379> ttl 123
(integer) 21
127.0.0.1:6379> ttl 123
(integer) 20
127.0.0.1:6379> ttl 123
(integer) 30
127.0.0.1:6379> ttl 123
(integer) 28
127.0.0.1:6379>

我们再改改代码，看看是否可重入和字段名称是否和我们预期一致

package com.fandf.test.redis;

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;

/**
 * @Description:
 * @author: fandongfeng
 * @date: 2023-3-24 16:45
 */
@SpringBootTest
class RedissonTest {

    @Resource
    private RedissonClient redisson;


    @Test
    public void watchDog() throws InterruptedException {
        RLock lock = redisson.getLock("123");
        lock.lock();
        lock.lock();
        lock.lock();
        //加了三次锁，此时重入次数为3
        Thread.sleep(3000);
        //解锁一次，此时重入次数变为3
        lock.unlock();
        Thread.sleep(1000000);
    }
}

127.0.0.1:6379> keys *
1) "123"
127.0.0.1:6379>
127.0.0.1:6379> ttl 123
(integer) 24
127.0.0.1:6379> hgetall 123
1) "df7f4c71-b57b-455f-acee-936ad8475e01:12"
2) "3"
127.0.0.1:6379>
127.0.0.1:6379> hgetall 123
1) "df7f4c71-b57b-455f-acee-936ad8475e01:12"
2) "2"
127.0.0.1:6379>

我们加锁了三次，重入次数是3，字段值也是 uuid+:+threadId,和我们预期结果是一致的。

Redlock算法

redisson是基于Redlock算法实现的，那么什么是Redlock算法呢？

假设当前集群有5个节点，那么运行redlock算法的客户端会一次执行下面步骤

• 1.客户端记录当前系统时间，以毫秒为单位
• 2.依次尝试从5个redis实例中，使用相同key获取锁
  当redis请求获取锁时，客户端会设置一个网络连接和响应超时时间，避免因为网络故障等原因导致阻塞。
• 3.客户端使用当前时间减去开始获取锁时间（步骤1的时间），得到获取锁消耗的时间
  只有当半数以上redis节点加锁成功，并且加锁消耗的时间要小于锁失效时间，才算锁获取成功
• 4.如果获取到了锁，key的真正有效时间等于锁失效时间 减去 获取锁消耗的时间
• 5.如果获取锁失败，所有的redis实例都会进行解锁
  防止因为服务端响应消息丢失，但是实际数据又添加成功导致数据不一致问题

这里有下面几个点需要注意：

• 1.我们都知道单机的redis是cp的，但是集群情况下redis是ap的，所以运行Redisson的节点必须是主节点，不能有从节点，防止主节点加锁成功未同步从节点就宕机，而客户端却收到加锁成功，导致数据不一致问题。
• 2.为了提高redis节点宕机的容错率，可以使用公式2N（n指宕机数量）+1，假设宕机一台，Redisson还要继续运行，那么至少要部署2*1+1=3台主节点。

更多编程相关知识，请访问：编程视频！！

위 내용은 Redis의 분산 잠금에 대해 자세히 이야기해 보겠습니다.의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!