[[317484]]

緩存雪崩

出現過程

假設有如下一個系統，高峰期請求為5000次/秒，4000次走了緩存，只有1000次落到了數據庫上，數據庫每秒1000的并發是一個正常的指標，完全可以正常工作，但如果緩存宕機了，或者緩存設置了相同的過期時間，導致緩存在同一時刻同時失效，每秒5000次的請求會全部落到數據庫上，數據庫立馬就死掉了，因為數據庫一秒最多抗2000個請求，如果DBA重啟數據庫，立馬又會被新的請求打死了，這就是緩存雪崩。

解決方法

1. 事前：redis高可用，主從+哨兵，redis cluster，避免全盤崩潰
2. 事中：本地ehcache緩存 + hystrix限流&降級，避免MySQL被打死
3. 事后：redis持久化RDB+AOF，快速恢復緩存數據
4. 緩存的失效時間設置為隨機值，避免同時失效

緩存穿透

出現過程

假如客戶端每秒發送5000個請求，其中4000個為黑客的惡意攻擊，即在數據庫中也查不到。舉個例子，用戶id為正數，黑客構造的用戶id為負數，如果黑客每秒一直發送這4000個請求，緩存就不起作用，數據庫也很快被打死。

解決方法

1. 對請求參數進行校驗，不合理直接返回
2. 查詢不到的數據也放到緩存，value為空，如 set -999 ""
3. 使用布隆過濾器，快速判斷key是否在數據庫中存在，不存在直接返回

緩存擊穿

出現過程

設置了過期時間的key，承載著高并發，是一種熱點數據。從這個key過期到重新從MySQL加載數據放到緩存的一段時間，大量的請求有可能把數據庫打死。緩存雪崩是指大量緩存失效，緩存擊穿是指熱點數據的緩存失效

解決方法

1. 設置key永遠不過期，或者快過期時，通過另一個異步線程重新設置key
2. 當從緩存拿到的數據為null，重新從數據庫加載數據的過程上鎖，下面寫個分布式鎖實現的demo

Redis實現分布式鎖

我之前的文章寫到了Redis實現分布式鎖的原理，這里就不再詳細概述了

在Redis中使用簡單強大的Lua腳本

1.加鎖執行命令

SET resource_name random_value NX PX 30000 

2.解鎖執行腳本

if redis.call("get", KEYS[1]) == ARGV[1] then  
    return redis.call("del", KEYS[1])  
else  
    return 0  
end 

寫一個分布式鎖工具類

public class LockUtil { 
 
    private static final String OK = "OK"; 
    private static final Long LONG_ONE = 1L; 
    private static final String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end"; 
 
    public static boolean tryLock(String key, String value, long expire) { 
        Jedis jedis = RedisPool.getJedis(); 
        SetParams setParams = new SetParams(); 
        setParams.nx().px(expire); 
        return OK.equals(jedis.set(key, value, setParams)); 
    } 
 
    public static boolean releaseLock(String key, String value) { 
        Jedis jedis = RedisPool.getJedis(); 
        return LONG_ONE.equals(jedis.eval(script, 1, key, value)); 
    } 
} 

工具類寫起來還是挺簡單的

示例代碼

public String getData(String key) { 
    String lockKey = "key"; 
    String lockValue = String.valueOf(System.currentTimeMillis()); 
    long expireTime = 1000L; 
    String value = getFromRedis(key); 
    if (value == null) { 
        if (LockUtil.tryLock(lockKey, lockValue, expireTime)) { 
            // 從數據庫取值并放到redis中 
            LockUtil.releaseLock(lockKey, lockValue); 
        } else { 
            // sleep一段時間再從緩存中拿 
            Thread.sleep(100); 
            getFromRedis(key); 
        } 
    } 
    return value; 
}