摘要：首先談到分布式鎖自然也就聯想到分布式應用。如基于的唯一索引?；诘呐R時有序節點。這里主要基于進行討論。該命令可以保證的原子性。所以最好的方式是在每次解鎖時都需要判斷鎖是否是自己的?？偨Y至此一個基于的分布式鎖完成，但是依然有些問題。

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

首先談到分布式鎖自然也就聯想到分布式應用。

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

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

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

基于 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 依賴：

    top.crossoverjie.opensource
    distributed-redis-lock
    1.0.0

配置 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 的實例。各有優劣吧。

項目源碼在：

https://github.com/crossoverJie/distributed-lock-redis

歡迎討論。

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

因為這個應用是強依賴于第三方組件的(Redis)，但是在單測中我們需要排除掉這種依賴。比如其他伙伴 fork 了該項目想在本地跑一遍單測，結果運行不起來：

有可能是 Redis 的 ip、端口和單測里的不一致。

Redis 自身可能也有問題。

也有可能是該同學的環境中并沒有 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 的實現。

最近在總結一些 Java 相關的知識點，感興趣的朋友可以一起維護。

地址: https://github.com/crossoverJie/Java-Interview