典型緩存案例
當(dāng)我們使用redis做緩存時(shí)一般步驟如下
- 請(qǐng)求進(jìn)來時(shí)候首先查詢r(jià)edis判斷是否存在緩存且緩存是否過期
- 若已經(jīng)存在不過期的緩存則直接獲取返回
- 若緩存不存在或已過期則重新查詢數(shù)據(jù)庫并將該數(shù)據(jù)存到redis中
代碼可以如下表示:
@Autowired
private RedisTemplate redisTemplate;
public List<String> getValueBySql(String key){
System.out.println("這里模擬從數(shù)據(jù)庫中獲取數(shù)據(jù)");
return new ArrayList<>();
}
public List<String> getCache(String key){
List<String> resultList = (List<String>)redisTemplate.opsForValue().get(key);
if(resultList == null || CollectionUtils.isEmpty(resultList)){
//若緩存不存在則從數(shù)據(jù)庫獲取并設(shè)置時(shí)間
resultList = getValueBySql(key);
redisTemplate.opsForValue().set(key, resultList, 1000, TimeUnit.SECONDS);
return resultList;
}else{
return resultList;
}
}
緩存擊穿
什么是緩存擊穿?
如上面的經(jīng)典緩存流程悍赢，在整個(gè)流程中我們需要先查詢r(jià)edis,在redis沒有的時(shí)候再去查數(shù)據(jù)庫最后再將數(shù)據(jù)庫返回的數(shù)據(jù)存到redis中决瞳。如果有一些非常經(jīng)常被訪問的數(shù)據(jù),例如一分鐘內(nèi)有超高的訪問請(qǐng)求货徙。試想一下剛某個(gè)熱點(diǎn)數(shù)據(jù)key在這個(gè)時(shí)刻過期。下一時(shí)刻有好幾個(gè)請(qǐng)求同時(shí)來請(qǐng)求key,這時(shí)候由于redisTemplate.opsForValue().get(key)為空皮胡，所有的數(shù)據(jù)必將直接訪問數(shù)據(jù)庫,這個(gè)時(shí)候大并發(fā)的請(qǐng)求可能會(huì)瞬間把后端DB壓垮
解決方案1: 使用synchronized+雙檢查機(jī)制
此方法適用于單機(jī)模式
/***
* synchronized + 雙重檢查機(jī)制
* @param key
* @return
*/
public List<String> getCacheSave(String key){
List<String> resultList = (List<String>)redisTemplate.opsForValue().get(key);
if(resultList == null || CollectionUtils.isEmpty(resultList)){
//采用synchronized保證一次只有一個(gè)請(qǐng)求進(jìn)入到這個(gè)代碼塊
synchronized (this){
resultList = (List<String>)redisTemplate.opsForValue().get(key);
if(CollectionUtils.isEmpty(resultList)){
return resultList;
}
resultList = getValueBySql(key);
redisTemplate.opsForValue().set(key, resultList, 1000, TimeUnit.SECONDS);
return resultList;
}
}else{
return resultList;
}
}
- 上面代碼第一個(gè)判斷保證在緩存有數(shù)據(jù)時(shí)痴颊，讓查詢緩存的請(qǐng)求不必排隊(duì)，減小了同步的粒度
- synchronized (this)保證查詢數(shù)據(jù)庫是同步操作屡贺，同一時(shí)刻只能有一個(gè)請(qǐng)求查詢數(shù)據(jù)庫
- 第二個(gè)判斷保證所有在redis有緩存時(shí),其他請(qǐng)求無需在查意思數(shù)據(jù)庫蠢棱。若沒有這個(gè)判斷，其他已經(jīng)等待synchronized 解鎖的請(qǐng)求會(huì)在請(qǐng)求一次數(shù)據(jù)庫
解決方案2:采用互斥鎖
適用于分布式模式
使用分布式鎖的方式甩栈。如圖泻仙，使用分布式鎖保證只有一個(gè)線程查詢數(shù)據(jù)庫,其他線程采用重試的方式進(jìn)行獲取
代碼參考如下
/***
*
* @param key
* @param retryCount 重試次數(shù)
* @return
* @throws InterruptedException
*/
public List<String> getCacheSave2(String key,int retryCount) throws InterruptedException {
List<String> resultList = (List<String>)redisTemplate.opsForValue().get(key);
if(CollectionUtils.isEmpty(resultList)){
final String mutexKey = key + "_lock";
boolean isLock = (Boolean) redisTemplate.execute(new RedisCallback() {
@Override
public Object doInRedis(RedisConnection connection) throws DataAccessException {
//只在鍵key不存在的情況下，將鍵key的值設(shè)置為value,若鍵key已經(jīng)存在量没，則 SETNX 命令不做任何動(dòng)作
//命令在設(shè)置成功時(shí)返回 1 玉转， 設(shè)置失敗時(shí)返回 0
return connection.setNX(mutexKey.getBytes(),"1".getBytes());
}
});
if(isLock){
//設(shè)置成1秒過期
redisTemplate.expire(mutexKey, 1000, TimeUnit.MILLISECONDS);
resultList = getValueBySql(key);
redisTemplate.opsForValue().set(key, resultList, 1000, TimeUnit.SECONDS);
redisTemplate.delete(mutexKey);
}else{
//線程休息50毫秒后重試
Thread.sleep(50);
retryCount--;
System.out.println("=====進(jìn)行重試，當(dāng)前次數(shù):" + retryCount);
if(retryCount == 0){
System.out.println("====這里發(fā)郵件或者記錄下獲取不到數(shù)據(jù)的日志殴蹄，并為key設(shè)置一個(gè)空置防止重復(fù)獲取");
List<String> list = Lists.newArrayList("no find");
redisTemplate.opsForValue().set(key, list, 1000, TimeUnit.SECONDS);
return list;
}
return getCacheSave2(key,retryCount);
}
}
return resultList;
}
解決方案3:提前設(shè)置鎖
這是網(wǎng)上看到的方案
https://carlosfu.iteye.com/blog/2269687
感覺還是采用分布式鎖的方式究抓，只不過是每次獲取的時(shí)候先獲取一下key的過期時(shí)間，如果過期時(shí)間快到了就提前重新設(shè)置下超時(shí)時(shí)間袭灯，并從數(shù)據(jù)庫中獲取最新的數(shù)據(jù)覆蓋
解決方案:資源保護(hù)
采用netflix的hystrix刺下，可以做資源的隔離保護(hù)主線程池(不懂,后面學(xué)習(xí)下)
緩存雪崩
什么是緩存雪崩？
緩存雪崩是指在我們?cè)O(shè)置緩存時(shí)采用了相同的過期時(shí)間稽荧，導(dǎo)致緩存在某一時(shí)刻同時(shí)失效橘茉，請(qǐng)求全部轉(zhuǎn)發(fā)到DB，DB瞬時(shí)壓力過重雪崩姨丈。
解決方案:在設(shè)置過期時(shí)間時(shí)加隨機(jī)值保證不同時(shí)失效
緩存失效時(shí)的雪崩效應(yīng)對(duì)底層系統(tǒng)的沖擊非侈囫可怕。大多數(shù)系統(tǒng)設(shè)計(jì)者考慮用加鎖或者隊(duì)列的方式保證緩存的單線程（進(jìn)程）寫构挤，從而避免失效時(shí)大量的并發(fā)請(qǐng)求落到底層存儲(chǔ)系統(tǒng)上髓介。這里分享一個(gè)簡單方案就時(shí)講緩存失效時(shí)間分散開，比如我們可以在原有的失效時(shí)間基礎(chǔ)上增加一個(gè)隨機(jī)值筋现，比如1-5分鐘隨機(jī)唐础，這樣每一個(gè)緩存的過期時(shí)間的重復(fù)率就會(huì)降低，就很難引發(fā)集體失效的事件
緩存擊穿
例如上面的經(jīng)典流程矾飞，如果我輸入一個(gè)不在我們規(guī)劃范圍的key一膨，也就是說這個(gè)key永遠(yuǎn)也查不到數(shù)據(jù)，則按照流程每次都要先去查數(shù)據(jù)庫洒沦，要是有人利用不存在的key頻繁攻擊我們的應(yīng)用豹绪，這就是漏洞。
解決方案1:設(shè)置白名單
設(shè)置key的白名單申眼，只有在白名單的key才能允許查詢(如果key的數(shù)量很多或key不是事先知道的情況下這種方式就不太好用)瞒津〔跻拢或者更高級(jí)點(diǎn)用布隆過濾器記錄所有可能的key，每次請(qǐng)求時(shí)進(jìn)行攔截
