java緩存一致性問題及解決方案
點(diǎn)擊上方藍(lán)色字體，選擇“標(biāo)星公眾號”
優(yōu)質(zhì)文章，第一時(shí)間送達(dá)
讀取緩存步驟一般沒有什么問題，但是一旦涉及到數(shù)據(jù)更新：數(shù)據(jù)庫和緩存更新，就容 易出現(xiàn)緩存(Redis)和數(shù)據(jù)庫（MySQL）間的數(shù)據(jù)一致性問題。
一、討論一致性問題之前，先來看一個更新的操作順序問題：
先刪除緩存，再更新數(shù)據(jù)庫
問題：同時(shí)有一個請求 A 進(jìn)行更新操作，一個請求 B 進(jìn)行查詢操作?？赡艹霈F(xiàn):
（1）請求 A 進(jìn)行寫操作（key = 1 value = 2），先刪除緩存 key = 1 value = 1
（2）請求 B 查詢發(fā)現(xiàn)緩存不存在
（3）請求 B 去數(shù)據(jù)庫查詢得到舊值 key = 1 value = 1
（4）請求 B 將舊值寫入緩存 key = 1 value = 1
（5）請求 A 將新值寫入數(shù)據(jù)庫 key = 1 value = 2
緩存中數(shù)據(jù)永遠(yuǎn)都是臟數(shù)據(jù)
我們比較推薦操作順序：
先刪除緩存，再更新數(shù)據(jù)庫，再刪緩存（雙刪，第二次刪可異步延時(shí)）
public void write(String key,Object data){
redis.delKey(key);
db.updateData(data);
Thread.sleep(500);
redis.delKey(key);
}接下來，看一看緩存同步的一些方案，見下圖：
1、 數(shù)據(jù)實(shí)時(shí)同步更新
更新數(shù)據(jù)庫同時(shí)更新緩存，使用緩存工具類和或編碼實(shí)現(xiàn)。
優(yōu)點(diǎn)：數(shù)據(jù)實(shí)時(shí)同步更新，保持強(qiáng)一致性
缺點(diǎn)：代碼耦合，對業(yè)務(wù)代碼有侵入性
2、 數(shù)據(jù)準(zhǔn)實(shí)時(shí)更新
準(zhǔn)一致性，更新數(shù)據(jù)庫后，異步更新緩存，使用觀察者模式/發(fā)布訂閱/MQ 實(shí)現(xiàn)；
優(yōu)點(diǎn)：數(shù)據(jù)同步有較短延遲 ，與業(yè)務(wù)解耦
缺點(diǎn)：實(shí)現(xiàn)復(fù)雜，架構(gòu)較重
3 、緩存失效機(jī)制
弱一致性，基于緩存本身的失效機(jī)制
優(yōu)點(diǎn)：實(shí)現(xiàn)簡單，無須引入額外邏輯
缺點(diǎn)：有一定延遲，存在緩存擊穿/雪崩問題
4、 定時(shí)任務(wù)更新
最終一致性，采用任務(wù)調(diào)度框架，按照一定頻率更新
優(yōu)點(diǎn)：不影響正常業(yè)務(wù)
優(yōu)點(diǎn)：不保證一致性，依賴定時(shí)任務(wù)
二、 緩存擊穿、緩存雪崩及解決方案
1 、緩存擊穿
緩存擊穿是指緩存中沒有但數(shù)據(jù)庫中有的數(shù)據(jù)（一般是緩存時(shí)間到期），這時(shí)由于 并發(fā)用戶特別多，同時(shí)讀緩存沒讀到數(shù)據(jù)，又同時(shí)去數(shù)據(jù)庫去取數(shù)據(jù)，引起數(shù)據(jù)庫壓力
瞬間增大，造成過大壓力
2 、緩存雪崩
緩存雪崩是指緩存中數(shù)據(jù)大批量到過期時(shí)間，而查詢數(shù)據(jù)量巨大，引起數(shù)據(jù)庫壓 力過大甚至 down 機(jī)。和緩存擊穿不同的是，緩存擊穿指并發(fā)查同一條數(shù)據(jù)，緩存雪崩
是不同數(shù)據(jù)都過期了，很多數(shù)據(jù)都查不到從而查數(shù)據(jù)庫。
解決方案：
1）單體服務(wù)：此時(shí)需要對數(shù)據(jù)庫的查詢操作，加鎖 ---- lock （因考慮到是對同一個參數(shù)數(shù)值上 一把鎖，此處 synchronized 機(jī)制無法使用） 加鎖的標(biāo)準(zhǔn)流程代碼如下：
/**
* 解決緩存雪崩和擊穿方案
*/
@Service("provincesService")
public class ProvincesServiceImpl3 extends ProvincesServiceImpl implements ProvincesService{
private static final Logger logger = LoggerFactory.getLogger(ProvincesServiceImpl3.class);
@Resource
private CacheManager cm;//使用注解緩存
private ConcurrentHashMap<String, Lock> locks = new ConcurrentHashMap<>();//線程安全的
private static final String CACHE_NAME = "province";
public Provinces detail(String provinceid) {
// 1.從緩存中取數(shù)據(jù)
Cache.ValueWrapper valueWrapper = cm.getCache(CACHE_NAME).get(provinceid);
if (valueWrapper != null) {
logger.info("緩存中得到數(shù)據(jù)");
return (Provinces) (valueWrapper.get());
}
//2.加鎖排隊(duì)，阻塞式鎖---100個線程走到這里---同一個sql的取同一把鎖
doLock(provinceid);//32個省，最多只有32把鎖，1000個線程
try{//第二個線程進(jìn)來了
// 一次只有一個線程
//雙重校驗(yàn)，不加也沒關(guān)系，無非是多刷幾次庫
valueWrapper = cm.getCache(CACHE_NAME).get(provinceid);//第二個線程，能從緩存里拿到值？
if (valueWrapper != null) {
logger.info("緩存中得到數(shù)據(jù)");
return (Provinces) (valueWrapper.get());//第二個線程，這里返回
}
Provinces provinces = super.detail(provinceid);
// 3.從數(shù)據(jù)庫查詢的結(jié)果不為空，則把數(shù)據(jù)放入緩存中，方便下次查詢
if (null != provinces){
cm.getCache(CACHE_NAME).put(provinceid, provinces);
}
return provinces;
}catch(Exception e){
return null;
}finally{
//4.解鎖
releaseLock(provinceid);
}
}
private void releaseLock(String userCode) {
ReentrantLock oldLock = (ReentrantLock) locks.get(userCode);
//查詢鎖是否存在和查詢當(dāng)前線程是否保持此鎖
if(oldLock !=null && oldLock.isHeldByCurrentThread()){
oldLock.unlock();
}
}
private void doLock(String lockcode) {//給一個搜索條件，對應(yīng)一個鎖
//provinceid有不同的值，參數(shù)多樣化
//provinceid相同的，加一個鎖，---- 不是同一個key，不能用同一個鎖
ReentrantLock newLock = new ReentrantLock();//創(chuàng)建一個鎖
Lock oldLock = locks.putIfAbsent(lockcode, newLock);//若已存在，則newLock直接丟棄
if(oldLock == null){
newLock.lock();//首次加鎖，成功取鎖，執(zhí)行
}else{
oldLock.lock();//阻塞式等待取鎖
}
}
}
此場景下的鎖換成分布式鎖（redis或zk等）；同時(shí)設(shè)置多次取鎖功能；
/**
* 解決緩存雪崩和擊穿方案
*/
@Service("provincesService")
public class ProvincesServiceImpl5 extends ProvincesServiceImpl implements ProvincesService{
private static final Logger logger = LoggerFactory.getLogger(ProvincesServiceImpl3.class);
@Resource
private CacheManager cm;//使用注解緩存
@Autowired
private RedisTemplate<String, Object> redisTemplate;
private ConcurrentHashMap<String, Lock> locks = new ConcurrentHashMap<>();//線程安全的
private static final String CACHE_NAME = "province";
public Provinces detail(String provinceid) throws Exception{
// 1.從緩存中取數(shù)據(jù)
Cache.ValueWrapper valueWrapper = cm.getCache(CACHE_NAME).get(provinceid);
if (valueWrapper != null) {
logger.info("緩存中得到數(shù)據(jù)");
return (Provinces) (valueWrapper.get());
}
//2.加鎖排隊(duì)，阻塞式鎖---100個線程走到這里---同一個sql的取同一把鎖
//32個省，最多只有32把鎖，1000個線程
boolean flag=false;
flag = RedisUtil.setNX(provinceid, 3000);
//如果首次沒有取到鎖，可以取10次
if(!flag){
for(int i=0;i<10;i++){
Thread.sleep(200);
flag = RedisUtil.setNX(provinceid, 3000);//分布式鎖
if(flag){
break;
}
}
}
//如果首次沒有取到鎖，一直取直到取到為止
/* if(!flag){
for (;;){
Thread.sleep(200);
flag = RedisUtil.setNX(provinceid, 3000);//分布式鎖
if(flag){
break;
}
}
}*/
try{//第二個線程進(jìn)來了
// 一次只有一個線程
//雙重校驗(yàn)，不加也沒關(guān)系，無非是多刷幾次庫
valueWrapper = cm.getCache(CACHE_NAME).get(provinceid);//第二個線程，能從緩存里拿到值？
if (valueWrapper != null) {
logger.info("緩存中得到數(shù)據(jù)");
return (Provinces) (valueWrapper.get());//第二個線程，這里返回
}
Provinces provinces = super.detail(provinceid);
// 3.從數(shù)據(jù)庫查詢的結(jié)果不為空，則把數(shù)據(jù)放入緩存中，方便下次查詢
if (null != provinces){
cm.getCache(CACHE_NAME).put(provinceid, provinces);
}
return provinces;
}catch(Exception e){
return null;
}finally{
//4.解鎖
RedisUtil.releaseLock(provinceid);
}
}
}
版權(quán)聲明：本文為博主原創(chuàng)文章，遵循 CC 4.0 BY-SA 版權(quán)協(xié)議，轉(zhuǎn)載請附上原文出處鏈接和本聲明。
本文鏈接：
https://blog.csdn.net/nandao158/article/details/112757347
鋒哥最新SpringCloud分布式電商秒殺課程發(fā)布
??????
??長按上方微信二維碼 2 秒
感謝點(diǎn)贊支持下哈 