Redis實戰（黑馬點評）——關于緩存（緩存更新策略、緩存穿透、緩存雪崩、緩存擊穿、Redis工具）

redis實現查詢緩存的業務邏輯?

?service層實現

@Overridepublic Result queryById(Long id) {String key = CACHE_SHOP_KEY + id;// 現查詢redis內有沒有數據String shopJson = (String) redisTemplate.opsForValue().get(key);if(StrUtil.isNotBlank(shopJson)){ // 如果redis的數據為存在，那么解析為對象// 將json轉為對象Shop shop = JSONUtil.toBean(shopJson, Shop.class);return Result.ok(shop);}// 如果不存在，就先查數據庫，再存入redisShop shop = getById(id);if(shop == null){return Result.fail("店鋪不存在");}// 存在就寫入redis，包括將對象轉為jsonredisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop));return Result.ok(shop);}

緩存更新策略?

緩存更新策略的最佳實踐方案：

低一致性需求：使用Redis自帶的內存淘汰機制
高一致性需求：主動更新，并以超時刪除作為處理方案

讀操作：（查詢）

緩存命中則直接返回
緩存未命中則查詢數據庫，并寫入緩存，設定超時時間

寫操作：（增刪改）

先寫數據庫，然后再刪除緩存
要確保數據庫與緩存操作的原子性

緩存穿透

緩存穿透產生的原因是什么？

用戶請求的數據在緩存中和數據庫中都不存在，不斷發起這樣的請求，給數據庫帶來巨大壓力。

緩存穿透的解決方案有哪些？

緩存null值
布隆過濾器
增強id的復雜度，避免被猜測id規律
做好數據的基本格式校驗
加強用戶權限校驗
做好熱點參數的限流

?緩存空對象的方法解決緩存穿透

@Override  
public Result queryById(Long id) {  // 使用店鋪ID構建緩存鍵  String key = CACHE_SHOP_KEY + id;  // 檢查店鋪信息是否已經緩存到Redis中  String shopJson = (String) redisTemplate.opsForValue().get(key);  // 如果緩存中存在數據，則將JSON字符串解析為Shop對象  if (StrUtil.isNotBlank(shopJson)) {   // 將JSON轉換為Shop對象  Shop shop = JSONUtil.toBean(shopJson, Shop.class);  return Result.ok(shop); // 返回店鋪對象作為成功結果  }  // 如果緩存中包含表示無數據的占位符，則返回錯誤信息  if ("#".equals(shopJson)) {  return Result.fail("沒有店鋪相關信息"); // 沒有店鋪信息可用  }  // 如果緩存中未找到店鋪數據，則查詢數據庫  Shop shop = getById(id);  // 如果數據庫中不存在該店鋪  if (shop == null) {  // 在緩存中存儲一個占位符，以表示該店鋪不存在  // 這可以防止對同一ID的進一步查詢再次訪問數據庫  redisTemplate.opsForValue().set(key, "#", CACHE_NULL_TTL, TimeUnit.MINUTES);  return Result.fail("店鋪不存在"); // 返回錯誤，指示店鋪不存在  }  // 如果找到店鋪，則將店鋪對象以JSON字符串的形式存儲到緩存中  redisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.MINUTES);  return Result.ok(shop); // 返回店鋪對象作為成功結果  
}

緩存雪崩

緩存擊穿?

緩存擊穿是指在高并發環境下，某個熱點數據的緩存同時失效，導致大量請求直接訪問數據庫，從而造成數據庫壓力驟增的現象。

緩存擊穿解決方案

解決方案	優點	缺點
互斥鎖	- 沒有額外的內存消耗 - 保證一致性 - 實現簡單	- 線程需要等待，性能受到影響 - 可能有死鎖風險
邏輯鎖	- 線程無需等待，性能較好	- 不保證一致性 - 有額外的內存消耗 - 實現復雜

解決方案

優點

缺點

互斥鎖

- 沒有額外的內存消耗

- 保證一致性

- 實現簡單

- 線程需要等待，性能受到影響

- 可能有死鎖風險

邏輯鎖

- 線程無需等待，性能較好

- 不保證一致性

- 有額外的內存消耗

- 實現復雜

基于互斥鎖解決緩存擊穿?

?基于互斥鎖解決緩存擊穿 + 緩存空對象的方法解決緩存穿透

// 查詢店鋪信息，使用互斥鎖解決緩存擊穿問題
public Shop queryWithMutex(Long id) {// 構造緩存的keyString key = CACHE_SHOP_KEY + id;// 1. 先從Redis中查詢店鋪信息String shopJson = (String) redisTemplate.opsForValue().get(key);// 如果Redis中存在緩存數據，直接解析JSON并返回對象if (StrUtil.isNotBlank(shopJson)) {// 將JSON字符串轉換為Shop對象Shop shop = JSONUtil.toBean(shopJson, Shop.class);return shop;}// 如果Redis中緩存的值為"#", 表示數據庫中沒有該店鋪信息if ("#".equals(shopJson)) {return null;}// 2. 構造互斥鎖的keyString lockKey = "lock:shop:" + id;// 定義店鋪對象Shop shop = null;try {// 嘗試獲取互斥鎖boolean isLock = tryLock(lockKey);// 如果獲取鎖失敗，線程休眠50ms后重試if (!isLock) {Thread.sleep(50); // 等待50msreturn queryWithMutex(id); // 遞歸調用，再次嘗試獲取鎖}// 3. 如果沒有獲取到緩存數據，查詢數據庫shop = getById(id);// 模擬數據庫查詢的延時，生產環境應該去掉這段代碼// Thread.sleep(200);// 4. 如果數據庫中沒有該店鋪信息if (shop == null) {// 在Redis中存儲一個特殊的標記值"#", 表示該店鋪不存在redisTemplate.opsForValue().set(key, "#", CACHE_NULL_TTL, TimeUnit.MINUTES);return null;}// 5. 如果數據庫中有數據，將店鋪信息存入Redis// 將Shop對象轉換為JSON字符串redisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.MINUTES);} catch (InterruptedException e) {// 捕獲線程中斷異常throw new RuntimeException(e);} finally {// 釋放互斥鎖unlock(lockKey);}// 返回查詢到的店鋪信息return shop;
}// 封裝獲取鎖，釋放鎖
// 嘗試獲取互斥鎖
private boolean tryLock(String key) {// 使用Redis的setIfAbsent方法嘗試設置鎖// 如果key不存在，則設置成功并返回true；否則返回falseBoolean flag = redisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);// 并設置一個過期時間（這里是 10 秒）return BooleanUtil.isTrue(flag);
}// 釋放互斥鎖
private void unlock(String key) {// 刪除鎖對應的keyredisTemplate.delete(key);
}

setIfAbsent?方法是 Redis 中的一種操作，用于設置一個鍵的值，僅在該鍵不存在的情況下進行設置。具體來說，它的功能如下：

鍵不存在時：如果指定的鍵（key）在 Redis 中不存在，則將其設置為指定的值（在這個例子中是?"1"），并可以指定該鍵的過期時間（這里是 10 秒）。此時，方法返回?true。
鍵已存在時：如果指定的鍵已經存在于 Redis 中，則不會進行任何操作，保持原有的值不變，方法返回?false。

基于邏輯鎖解決緩存擊穿?

 // 創建一個固定大小的線程池，用于緩存重建任務，避免頻繁創建線程帶來的開銷private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);/*** 查詢商鋪信息，考慮邏輯過期* @param id 商鋪ID* @return 商鋪信息，如果不存在或已過期則返回null*/public Shop queryWithLogicalExpire(Long id){// 構建緩存的key，用于從Redis中查詢對應的商鋪信息String key = CACHE_SHOP_KEY + id;// 1. 從Redis查詢商鋪緩存，獲取商鋪信息的JSON字符串String shopJson = (String) redisTemplate.opsForValue().get(key);// 2. 判斷緩存是否存在if (StrUtil.isBlank(shopJson)) {// 3. 緩存不存在，直接返回nullreturn null;}// 4. 緩存命中，需要先將JSON字符串反序列化為RedisData對象RedisData redisData = JSONUtil.toBean(shopJson, RedisData.class);// 從RedisData對象中提取商鋪信息，并將其反序列化為Shop對象Shop shop = JSONUtil.toBean((JSONObject) redisData.getData(), Shop.class);// 獲取緩存的過期時間LocalDateTime expireTime = redisData.getExpireTime();// 5. 判斷緩存是否過期if (expireTime.isAfter(LocalDateTime.now())) {// 5.1. 緩存未過期，直接返回商鋪信息return shop;}// 構建鎖的key，用于控制緩存重建的并發訪問String lockKey = LOCK_SHOP_KEY + id;// 嘗試獲取鎖，確保緩存重建操作的線程安全boolean isLock = tryLock(lockKey);// 6.2. 判斷是否成功獲取鎖if (isLock) {// 6.3. 成功獲取鎖，開啟獨立線程進行緩存重建，避免阻塞主線程CACHE_REBUILD_EXECUTOR.submit(() -> {try {// 重建緩存，將最新的商鋪信息保存到Redis中，并設置過期時間this.saveShop2Redis(id, 20L); // 假設20L是過期時間，單位為秒} catch (Exception e) {// 如果在緩存重建過程中發生異常，拋出運行時異常，并記錄日志throw new RuntimeException(e);} finally {// 無論緩存重建成功與否，都需要釋放鎖，避免死鎖unlock(lockKey);}});}// 返回當前查詢到的商鋪信息（可能已過期）return shop;}public void saveShop2Redis(Long id, Long expireSeconds) {// 1. 查詢店鋪數據Shop shop = getById(id);// 2. 封裝邏輯過期時間RedisData redisData = new RedisData();redisData.setData(shop);redisData.setExpireTime(LocalDateTime.now().plusSeconds(expireSeconds));// 3. 寫入RedisredisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, JSONUtil.toJsonStr(redisData));}

RedisData類

@Data
public class RedisData {private LocalDateTime expireTime;private Object data;
}

?封裝redis工具類

    @Resourceprivate CacheClient cacheClient;public static final String CACHE_SHOP_KEY = "cache:shop:";public static final Long CACHE_SHOP_TTL = 30L;@Overridepublic Result queryById(Long id)  {// 解決緩存穿透
//        Shop shop = cacheClient // 傳入一個從數據庫內獲取Shop對象的函數：this::getById
//                .queryWithPassThrough(CACHE_SHOP_KEY, id, Shop.class, this::getById, CACHE_SHOP_TTL, TimeUnit.MINUTES);// 互斥鎖解決緩存擊穿Shop shop = cacheClient.queryWithMutex(CACHE_SHOP_KEY, id, Shop.class, this::getById, CACHE_SHOP_TTL, TimeUnit.MINUTES);// 邏輯過期解決緩存擊穿
//        Shop shop = cacheClient
//                .queryWithLogicalExpire(CACHE_SHOP_KEY, id, Shop.class, this::getById, CACHE_SHOP_TTL, TimeUnit.MINUTES);if(shop == null){return Result.fail("店鋪不存在");}return Result.ok(shop);}

1.?`queryWithPassThrough`

這個方法用于處理緩存穿透問題。緩存穿透是指查詢一個數據庫中不存在的數據，由于緩存中也沒有這個數據，所以每次查詢都會直接打到數據庫上，增加數據庫的壓力。

參數：
- keyPrefix：緩存的前綴。
- id：緩存的ID。
- type：返回對象的類型。
- dbFallback：數據庫查詢的回調函數。
- time：緩存時間。
- unit：時間單位。

2.?`queryWithLogicalExpire`

這個方法用于處理緩存擊穿問題。緩存擊穿是指一個緩存中非常熱門的數據突然過期，導致大量請求同時打到數據庫上，增加數據庫的壓力。

參數：與 queryWithPassThrough 相同。

最大的缺點是運行前要把所有緩存加到redis內，不然怎么查都是null

3.?`queryWithMutex（互斥鎖）`

這個方法結合了 queryWithPassThrough 和 queryWithLogicalExpire 的功能，用于處理緩存穿透和緩存擊穿問題。

參數：與 queryWithPassThrough 相同。

@Slf4j
@Component
public class CacheClient {public static final Long CACHE_NULL_TTL = 2L;public static final String LOCK_SHOP_KEY = "lock:shop:";private final RedisTemplate redisTemplate;private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);public CacheClient(RedisTemplate redisTemplate) {this.redisTemplate = redisTemplate;}public void set(String key, Object value, Long time, TimeUnit unit) {redisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(value), time, unit);}public void setWithLogicalExpire(String key, Object value, Long time, TimeUnit unit) {// 設置邏輯過期RedisData redisData = new RedisData();redisData.setData(value);redisData.setExpireTime(LocalDateTime.now().plusSeconds(unit.toSeconds(time)));// 寫入RedisredisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(redisData));}public <R,ID> R queryWithPassThrough(String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit){String key = keyPrefix + id;// 1.從redis查詢商鋪緩存String json = (String) redisTemplate.opsForValue().get(key);// 2.判斷是否存在if (StrUtil.isNotBlank(json)) {// 3.存在，直接返回return JSONUtil.toBean(json, type);}// 判斷命中的是否是空值if (json != null) {// 返回一個錯誤信息return null;}// 4.不存在，根據id查詢數據庫R r = dbFallback.apply(id);// 5.不存在，返回錯誤if (r == null) {// 將空值寫入redisredisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.MINUTES);// 返回錯誤信息return null;}// 6.存在，寫入redisthis.set(key, r, time, unit);return r;}public <R, ID> R queryWithLogicalExpire(String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit) {String key = keyPrefix + id;// 1.從redis查詢商鋪緩存String json = (String) redisTemplate.opsForValue().get(key);// 2.判斷是否存在if (StrUtil.isBlank(json)) {// 3.存在，直接返回return null;}// 4.命中，需要先把json反序列化為對象RedisData redisData = JSONUtil.toBean(json, RedisData.class);R r = JSONUtil.toBean((JSONObject) redisData.getData(), type);LocalDateTime expireTime = redisData.getExpireTime();// 5.判斷是否過期if(expireTime.isAfter(LocalDateTime.now())) {// 5.1.未過期，直接返回店鋪信息return r;}// 5.2.已過期，需要緩存重建// 6.緩存重建// 6.1.獲取互斥鎖String lockKey = LOCK_SHOP_KEY + id;boolean isLock = tryLock(lockKey);// 6.2.判斷是否獲取鎖成功if (isLock){// 6.3.成功，開啟獨立線程，實現緩存重建CACHE_REBUILD_EXECUTOR.submit(() -> {try {// 查詢數據庫R newR = dbFallback.apply(id);// 重建緩存this.setWithLogicalExpire(key, newR, time, unit);} catch (Exception e) {throw new RuntimeException(e);}finally {// 釋放鎖unlock(lockKey);}});}// 6.4.返回過期的商鋪信息return r;}public <R, ID> R queryWithMutex(String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit) {String key = keyPrefix + id;// 1.從redis查詢商鋪緩存String shopJson = (String) redisTemplate.opsForValue().get(key);// 2.判斷是否存在if (StrUtil.isNotBlank(shopJson)) {// 3.存在，直接返回return JSONUtil.toBean(shopJson, type);}// 判斷命中的是否是空值if (shopJson != null) {// 返回一個錯誤信息return null;}// 4.實現緩存重建// 4.1.獲取互斥鎖String lockKey = LOCK_SHOP_KEY + id;R r = null;try {boolean isLock = tryLock(lockKey);// 4.2.判斷是否獲取成功if (!isLock) {// 4.3.獲取鎖失敗，休眠并重試Thread.sleep(50);return queryWithMutex(keyPrefix, id, type, dbFallback, time, unit);}// 4.4.獲取鎖成功，根據id查詢數據庫r = dbFallback.apply(id);// 5.不存在，返回錯誤if (r == null) {// 將空值寫入redisredisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.MINUTES);// 返回錯誤信息return null;}// 6.存在，寫入redisthis.set(key, r, time, unit);} catch (InterruptedException e) {throw new RuntimeException(e);}finally {// 7.釋放鎖unlock(lockKey);}// 8.返回return r;}private boolean tryLock(String key) {Boolean flag = redisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);return BooleanUtil.isTrue(flag);}private void unlock(String key) {redisTemplate.delete(key);}
}