如何實現基於 Redis 的分散式鎖

前言

分散式鎖定在分散式應用中應用廣泛，想要搞懂一個新事物首先得了解它的由來，這樣才能更加的理解甚至可以舉一反三。

首先談到分散式鎖定自然也聯想到分散式應用。

在我們將應用程式拆分為分散式應用之前的單機系統中，對一些並發場景讀取公共資源時如扣庫存，賣車票之類的需求可以簡單的使用同步或者是加鎖就可以實現。

但是應用分散式了之後系統由以前的單進程多線程的程式變為了多進程多線程，這時使用以上的解決方案明顯就不夠了。

因此業界常用的解決方案通常是藉助於一個第三方元件並利用它自身的排他性來達到多進程的互斥。如：

• 基於 DB 的唯一索引。

• 基於 ZK 的暫時有序節點。

• 基於 Redis 的 NX EX 參數。

這裡主要基於 Redis 進行討論。

實作

既然是選用了 Redis，那麼它就得具有排他性才行。同時它最好也有鎖的一些基本特性：

• 高效能(加、解鎖時高效能)

• 可以使用阻塞鎖與非阻塞鎖。

• 不能出現死鎖。

• 可用性(不能出現節點 down 掉後加鎖失敗)。

這裡利用 Redis set key 時的一個 NX 參數可以保證在這個 key 不存在的情況下寫入成功。並且再加上 EX 參數可以讓該 key 在逾時之後自動刪除。

所以利用以上兩個特性可以保證在同一時刻只會有一個進程獲得鎖，並且不會出現死鎖(最壞的情況就是超時自動刪除 key)。

加上鎖定

實作程式碼如下：

    private static final String SET_IF_NOT_EXIST = "NX";
    private static final String SET_WITH_EXPIRE_TIME = "PX";
    
    public  boolean tryLock(String key, String request) {
        String result = this.jedis.set(LOCK_PREFIX + key, request, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, 10 * TIME);

        if (LOCK_MSG.equals(result)){
            return true ;
        }else {
            return false ;
        }
    }

注意這裡使用的 jedis 的

String set(String key, String value, String nxxx, String expx, long time);

api。

此指令可以保證 NX EX 的原子性。

一定不要把兩個指令(NX EX)分開執行，如果在 NX 之後程式出現問題就有可能產生死鎖。

阻塞鎖定

同時也可以實現一個阻塞鎖：

    //一直阻塞
    public void lock(String key, String request) throws InterruptedException {

        for (;;){
            String result = this.jedis.set(LOCK_PREFIX + key, request, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, 10 * TIME);
            if (LOCK_MSG.equals(result)){
                break ;
            }
                
              //防止一直消耗 CPU  
            Thread.sleep(DEFAULT_SLEEP_TIME) ;
        }

    }
    
     //自定义阻塞时间
     public boolean lock(String key, String request,int blockTime) throws InterruptedException {

        while (blockTime >= 0){

            String result = this.jedis.set(LOCK_PREFIX + key, request, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, 10 * TIME);
            if (LOCK_MSG.equals(result)){
                return true ;
            }
            blockTime -= DEFAULT_SLEEP_TIME ;

            Thread.sleep(DEFAULT_SLEEP_TIME) ;
        }
        return false ;
    }

解鎖

解鎖也很簡單，其實就是把這個key 刪掉就萬事大吉了，例如使用del key 指令。

但現實往往沒有那麼 easy。

如果進程 A 取得了鎖定設定了逾時時間，但是由於執行週期較長導致到了逾時時間之後鎖定就自動釋放了。這時進程 B 取得了該鎖執行很快就釋放鎖。這樣就會出現進程 B 將進程 A 的鎖定釋放了。

所以最好的方法是在每次解鎖時都需要判斷鎖定是否是自己的。

這時就需要結合加鎖機制一起實現了。

加上鎖定時需要傳遞一個參數，並將該參數作為這個 key 的 value，這樣每次解鎖時判斷 value 是否相等即可。

所以解鎖程式碼就不能是簡單的 del了。

    public  boolean unlock(String key,String request){
        //lua script
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";

        Object result = null ;
        if (jedis instanceof Jedis){
            result = ((Jedis)this.jedis).eval(script, Collections.singletonList(LOCK_PREFIX + key), Collections.singletonList(request));
        }else if (jedis instanceof JedisCluster){
            result = ((JedisCluster)this.jedis).eval(script, Collections.singletonList(LOCK_PREFIX + key), Collections.singletonList(request));
        }else {
            //throw new RuntimeException("instance is error") ;
            return false ;
        }

        if (UNLOCK_MSG.equals(result)){
            return true ;
        }else {
            return false ;
        }
    }

這裡使用了一個 lua 腳本來判斷 value 是否相等，相等才執行 del 指令。

使用 lua 也可以保證這裡兩個運算的原子性。

因此上文提到的四個基本特性也能滿足了：

• 使用 Redis 可以保證效能。

• 阻塞鎖定與非阻塞鎖定見上文。

• 利用超時機制解決了死鎖。

• Redis 支援叢集部署提高了可用性。

使用

我自己有擼了一個完整的實現，並且已經用於了生產，有興趣的朋友可以開箱使用:

maven 依賴：

<dependency>
    <groupId>top.crossoverjie.opensource</groupId>
    <artifactId>distributed-redis-lock</artifactId>
    <version>1.0.0</version>
</dependency>

設定bean :

@Configuration
public class RedisLockConfig {

    @Bean
    public RedisLock build(){
        RedisLock redisLock = new RedisLock() ;
        HostAndPort hostAndPort = new HostAndPort("127.0.0.1",7000) ;
        JedisCluster jedisCluster = new JedisCluster(hostAndPort) ;
        // Jedis 或 JedisCluster 都可以
        redisLock.setJedisCluster(jedisCluster) ;
        return redisLock ;
    }

}

使用：

    @Autowired
    private RedisLock redisLock ;

    public void use() {
        String key = "key";
        String request = UUID.randomUUID().toString();
        try {
            boolean locktest = redisLock.tryLock(key, request);
            if (!locktest) {
                System.out.println("locked error");
                return;
            }


            //do something

        } finally {
            redisLock.unlock(key,request) ;
        }

    }

使用很簡單。這裡主要是想利用 Spring 來幫我們管理 RedisLock 這個單例的 bean，所以在釋放鎖的時候需要手動(因為整個上下文只有一個 RedisLock 實例)的傳入 key 以及 request(api 看起來不是特別優雅)。

也可以在每次使用鎖的時候 new 一個 RedisLock 傳入 key 以及 request，這樣倒是在解鎖時很方便。但需要自行管理 RedisLock 的實例。各有優劣吧。

單測

在做這個專案的時候讓我不得不想提一下單測。

因為這個應用是強烈依賴第三方元件的(Redis)，但是在單測中我們需要排除掉這種依賴。例如其他夥伴 fork 了這個專案想在本地跑一遍單測，結果運作不起來：

1. 有可能是 Redis 的 ip、連接埠和單測裡的不一致。

2. Redis 本身可能也有問題。

3. 也有可能是該同學的環境中並沒有 Redis。

所以最好是要把這些外在不穩定的因素排除掉，單測只測我們寫好的程式碼。

於是就可以引入單測利器 Mock 了。

它的想法很簡答，就是要把你所依賴的外部資源統統屏蔽掉。如：資料庫、外部介面、外部文件等等。

使用方式也挺简单，可以参考该项目的单测：

    @Test
    public void tryLock() throws Exception {
        String key = "test";
        String request = UUID.randomUUID().toString();
        Mockito.when(jedisCluster.set(Mockito.anyString(), Mockito.anyString(), Mockito.anyString(),
                Mockito.anyString(), Mockito.anyLong())).thenReturn("OK");

        boolean locktest = redisLock.tryLock(key, request);
        System.out.println("locktest=" + locktest);

        Assert.assertTrue(locktest);

        //check
        Mockito.verify(jedisCluster).set(Mockito.anyString(), Mockito.anyString(), Mockito.anyString(),
                Mockito.anyString(), Mockito.anyLong());
    }

这里只是简单演示下，可以的话下次仔细分析分析。

它的原理其实也挺简单，debug 的话可以很直接的看出来：

这里我们所依赖的 JedisCluster 其实是一个 cglib 代理对象。所以也不难想到它是如何工作的。

比如这里我们需要用到 JedisCluster 的 set 函数并需要它的返回值。

Mock 就将该对象代理了，并在实际执行 set 方法后给你返回了一个你自定义的值。

这样我们就可以随心所欲的测试了，完全把外部依赖所屏蔽了。

总结

至此一个基于 Redis 的分布式锁完成，但是依然有些问题。

• 如在 key 超时之后业务并没有执行完毕但却自动释放锁了，这样就会导致并发问题。

• 就算 Redis 是集群部署的，如果每个节点都只是 master 没有 slave，那么 master 宕机时该节点上的所有 key 在那一时刻都相当于是释放锁了，这样也会出现并发问题。就算是有 slave 节点，但如果在数据同步到 salve 之前 master 宕机也是会出现上面的问题。

感兴趣的朋友还可以参考 Redisson 的实现。

以上是如何實現基於 Redis 的分散式鎖的詳細內容。更多資訊請關注PHP中文網其他相關文章！