分散式鎖

概念

對某個操作上了鎖後，這個鎖對所有節點都有效，也就是保證所有節點互斥訪問。

傳統的鎖只對自己的節點有效。

基於Redis實現分散式鎖

原理

使用Redis的setnx命令，setnx在設定鍵值對的時候，先判斷鍵值對是否存在，存在返回0，不存在返回1，根據這個原理，需要加鎖的時候使用setnx設定一個lock，在解鎖之前，其他人上鎖lock會返回0，表示上鎖失敗，也就是這個鎖已經存在。在解鎖的時候使用del刪除這個鎖即可。這就達成了加鎖解鎖的目的。

產生的問題

如果上鎖後，上鎖的伺服器掛掉了，其他伺服器就沒辦法解鎖，就導致死鎖。

解決方案

上鎖後，再設定鎖的過期時間，過了期限這個鎖自動失效。

使用setnx和expire命令。

setnx <key> <value>

expire <key> <seconds>

方案的優化——原子操作

需要優化原因

這涉及到原子操作，若上鎖操作執行後，還沒設定過期時間，此時伺服器掛掉了，也會導致死鎖。

所以需要上鎖的同時設定過期時間。

解決方案

可以使用命令set <key><value>nx ex <time>

方案的繼續優化——UUID防止誤刪

需要優化原因

假設A上鎖後，做了操作，然後伺服器卡頓了，A的鎖過期後A還沒反應過來。

此時B上鎖做操作，還沒等到B解鎖，此時A反應過來進行解鎖，這時候解鎖解的是B的鎖。

解決方案

使用uuid表示不同的操作，首先做操作的時候自己隨機生成一個自己的uuid

set lock uuid nx ex 10

釋放鎖的時候，首先判斷自己的uuid和要釋放鎖的uuid是否一致

程式碼

@RestController
@RequestMapping("/redisTest")
public class RedisTestController {
    @Autowired
    private RedisTemplate redisTemplate;

    @GetMapping("testLock")
    public void testLock(){
        String uuid = UUID.randomUUID().toString();  //隨機生成當前操作的uuid
        Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock",uuid,3, TimeUnit.SECONDS); //上鎖，lock儲存返回資訊
        if(lock){  //如果上鎖成功

//----------------------對資料操作------------------------------------------
            Object value = redisTemplate.opsForValue().get("num"); 
            if(StringUtils.isEmpty((String)value)) return; 
            int num = Integer.parseInt(value+"");     
            redisTemplate.opsForValue().set("num",++num);

//----------------------開始判定自己的uuid與鎖的uuid是否一致---------------------------------------
          String lockUuid=  (String) redisTemplate.opsForValue().get("lock");
          if(lockUuid.equals(uuid))         //如果uuid一致，則解鎖
            redisTemplate.delete("lock");
        }
        else{  //如果上鎖失敗，則等一等再試著上鎖
            try {
                Thread.sleep(100);
                testLock();
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
}

方案最終優化——LUA保證刪除的原子性

需要優化原因

A上鎖，做了具體操作，在正準備刪除的時候（比較了uuid，且一致），A剛好過期，此時B上鎖，A再刪除時刪除的是B的鎖。

解決方案

可以使用LUA指令碼保證刪除操作的原子性

程式碼

@RestController
@RequestMapping("/redisTest")
public class RedisTestController {
    @Autowired
    private RedisTemplate redisTemplate;

    @GetMapping("testLock")
    public void testLock(){
        String uuid = UUID.randomUUID().toString(); //隨機生成uuid
        String skuId= "25";  
        String locKey = "lock:"+skuId;

        Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock",uuid,3, TimeUnit.SECONDS);
        if(lock){

//-----------------------------對資料操作--------------------------------------
            Object value = redisTemplate.opsForValue().get("num");
            if(StringUtils.isEmpty((String)value)) return;
            int num = Integer.parseInt(value+"");
            redisTemplate.opsForValue().set("num",++num);

//------------------------------刪除操作--------------------------------------
            String script="if redis.call('get',KEYS[1]) == ARGV[1] " +
                    "then return redis.call('del',KEYS[1]) else return 0 end";   //指令碼內容

            DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
            redisScript.setScriptText(script);   //載入指令碼
            redisScript.setResultType(Long.class);
            redisTemplate.execute(redisScript, Arrays.asList(locKey),uuid);  //執行
        }
        else{
            try {
                Thread.sleep(100);
                testLock();
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
}