redis中的分布式鎖有哪些特點(diǎn)

本篇內(nèi)容主要講解“redis中的分布式鎖有哪些特點(diǎn)”，感興趣的朋友不妨來(lái)看看。本文介紹的方法操作簡(jiǎn)單快捷，實(shí)用性強(qiáng)。下面就讓小編來(lái)帶大家學(xué)習(xí)“redis中的分布式鎖有哪些特點(diǎn)”吧!

分布式鎖的特點(diǎn)

• 1.獨(dú)占性

不論在任何情況下都只能有一個(gè)線程持有鎖。

• 2.高可用

redis集群環(huán)境不能因?yàn)槟骋粋€(gè)節(jié)點(diǎn)宕機(jī)而出現(xiàn)獲取鎖或釋放鎖失敗。

• 3.防死鎖

必須有超時(shí)控制機(jī)制或者撤銷操作。

• 4.不亂搶

自己加鎖，自己釋放。不能釋放別人加的鎖。

• 5.重入性

同一線程可以多次加鎖。

redis單機(jī)怎么實(shí)現(xiàn)

一般情況下都是使用setnx+lua腳本實(shí)現(xiàn)。

直接貼代碼

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 單機(jī)鎖
 *
 * @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,解鎖時(shí) 用來(lái)判斷當(dāng)前鎖是否是自己加的
        String value = IdUtil.fastSimpleUUID();
        //加鎖 十秒鐘過(guò)期 防死鎖
        Boolean flag = redisTemplate.opsForValue().setIfAbsent(key, value, 10, TimeUnit.SECONDS);
        if (!flag) {
            return "加鎖失敗";
        }
        try {
            String productKey = "good123";
            //獲取商品庫(kù)存
            Integer stock = (Integer) redisTemplate.opsForValue().get(productKey);
            if (stock == null) {
                //模擬錄入數(shù)據(jù)， 實(shí)際應(yīng)該加載時(shí)從數(shù)據(jù)庫(kù)讀取
                redisTemplate.opsForValue().set(productKey, 100);
                stock = 100;
            }
            if (stock <= 0) {
                return "賣完了，下次早點(diǎn)來(lái)吧";
            }
            //扣減庫(kù)存, 模擬隨機(jī)賣出數(shù)量
            int randomInt = RandomUtil.randomInt(1, 10);
            redisTemplate.opsForValue().decrement(productKey, randomInt);
            // 修改db,可以丟到隊(duì)列里慢慢處理
            return "成功賣出" + randomInt + "個(gè)，庫(kù)存剩余" + redisTemplate.opsForValue().get(productKey) + "個(gè)";
        } finally {

//            //這種方法會(huì)存在刪除別人加的鎖的可能
//            redisTemplate.delete(key);

//            if(value.equals(redisTemplate.opsForValue().get(key))){
//                //因?yàn)閕f條件的判斷和 delete不是原子性的，
//                //if條件判斷成功后，恰好鎖到期自己解鎖
//                //此時(shí)別的線程如果持有鎖了，就會(huì)把別人的鎖刪除掉
//                redisTemplate.delete(key);
//            }

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


}

進(jìn)行單元測(cè)試，模擬一百個(gè)線程同時(shí)進(jìn)行秒殺

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

只有三個(gè)線程搶到了鎖

成功賣出5個(gè)，庫(kù)存剩余95個(gè)
成功賣出8個(gè)，庫(kù)存剩余87個(gè)
成功賣出7個(gè)，庫(kù)存剩余80個(gè)

redis鎖有什么問(wèn)題？

總的來(lái)說(shuō)有兩個(gè)：

• 1.無(wú)法重入。

• 2.我們?yōu)榱朔乐顾梨i，加鎖時(shí)都會(huì)加上過(guò)期時(shí)間，這個(gè)時(shí)間大部分情況下都是根據(jù)經(jīng)驗(yàn)對(duì)現(xiàn)有業(yè)務(wù)評(píng)估得出來(lái)的，但是萬(wàn)一程序阻塞或者異常，導(dǎo)致執(zhí)行了很長(zhǎng)時(shí)間，鎖過(guò)期就會(huì)自動(dòng)釋放了。此時(shí)如果別的線程拿到鎖，執(zhí)行邏輯，就有可能出現(xiàn)問(wèn)題。

那么這兩個(gè)問(wèn)題有沒(méi)有辦法解決呢？有，接下來(lái)我們就來(lái)講講Redisson

Redisson實(shí)現(xiàn)分布式鎖

Redisson是什么？

Redisson是一個(gè)在Redis的基礎(chǔ)上實(shí)現(xiàn)的Java駐內(nèi)存數(shù)據(jù)網(wǎng)格（In-Memory Data Grid）。它不僅提供了一系列的分布式的Java常用對(duì)象，還提供了許多分布式服務(wù)。其中包括(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的最簡(jiǎn)單和最便捷的方法。Redisson的宗旨是促進(jìn)使用者對(duì)Redis的關(guān)注分離（Separation of Concern），從而讓使用者能夠?qū)⒕Ω械胤旁谔幚順I(yè)務(wù)邏輯上。

springboot集成Redisson

集成很簡(jiǎn)單，只需兩步

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

使用也很簡(jiǎn)單，只需要注入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("業(yè)務(wù)異常", e);
        } finally {
            rLock.unlock();
        }

    }
    
}

可能不了解redisson的小伙伴會(huì)不禁發(fā)出疑問(wèn)。
what?加鎖時(shí)不需要加過(guò)期時(shí)間嗎？這樣會(huì)不會(huì)導(dǎo)致死鎖啊。解鎖不需要判斷是不是自己持有嗎？
哈哈，別著急，我們接下來(lái)一步步揭開redisson的面紗。

Redisson lock()源碼跟蹤

我們來(lái)一步步跟著lock()方法看下源碼(本地redisson版本為3.20.0)

//RedissonLock.class

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

查看lock(-1, null, false);方法

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
        //獲取當(dāng)前線程id
        long threadId = Thread.currentThread().getId();
        //加鎖代碼塊， 返回鎖的失效時(shí)間
        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));
    }

我們看下它是怎么上鎖的，也就是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，所以走這里   
        //過(guò)期時(shí)間internalLockLeaseTime初始化的時(shí)候賦值commandExecutor.getServiceManager().getCfg().getLockWatchdogTimeout();
        //跟進(jìn)去源碼發(fā)現(xiàn)默認(rèn)值是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);
    //加鎖成功，開啟子線程進(jìn)行續(xù)約
    CompletionStage<Long> f = ttlRemainingFuture.thenApply(ttlRemaining -> {
        // lock acquired
        if (ttlRemaining == null) {
            if (leaseTime > 0) {
                //如果指定了過(guò)期時(shí)間，則不續(xù)約
                internalLockLeaseTime = unit.toMillis(leaseTime);
            } else {
                //沒(méi)指定過(guò)期時(shí)間，或者小于0，在這里實(shí)現(xiàn)鎖自動(dòng)續(xù)約
                scheduleExpirationRenewal(threadId);
            }
        }
        return ttlRemaining;
    });
    return new CompletableFutureWrapper<>(f);
}

上面代碼里面包含加鎖和鎖續(xù)約的邏輯，我們先來(lái)看看加鎖的代碼

<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
            "if ((redis.call('exists', KEYS[1]) == 0) " +
                        "or (redis.call('hexists', KEYS[1], ARGV[2]) == 1)) then " +
                    "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                    "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                    "return nil; " +
                "end; " +
                "return redis.call('pttl', KEYS[1]);",
            Collections.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
}

這里就看的很明白了吧，redisson使用了lua腳本來(lái)保證了命令的原子性。
redis.call('hexists', KEYS[1], ARGV[2])   查看 key value 是否存在。

Redis Hexists 命令用于查看哈希表的指定字段是否存在。
如果哈希表含有給定字段，返回 1 。 如果哈希表不含有給定字段，或 key 不存在，返回 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>

當(dāng)key不存在，或者已經(jīng)含有給定字段(也就是已經(jīng)加過(guò)鎖了，這里是為了實(shí)現(xiàn)重入性)，直接對(duì)字段的值+1
這個(gè)字段的值，也就是ARGV[2]， 取得是getLockName(threadId)方法，我們?cè)倏纯催@個(gè)字段的值是什么

    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默認(rèn)值
    private final String id = UUID.randomUUID().toString();

這里就明白了，字段名稱是 uuid + : + threadId

接下來(lái)我們看看鎖續(xù)約的代碼scheduleExpirationRenewal(threadId);

protected void scheduleExpirationRenewal(long threadId) {
    ExpirationEntry entry = new ExpirationEntry();
    //判斷該實(shí)例是否加過(guò)鎖
    ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
    if (oldEntry != null) {
        //重入次數(shù)+1
        oldEntry.addThreadId(threadId);
    } else {
        //第一次加鎖
        entry.addThreadId(threadId);
        try {
            //鎖續(xù)約核心代碼
            renewExpiration();
        } finally {
            if (Thread.currentThread().isInterrupted()) {
                //如果線程異常終止，則關(guān)閉鎖續(xù)約線程
                cancelExpirationRenewal(threadId);
            }
        }
    }
}

我們看看renewExpiration()方法

private void renewExpiration() {
    ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
    if (ee == null) {
        return;
    }
    //新建一個(gè)線程執(zhí)行
    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;
            }
            //設(shè)置鎖過(guò)期時(shí)間為30秒
            CompletionStage<Boolean> future = renewExpirationAsync(threadId);
            future.whenComplete((res, e) -> {
                if (e != null) {
                    log.error("Can't update lock {} expiration", getRawName(), e);
                    EXPIRATION_RENEWAL_MAP.remove(getEntryName());
                    return;
                }
                //檢查鎖是還否存在
                if (res) {
                    // reschedule itself 10后調(diào)用自己
                    renewExpiration();
                } else {
                    //關(guān)閉續(xù)約
                    cancelExpirationRenewal(null);
                }
            });
        }
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
    //注意上行代碼internalLockLeaseTime / 3，
    //internalLockLeaseTime默認(rèn)30s，那么也就是10s檢查一次
    ee.setTimeout(task);
}

//設(shè)置鎖過(guò)期時(shí)間為internalLockLeaseTime  也就是30s  lua腳本保證原子性
protected CompletionStage<Boolean> renewExpirationAsync(long threadId) {
    return evalWriteAsync(getRawName(), 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(getRawName()),
            internalLockLeaseTime, getLockName(threadId));
}

OK，分析到這里我們已經(jīng)知道了，lock()，方法會(huì)默認(rèn)加30秒過(guò)期時(shí)間，并且開啟一個(gè)新線程，每隔10秒檢查一下，鎖是否釋放，如果沒(méi)釋放，就將鎖過(guò)期時(shí)間設(shè)置為30秒，如果鎖已經(jīng)釋放，那么就將這個(gè)新線程也關(guān)掉。

我們寫個(gè)測(cè)試類看看

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

查看鎖的過(guò)期時(shí)間，及是否續(xù)約

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>

我們?cè)俑母拇a，看看是否可重入和字段名稱是否和我們預(yù)期一致

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();
        //加了三次鎖，此時(shí)重入次數(shù)為3
        Thread.sleep(3000);
        //解鎖一次，此時(shí)重入次數(shù)變?yōu)?
        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>

我們加鎖了三次，重入次數(shù)是3，字段值也是 uuid+:+threadId,和我們預(yù)期結(jié)果是一致的。

Redlock算法

redisson是基于Redlock算法實(shí)現(xiàn)的，那么什么是Redlock算法呢？

假設(shè)當(dāng)前集群有5個(gè)節(jié)點(diǎn)，那么運(yùn)行redlock算法的客戶端會(huì)一次執(zhí)行下面步驟

• 1.客戶端記錄當(dāng)前系統(tǒng)時(shí)間，以毫秒為單位

• 2.依次嘗試從5個(gè)redis實(shí)例中，使用相同key獲取鎖
  當(dāng)redis請(qǐng)求獲取鎖時(shí)，客戶端會(huì)設(shè)置一個(gè)網(wǎng)絡(luò)連接和響應(yīng)超時(shí)時(shí)間，避免因?yàn)榫W(wǎng)絡(luò)故障等原因?qū)е伦枞?/p>

• 3.客戶端使用當(dāng)前時(shí)間減去開始獲取鎖時(shí)間（步驟1的時(shí)間），得到獲取鎖消耗的時(shí)間
  只有當(dāng)半數(shù)以上redis節(jié)點(diǎn)加鎖成功，并且加鎖消耗的時(shí)間要小于鎖失效時(shí)間，才算鎖獲取成功

• 4.如果獲取到了鎖，key的真正有效時(shí)間等于鎖失效時(shí)間 減去 獲取鎖消耗的時(shí)間

• 5.如果獲取鎖失敗，所有的redis實(shí)例都會(huì)進(jìn)行解鎖
  防止因?yàn)榉?wù)端響應(yīng)消息丟失，但是實(shí)際數(shù)據(jù)又添加成功導(dǎo)致數(shù)據(jù)不一致問(wèn)題

這里有下面幾個(gè)點(diǎn)需要注意：

• 1.我們都知道單機(jī)的redis是cp的，但是集群情況下redis是ap的，所以運(yùn)行Redisson的節(jié)點(diǎn)必須是主節(jié)點(diǎn)，不能有從節(jié)點(diǎn)，防止主節(jié)點(diǎn)加鎖成功未同步從節(jié)點(diǎn)就宕機(jī)，而客戶端卻收到加鎖成功，導(dǎo)致數(shù)據(jù)不一致問(wèn)題。

• 2.為了提高redis節(jié)點(diǎn)宕機(jī)的容錯(cuò)率，可以使用公式2N（n指宕機(jī)數(shù)量）+1，假設(shè)宕機(jī)一臺(tái)，Redisson還要繼續(xù)運(yùn)行，那么至少要部署2*1+1=3臺(tái)主節(jié)點(diǎn)。

到此，相信大家對(duì)“redis中的分布式鎖有哪些特點(diǎn)”有了更深的了解，不妨來(lái)實(shí)際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！