Redis中怎樣實現分布式鎖

這期內容當中小編將會給大家帶來有關Redis中怎樣實現分布式鎖，文章內容豐富且以專業的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

在Java里面，如果多個線程同時訪問公共資源，不做些同步措施可能就會對數據的一致性造成破壞，例如下面的例子，多個線程同時對COUNT做 +1操作（為了測試方便就使用單機多個線程執行）

import lombok.SneakyThrows;import java.util.concurrent.Semaphore;public class ThreadLockTest extends Thread{// 多個線程對COUNT進行操作    public static volatile int COUNT = 0;    // 信號燈，用于所有線程執行完后主線程輸出    public static Semaphore semaphore = new Semaphore(0);    @SneakyThrows    @Override    public void run() {for (int i = 0; i < 10; i++) {
            System.out.println("線程：" + Thread.currentThread().getName() + " 操作：" + COUNT++);            Thread.sleep(100);        }try {
            System.out.println("線程：" + Thread.currentThread().getName() + " 執行完成");            semaphore.release();        } catch (Exception e) {
            e.printStackTrace();        }
    }public static void main(String[] args) throws InterruptedException {new ThreadLockTest().start();        new ThreadLockTest().start();        new ThreadLockTest().start();        semaphore.acquire(3);        System.out.println("主線程輸入總數：" + COUNT);    }
}

3個線程，每個線程循環執行 "i+1" 10次，預期輸出結果應該是30，輸出結果不一定是30，這是因為沒同步處理，對一致性造成破壞

不做同步處理會造成結果與我們期望的不一樣，所以需要用到synchronized關鍵字或者Lock對象來做同步處理，但是這兩者都只適用于單機的服務器。在分布式服務器下，無法使用synchronized關鍵字和Lock對象進行同步操作，這個時候就要想想其他辦法，Redis就提供了一些分布式鎖的方案。如下：

Redis提供呢NX函數給我們設置鎖，就是向Redis插入一個key-value,key就是鎖名稱，value可以使用UUID或者 機器號+線程名稱，當用完的時候再刪除這個key，但是只有value一致的情況下才能刪除，這樣才能保證加鎖與解鎖的線程是同一個，SETNX是Redis提供的一個命令，當Redis中不存在指定的key時才能成功插入，Jedis也有對應的函數。

public static long lock(String key, String requestId){return jedis.setnx(key, requestId);    // 獲取鎖，返回1則獲取鎖成功，返回0則獲取鎖失敗}

這里順便把解鎖的函數貼出來，這里使用到lua腳本直接判斷刪除。如果不用lua腳本刪除我們的程序要分成幾部: 1.根據key查詢值；2.判斷值與當前線程值是否一致；3.刪除key。但是這幾部走下來，如果中間出了什么問題就會造成一些麻煩（此處下面再展開可能會有哪些麻煩）。所以為了保證原子性，用lua一步到位。

//刪除key的lua腳本，先比較requestId是否相等，相等則刪除,使用腳本執行保證原子性private static final String DEL_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";public static long unLock(String key, String requestId) {//刪除成功表示解鎖成功,返回1則解鎖成功，返回0則解鎖失敗    return jedis.eval(DEL_SCRIPT, Collections.singletonList(key), Collections.singletonList(requestId));}

再執行一下代碼加上鎖之后的執行結果

@SneakyThrows@Overridepublic void run() {for (int i = 0; i < 10; i++) {
        String uuid = UUID.randomUUID().toString();        while(1 != JedisTest.lock("COUNT_ADD_LOCK", uuid));        System.out.println("線程：" + Thread.currentThread().getName() + " 操作：" + COUNT++);        Thread.sleep(100);        JedisTest.unLock("COUNT_ADD_LOCK", uuid);    }try {
        System.out.println("線程：" + Thread.currentThread().getName() + " 執行完成");        semaphore.release();    } catch (Exception e) {
        e.printStackTrace();    }
}

執行結果也每次都能保證輸出的是我們的預期值

上面的加鎖方案會有個問題，就是當加鎖的那個線程所在的服務器宕機之后，鎖就一致存在，會導致其他線程一直在阻塞，所以需要加上一個失效時間。當加鎖的服務器宕機之后過了過了有效時間其他服務器就能再上鎖。

加失效時間

private static final int LOCK_EX_SECONDS = 5;public static long lock(String key, String requestId){return jedis.setex(key,LOCK_EX_SECONDS,requestId);}

這里往Redis插入數據的時候必須同時加上過期時間，如果分成兩步走：
    1.插入數據，
    2.給key設置失效時間 
有可能在第一步插入數據之后服務器宕機了，這樣就造成失效時間沒有設置上，死鎖的問題就又出現了，所以必須插入數據的時候同時設置失效時間。

給key加了失效時間比沒有好，但是缺陷還是有很多，因為失效時間都是根據我們開發人員評估這段業務執行起來需要多少時間，并不是準確計算每次請求這段業務所需的時間。這樣就會造成一個問題：我們業務還沒執行完，鎖因為時間到期而被釋放，再有其他線程同時訪問到這部分公共資源，也會破壞其一致性。還有一個就是上面提到解鎖操作那3步（1.根據key查詢值；2.判斷值與當前線程值是否一致；3.刪除key）不能分開執行的原因，例如：
    線程A：lock()，插入數據并且設置過期時間
    線程A：執行業務完畢
    線程A：根據鎖的key獲取value--解鎖第一步
    線程A：判斷當前線程的value與鎖的value一致--解鎖第二部
        上面判斷結果一致情況下，key這個時候剛好過了有效時間
    線程B：lock(),獲取鎖成功
    線程A：unlock() 解鎖成功，因為上面已經判斷了value一致，所以只需要del(key)就能解鎖
從上面這個流程看到，如果解鎖不保證原子性可能會出現線程A把線程B的鎖釋放的問題。
還有鎖失效的問題需要如何處理？

Redisson

可以使用Redisson來實現鎖，Redisson具有一個看門狗機制，Redisson實現鎖的方案是給鎖設置一個過期時間，當業務還沒執行完Redisson會更新一下鎖的失效時間，如果發生宕機情況鎖的有效時間過了自然就釋放了，而且Redisson是可重入鎖。下面看一下Redisson加鎖解鎖代碼

public class RedissonConfig {public static final String REDIS_ADDRESS_PORT = "redis://192.168.0.90:6380";    public static RedissonClient REDISSON_CLIENT;    public static synchronized RedissonClient getRedisson(){if(REDISSON_CLIENT == null) {
            Config config = new Config();            SingleServerConfig singleServerConfig = config.useSingleServer();            singleServerConfig.setAddress(REDIS_ADDRESS_PORT);            REDISSON_CLIENT = Redisson.create(config);        }return REDISSON_CLIENT;    }

}

@SneakyThrows@Overridepublic void run() {for (int i = 0; i < 10; i++) {
        RLock lock = RedissonConfig.getRedisson().getLock("COUNT_ADD_LOCK");        while(!lock.tryLock());        System.out.println("線程：" + Thread.currentThread().getName() + " 操作：" + COUNT++);        Thread.sleep(100);        lock.unlock();    }try {
        System.out.println("線程：" + Thread.currentThread().getName() + " 執行完成");        semaphore.release();    } catch (Exception e) {
        e.printStackTrace();    }
}

多次的執行結果都能達到預期值

上述就是小編為大家分享的Redis中怎樣實現分布式鎖了，如果剛好有類似的疑惑，不妨參照上述分析進行理解。如果想知道更多相關知識，歡迎關注億速云行業資訊頻道。