巧用Redis實(shí)現(xiàn)分布式鎖詳細(xì)介紹

更新時(shí)間：2021年12月23日 10:20:30 作者：Monster_起飛

大家好，本篇文章主要講的是巧用Redis實(shí)現(xiàn)分布式鎖詳細(xì)介紹，感興趣的同學(xué)趕快來(lái)看一看吧，對(duì)你有幫助的話記得收藏一下，方便下次瀏覽

前言

無(wú)論是synchronized還是Lock，都運(yùn)行在線程級(jí)別上，必須運(yùn)行在同一個(gè)JVM中。如果競(jìng)爭(zhēng)資源的進(jìn)程不在同一個(gè)JVM中時(shí)，這樣線程鎖就無(wú)法起到作用，必須使用分布式鎖來(lái)控制多個(gè)進(jìn)程對(duì)資源的訪問(wèn)。

分布式鎖的實(shí)現(xiàn)一般有三種方式，使用MySql數(shù)據(jù)庫(kù)行鎖，基于Redis的分布式鎖，以及基于Zookeeper的分布式鎖。本文中我們重點(diǎn)看一下Redis如何實(shí)現(xiàn)分布式鎖。

首先，看一下用于實(shí)現(xiàn)分布式鎖的兩個(gè)Redis基礎(chǔ)命令：

setnx key value

這里的setnx，是"set if Not eXists"的縮寫，表示當(dāng)指定的key值不存在時(shí)，為key設(shè)定值為value。如果key存在，則設(shè)定失敗。

setex key timeout value

setex命令為指定的key設(shè)置值及其過(guò)期時(shí)間（以秒為單位）。如果key已經(jīng)存在，setex命令將會(huì)替換舊的值。

基于這兩個(gè)指令，我們能夠?qū)崿F(xiàn)：

使用setnx 命令，保證同一時(shí)刻只有一個(gè)線程能夠獲取到鎖使用setex 命令，保證鎖會(huì)超期釋放，從而不因一個(gè)線程長(zhǎng)期占有一個(gè)鎖而導(dǎo)致死鎖。

這里將兩個(gè)命令結(jié)合在一起使用的原因是，在正常情況下，如果只使用setnx 命令，使用完成后使用delete命令刪除鎖進(jìn)行釋放，不存在什么問(wèn)題。但是如果獲取分布式鎖的線程在運(yùn)行中掛掉了，那么鎖將不被釋放。如果使用setex 設(shè)置了過(guò)期時(shí)間，即使線程掛掉，也可以自動(dòng)進(jìn)行鎖的釋放。

手寫Redis分布式鎖

接下來(lái)，我們基于Redis+Spring手寫實(shí)現(xiàn)一個(gè)分布式鎖。首先配置Jedis連接池：

@Configuration
public class Config {
    @Bean
    public JedisPool jedisPool(){
        JedisPoolConfig jedisPoolConfig=new JedisPoolConfig();
        jedisPoolConfig.setMaxIdle(100);
        jedisPoolConfig.setMinIdle(1);
        jedisPoolConfig.setMaxWaitMillis(2000);
        jedisPoolConfig.setTestOnBorrow(true);
        jedisPoolConfig.setTestOnReturn(true);
        JedisPool jedisPool=new JedisPool(jedisPoolConfig,"127.0.0.1",6379);
        return  jedisPool;
    }
}

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

public class RedisLock implements Lock {
    @Autowired
    JedisPool jedisPool;

    private static final String key = "lock";
    private ThreadLocal<String> threadLocal = new ThreadLocal<>();

    @Override
    public void lock() {
        boolean b = tryLock();
        if (b) {
            return;
        }
        try {
            TimeUnit.MILLISECONDS.sleep(50);
        } catch (Exception e) {
            e.printStackTrace();
        }
        lock();//遞歸調(diào)用
    }

    @Override
    public boolean tryLock() {
        SetParams setParams = new SetParams();
        setParams.ex(10);
        setParams.nx();
        String s = UUID.randomUUID().toString();
        Jedis resource = jedisPool.getResource();
        String lock = resource.set(key, s, setParams);
        resource.close();
        if ("OK".equals(lock)) {
            threadLocal.set(s);
            return true;
        }
        return false;
    }

    //解鎖判斷鎖是不是自己加的
    @Override
    public void unlock(){
        //調(diào)用lua腳本解鎖
        String script="if redis.call(\"get\",KEYS[1]==ARGV[1] then\n"+
                "   return redis.call(\"del\",KEYS[1])\n"+
                "else\n"+
                "   return 0\n"+
                "end";
        Jedis resource = jedisPool.getResource();
        Object eval=resource.eval(script, Arrays.asList(key),Arrays.asList(threadLocal.get()));
        if (Integer.valueOf(eval.toString())==0){
            resource.close();
            throw new RuntimeException("解鎖失敗");
        }
        /*
        *不寫成下面這種也是因?yàn)椴皇窃硬僮?和ex、nx相同
        String s = resource.get(key);
        if (threadLocal.get().equals(s)){
            resource.del(key);
        }
        */
        resource.close();
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public Condition newCondition() {
        return null;
    }
}

簡(jiǎn)單對(duì)上面代碼中需要注意的地方做一解釋：

加鎖過(guò)程中，使用SetParams 同時(shí)設(shè)置nx和ex的值，保證原子操作通過(guò)ThreadLocal保存key對(duì)應(yīng)的value，通過(guò)value來(lái)判斷鎖是否當(dāng)前線程自己加的，避免線程錯(cuò)亂解鎖釋放鎖的過(guò)程中，使用lua腳本進(jìn)行刪除，保證Redis在執(zhí)行此腳本時(shí)不執(zhí)行其他操作，從而保證操作的原子性

但是，這段手寫的代碼可能會(huì)存在一個(gè)問(wèn)題，就是不能保證業(yè)務(wù)邏輯一定能被執(zhí)行完成，因?yàn)樵O(shè)置了鎖的過(guò)期時(shí)間可能導(dǎo)致過(guò)期。

Redisson

基于上面存在的問(wèn)題，我們可以使用Redisson分布式可重入鎖。Redisson內(nèi)部提供了一個(gè)監(jiān)控鎖的看門狗，它的作用是在Redisson實(shí)例被關(guān)閉前，不斷的延長(zhǎng)鎖的有效期。

引入依賴：

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

配置RedissonClient，然后我們對(duì)常用方法進(jìn)行測(cè)試。

@Configuration
public class RedissonConfig {
    @Bean
    public RedissonClient redissonClient(){
        Config config=new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        RedissonClient redissonClient= Redisson.create(config);
        return redissonClient;
    }
}

lock()

先寫一個(gè)測(cè)試接口：

@GetMapping("/lock")
public String test() {
    RLock lock = redissonClient.getLock("lock");
    lock.lock();
    System.out.println(Thread.currentThread().getName()+" get redisson lock");

    try {
        System.out.println("do something");
        TimeUnit.SECONDS.sleep(20);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    lock.unlock();
    System.out.println(Thread.currentThread().getName()+ " release lock");

   return "locked";
}

進(jìn)行測(cè)試，同時(shí)發(fā)送兩個(gè)請(qǐng)求，redisson鎖生效：

在這里插入圖片描述

lock(long leaseTime, TimeUnit unit)

Redisson可以給lock()方法提供leaseTime參數(shù)來(lái)指定加鎖的時(shí)間，超過(guò)這個(gè)時(shí)間后鎖可以自動(dòng)釋放。測(cè)試接口：

@GetMapping("/lock2")
public String test2() {
    RLock lock = redissonClient.getLock("lock");
    lock.lock(10,TimeUnit.SECONDS);
    System.out.println(Thread.currentThread().getName()+" get redisson lock");

    try {
        System.out.println("do something");
        TimeUnit.SECONDS.sleep(20);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println(Thread.currentThread().getName()+ " release lock");
    return "locked";
}

運(yùn)行結(jié)果：

在這里插入圖片描述

可以看出，在第一個(gè)線程還沒(méi)有執(zhí)行完成時(shí)，就釋放了redisson鎖，第二個(gè)線程進(jìn)入后，兩個(gè)線程可以同時(shí)執(zhí)行被鎖住的代碼邏輯。這樣可以實(shí)現(xiàn)無(wú)需調(diào)用unlock方法手動(dòng)解鎖。

tryLock(long waitTime, long leaseTime, TimeUnit unit)

tryLock方法會(huì)嘗試加鎖，最多等待waitTime秒，上鎖以后過(guò)leaseTime秒自動(dòng)解鎖；如果沒(méi)有等待時(shí)間，鎖不住直接返回false。

@GetMapping("/lock3")
public String test3() {
    RLock lock = redissonClient.getLock("lock");
    try {
        boolean res = lock.tryLock(5, 30, TimeUnit.SECONDS);
        if (res){
            try{
                System.out.println(Thread.currentThread().getName()+" 獲取到鎖，返回true");
                System.out.println("do something");
                TimeUnit.SECONDS.sleep(20);
            }finally {
                lock.unlock();
                System.out.println(Thread.currentThread().getName()+" 釋放鎖");
            }
        }else {
            System.out.println(Thread.currentThread().getName()+" 未獲取到鎖，返回false");
        }
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "lock";
}

運(yùn)行結(jié)果：

在這里插入圖片描述

可見(jiàn)在第一個(gè)線程獲得鎖后，第二個(gè)線程超過(guò)等待時(shí)間仍未獲得鎖，返回false放棄獲得鎖的過(guò)程。

除了以上單機(jī)Redisson鎖以外，還支持我們之前提到過(guò)的哨兵模式和集群模式，只需要改變Config的配置即可。以集群模式為例：

@Bean
public RedissonClient redissonClient(){
    Config config=new Config();
    config.useClusterServers().addNodeAddress("redis://172.20.5.170:7000")
        .addNodeAddress("redis://172.20.5.170:7001")
        .addNodeAddress("redis://172.20.5.170:7002")
        .addNodeAddress("redis://172.20.5.170:7003")
        .addNodeAddress("redis://172.20.5.170:7004")
        .addNodeAddress("redis://172.20.5.170:7005");
    RedissonClient redissonClient = Redisson.create(config);
    return redissonClient;
}

RedLock紅鎖

下面介紹一下Redisson紅鎖RedissonRedLock，該對(duì)象也可以用來(lái)將多個(gè)RLock對(duì)象關(guān)聯(lián)為一個(gè)紅鎖，每個(gè)RLock對(duì)象實(shí)例可以來(lái)自于不同的Redisson實(shí)例。

RedissonRedLock針對(duì)的多個(gè)Redis節(jié)點(diǎn)，這多個(gè)節(jié)點(diǎn)可以是集群，也可以不是集群。當(dāng)我們使用RedissonRedLock時(shí)，只要在大部分節(jié)點(diǎn)上加鎖成功就算成功。看一下使用：

@GetMapping("/testRedLock")
public void testRedLock() {
    Config config1 = new Config();
    config1.useSingleServer().setAddress("redis://172.20.5.170:6379");
    RedissonClient redissonClient1 = Redisson.create(config1);

    Config config2 = new Config();
    config2.useSingleServer().setAddress("redis://172.20.5.170:6380");
    RedissonClient redissonClient2 = Redisson.create(config2);

    Config config3 = new Config();
    config3.useSingleServer().setAddress("redis://172.20.5.170:6381");
    RedissonClient redissonClient3 = Redisson.create(config3);

    String resourceName = "REDLOCK";
    RLock lock1 = redissonClient1.getLock(resourceName);
    RLock lock2 = redissonClient2.getLock(resourceName);
    RLock lock3 = redissonClient3.getLock(resourceName);

    RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);
    boolean isLock;
    try {
        isLock = redLock.tryLock(5, 30, TimeUnit.SECONDS);
        if (isLock) {
            System.out.println("do something");
            TimeUnit.SECONDS.sleep(20);
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        redLock.unlock();
    }
}

相對(duì)于單Redis節(jié)點(diǎn)來(lái)說(shuō)，RedissonRedLock的優(yōu)點(diǎn)在于防止了單節(jié)點(diǎn)故障造成整個(gè)服務(wù)停止運(yùn)行的情況；并且在多節(jié)點(diǎn)中鎖的設(shè)計(jì)，及多節(jié)點(diǎn)同時(shí)崩潰等各種意外情況有自己獨(dú)特的設(shè)計(jì)方法。使用RedissonRedLock，性能方面會(huì)比單節(jié)點(diǎn)Redis分布式鎖差一些，但可用性比普通鎖高很多。