redis應(yīng)用問題解決

緩存穿透

什么是緩存穿透？

可以參考下圖，當(dāng)客戶端發(fā)送讀的請求過來時，會先訪問緩存中的數(shù)據(jù)，如果不存在則直接去訪問MySQL服務(wù)器中的數(shù)據(jù)。這時候如果MySQL服務(wù)器中并不存在他請求對應(yīng)的信息，請求就會反反復(fù)復(fù)一直訪問MySQL服務(wù)器，黑客利用此漏洞進(jìn)行攻擊可能壓垮數(shù)據(jù)庫。

解決方案

• 方案一：緩存空值

  如果MySQL服務(wù)器中不存在相對應(yīng)的數(shù)據(jù)，可以將對應(yīng)的key的value值設(shè)置為空，當(dāng)請求再次訪問時可以直接去緩存中讀取空值

• 方案二：布隆過濾器

  在訪問緩存層和存儲層之前，將存在的key用布隆過濾器提前保存起來，做第一層攔截，當(dāng)收到一個對key請求時先用布隆過濾器驗證是key否存在，如果存在在進(jìn)入緩存層、存儲層。可以使用bitmap做布隆過濾器。這種方法適用于數(shù)據(jù)命中不高、數(shù)據(jù)相對固定、實時性低的應(yīng)用場景，代碼維護(hù)較為復(fù)雜，但是緩存空間占用少。

緩存擊穿

什么是緩存擊穿？

比如微博的熱點事件，大家都在同一時間點訪問這個請求，就可能造成服務(wù)器崩潰的情況，就好像子彈打在墻上一樣，始終往一個地方大，遲早打穿這堵墻

解決方案

• 方案一：提前設(shè)置預(yù)熱數(shù)據(jù)

  在redis高峰訪問之前，把一些熱門數(shù)據(jù)提前存入到redis里面，加大這些熱門數(shù)據(jù)key的時長

• 方案二：使用分布式鎖

  類似于Java中的鎖機(jī)制一樣，一個時間只有一個用戶可以訪問，當(dāng)這個用戶訪問完之后，解鎖了，其他用戶才可以開始訪問，這個在后面也會詳細(xì)展開講解

緩存雪崩

什么時候緩存雪崩？

舉個栗子，在雙十一的時候，非常多的用戶訪問非常多的請求，我們雖然提前做了緩存，但在一定時間后這些緩存同時失效，那就會有一大批的請求直接訪問MySQL數(shù)據(jù)庫，給服務(wù)器帶來巨大的壓力

解決方案

• 方案一：構(gòu)建多級緩存架構(gòu)

• 方案二：設(shè)置過期標(biāo)志更新緩存

  記錄緩存過期時間，快到過期時間時觸發(fā)另外一個線程更新過期時間

• 方案三：將緩存失效時間分散開

  比如A請求的失效時間為10分鐘后，B請求的失效時間可以設(shè)置為10分01秒后，依次類推，即使過期了，請求直接訪問時也可以錯開時間點

分布式鎖

業(yè)務(wù)需求

我們上述說到的解決緩存擊穿的方案，可以使用分布式鎖，讓請求一個一個訪問。隨著時代的發(fā)展，現(xiàn)在我們使用redis已經(jīng)不是單單的一臺服務(wù)器了， 我們會建一個集群，但是鎖這個東西他是不能橫跨服務(wù)器的，這種情況redis也提供了解決方案

解決方案

1. 使用setnx命令

   我們都知道setnx命令如果這個key不存在的話可以對其value進(jìn)行設(shè)置，但key存在時是不允許設(shè)置的，我們就可以利用這一點，將此key作為鎖。

2. 設(shè)置過期時間

   就好像有一個人去上廁所，外面排著長隊，結(jié)果這個人上著上著突然睡著了，但是外面的人就無法使用。所以我們要將鎖設(shè)置一定的過期時間，如果長時間沒有完成就要自動釋放鎖

3. 具體命令

   # set key value nx ex second : 其中nx等同于setnx，ex設(shè)置過期時間單位為秒
   set k2 v2 nx ex 10
   

Java中使用

@GetMapping("/testLock")
public void testLock(){
    //1.獲取鎖
    Boolean isLock = redisTemplate.opsForValue().setIfAbsent("kkk", "vvv", 10, TimeUnit.SECONDS);
    //2.操作數(shù)據(jù)
    if (isLock){
        //查詢key為num的數(shù)值
        Object value = redisTemplate.opsForValue().get("num");
        if (StringUtils.isEmpty(value)){
            return;
        }
        //把值轉(zhuǎn)換為數(shù)字并加1
        int i = Integer.parseInt(value + "");
        redisTemplate.opsForValue().set("num",i+1);
        //釋放鎖
        redisTemplate.delete("kkk");
    }else {
        //3.獲取鎖失敗，每個0.1秒再獲取
        try {
            Thread.sleep(100);
            testLock();
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

設(shè)置num的值為0，使用ab工具進(jìn)行多次訪問，如果鎖有效的話，1000個請求num的值應(yīng)該為1000

ab -n 1000 -c 100 http://192.168.0.10:8080/redisTest/testLock
127.0.0.1:6379> get num
"1000"

問題解決

其實上述案例中還是有那么一些些問題的，需要我們來梳理一下

問題一

如下圖所示，A請求在執(zhí)行操作的過程中宕機(jī)了，但是key的過期時間已到，B請求獲取到了鎖并加上了鎖，B在執(zhí)行操作的過程中，A相應(yīng)過來了，手動釋放了鎖，這時其他的請求又會一起擠進(jìn)來，可能會出現(xiàn)兩個請求操作一個數(shù)據(jù)的情況

解決方案：使用UUID作為value值，防止誤刪除

@GetMapping("/testLock")
public void testLock(){
    String uuid = UUID.randomUUID().toString();
    //1.獲取鎖
    Boolean isLock = redisTemplate.opsForValue().setIfAbsent("kkk", "uuid", 10, TimeUnit.SECONDS);
    //2.操作數(shù)據(jù)
    if (isLock){
        //查詢key為num的數(shù)值
        Object value = redisTemplate.opsForValue().get("num");
        if (StringUtils.isEmpty(value)){
            return;
        }
        //把值轉(zhuǎn)換為數(shù)字并加1
        int i = Integer.parseInt(value + "");
        redisTemplate.opsForValue().set("num",i+1);
        //釋放鎖
        //判斷是否為自己的鎖
        String keyForUuid = (String) redisTemplate.opsForValue().get("kkk");
        if (uuid.equals(keyForUuid)){
            redisTemplate.delete("kkk");
        }
    }else {
        //3.獲取鎖失敗，每個0.1秒再獲取
        try {
            Thread.sleep(100);
            testLock();
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

問題二

如下入、圖所示，A在釋放鎖的過程中，先判斷了uuid是相同的，正準(zhǔn)備刪除時，鎖的過期時間到了，自動刪除后，B獲取到了鎖并加上了鎖，A再刪除了這個鎖

解決方案：使用LUA腳本保證刪除的原子性

@GetMapping("/testLock")
public void testLock(){
    String uuid = UUID.randomUUID().toString();
    //1.獲取鎖
    Boolean isLock = redisTemplate.opsForValue().setIfAbsent("kkk", "uuid", 10, TimeUnit.SECONDS);
    //2.操作數(shù)據(jù)
    if (isLock){
        //查詢key為num的數(shù)值
        Object value = redisTemplate.opsForValue().get("num");
        if (StringUtils.isEmpty(value)){
            return;
        }
        //把值轉(zhuǎn)換為數(shù)字并加1
        int i = Integer.parseInt(value + "");
        redisTemplate.opsForValue().set("num",i+1);

        // 定義lua 腳本
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        // 使用redis執(zhí)行l(wèi)ua執(zhí)行
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptText(script);
        // 設(shè)置一下返回值類型 為Long
        // 因為刪除判斷的時候，返回的0,給其封裝為數(shù)據(jù)類型。如果不封裝那么默認(rèn)返回String 類型，
        // 那么返回字符串與0 會有發(fā)生錯誤。
        redisScript.setResultType(Long.class);
        // 第一個要是script 腳本 ，第二個需要判斷的key，第三個就是key所對應(yīng)的值。
        redisTemplate.execute(redisScript, Arrays.asList("kkk"), uuid);
        
    }else {
        //3.獲取鎖失敗，每個0.1秒再獲取
        try {
            Thread.sleep(100);
            testLock();
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}