作者 |?Mark_MMXI

來源 | CSDN博客，責編 | 夕顏

出品 | CSDN（ID:CSDNnews）

緩存的存在是為了在高并發情形下，緩解DB壓力，提高業務系統體驗。業務系統訪問數據，先去緩存中進行查詢，假如緩存存在數據直接返回緩存數據，否則就去查詢數據庫再返回值。

Redis是一種緩存工具，是一種緩存解決方案，但是引入Redis又有可能出現緩存穿透、緩存擊穿、緩存雪崩等問題。本文就對緩存雪崩問題進行較深入剖析，并通過場景模型加深理解，基于場景使用對應的解決方案嘗試解決。

緩存原理及Redis解決方案

首先，我們來看一下緩存的工作原理圖：

Redis 本質上是一個 Key-Value 類型的內存數據庫。因為是純內存操作，Redis 的性能非常出色，每秒可以處理超過 10、萬次讀寫操作。Redis 還有一個優勢就是是支持保存多種數據結構，例如 String、List、Set、Sorted Set、hash等。

緩存雪崩

2.1 緩存雪崩解釋

緩存雪崩的情況是說，當某一時刻發生大規模的緩存失效的情況，比如你的緩存服務宕機了，DB直接負載大量請求壓力導致掛掉。

2.2 模擬緩存雪崩

按照緩存雪崩的解釋，其實我們要模擬，只需要達到以下幾個點：

同一時刻大規模緩存失效。

失效的時刻有大量的查詢請求沖擊DB

? ?

@Testpublic void testQuery(){ExecutorService es = Executors.newFixedThreadPool(10);int loop= 1000;int init=2000;//查詢1k個key放進緩存for (int i = init; i < loop+init; i++) {userService.queryById(i);}//緩存過期時間為1s,等待1s同時過期try {Thread.sleep(1000);}catch (Exception e){e.printStackTrace();}//開始了使用多線程瘋狂查詢for (int i = 0; i < 100; i++) {es.execute(() -> {for (int k = init; k < loop+init; k++) {userService.queryById(k);}});}}

為了加快崩壞的速度，把數據庫的最大連接數調整成5，同時增大數據庫表的數據量達到百萬級別。

然后執行測試程序，很快程序就報錯并停止，詳細錯誤如下：

Exception in thread "pool-1-thread-12" org.springframework.data.redis.RedisSystemException: Redis exception; nested exception is io.lettuce.core.RedisException: Connection is closed at org.springframework.data.redis.connection.lettuce.LettuceExceptionConverter.convert(LettuceExceptionConverter.java:74) at org.springframework.data.redis.connection.lettuce.LettuceExceptionConverter.convert(LettuceExceptionConverter.java:41) at org.springframework.data.redis.PassThroughExceptionTranslationStrategy.translate(PassThroughExceptionTranslationStrategy.java:44) at org.springframework.data.redis.FallbackExceptionTranslationStrategy.translate(FallbackExceptionTranslationStrategy.java:42) at org.springframework.data.redis.connection.lettuce.LettuceConnection.convertLettuceAccessException(LettuceConnection.java:270) at org.springframework.data.redis.connection.lettuce.LettuceStringCommands.convertLettuceAccessException(LettuceStringCommands.java:799) at org.springframework.data.redis.connection.lettuce.LettuceStringCommands.get(LettuceStringCommands.java:68) at org.springframework.data.redis.connection.DefaultedRedisConnection.get(DefaultedRedisConnection.java:260) at org.springframework.data.redis.cache.DefaultRedisCacheWriter.lambda$get$1(DefaultRedisCacheWriter.java:109) at org.springframework.data.redis.cache.DefaultRedisCacheWriter.execute(DefaultRedisCacheWriter.java:242) at org.springframework.data.redis.cache.DefaultRedisCacheWriter.get(DefaultRedisCacheWriter.java:109) at org.springframework.data.redis.cache.RedisCache.lookup(RedisCache.java:88) at org.springframework.cache.support.AbstractValueAdaptingCache.get(AbstractValueAdaptingCache.java:58) at org.springframework.cache.interceptor.AbstractCacheInvoker.doGet(AbstractCacheInvoker.java:73) at org.springframework.cache.interceptor.CacheAspectSupport.findInCaches(CacheAspectSupport.java:554) at org.springframework.cache.interceptor.CacheAspectSupport.findCachedItem(CacheAspectSupport.java:519) at org.springframework.cache.interceptor.CacheAspectSupport.execute(CacheAspectSupport.java:401) at org.springframework.cache.interceptor.CacheAspectSupport.execute(CacheAspectSupport.java:345) at org.springframework.cache.interceptor.CacheInterceptor.invoke(CacheInterceptor.java:61) at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:747) at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:689) at com.example.demo.user.service.impl.UserServiceImpl$$EnhancerBySpringCGLIB$$ba6638d2.queryById(<generated>) at com.example.demo.DemoApplicationTests$1.run(DemoApplicationTests.java:55) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Caused by: io.lettuce.core.RedisException: Connection is closed at io.lettuce.core.protocol.DefaultEndpoint.validateWrite(DefaultEndpoint.java:195) at io.lettuce.core.protocol.DefaultEndpoint.write(DefaultEndpoint.java:137) at io.lettuce.core.protocol.CommandExpiryWriter.write(CommandExpiryWriter.java:112) 2020-03-08 22:31:14.432 ERROR 37892 --- [eate-1895102622] com.alibaba.druid.pool.DruidDataSource : create connection SQLException, url: jdbc:mysql://localhost:3306/redis_demo?useUnicode=true&characterEncoding=UTF-8&allowMultiQueries=true&useJDBCCompliantTimezoneShift=true&useLegacyDatetimeCode=false&serverTimezone=UTC, errorCode 1040, state 08004java.sql.SQLNonTransientConnectionException: Data source rejected establishment of connection, message from server: "Too many connections"

主要問題出在數據庫連接已經滿了，無法獲取數據庫連接進行查詢，這個現象是就是緩存雪崩的效果。

‘

2.3 解決緩存雪崩

2.3.1 分析雪崩場景

用圖來說，實際上就是沒有了redis這層擔著上層流量壓力

其實從這張圖來看，對于我們一般的應用，客戶端去訪問應用到數據庫的整個鏈路過程，其實在面臨大流量的時候，我們一般是以"倒三角"模型進行流量緩沖，什么是“倒三角”模型

通過"倒三角"模型，按照并發需要優化系統，在面臨雪崩這種情形，可以按照“倒三角”模型進行優化，注意雪崩是理論上沒辦法徹底解決的，可能到最終得提高硬件配置。

2.3.1 雪崩優化方案

經過分析得解決雪崩方案：

1.隨機緩存過期時間，能一定程度緩解雪崩 2.使用鎖或隊列、設置過期標志更新緩存 3.添加本地緩存實現多級緩存 4.添加熔斷降級限流，緩沖壓力

2.3.1.1 隨機緩存時間

隨機緩存時間意在避免大量熱點key同時失效。

接下來，我們基于Redis+SpringBoot+SpringCache基礎項目搭建這個項目繼續進行實踐。

由于是使用了SpringCache，我們最優的方案就是直接在@Cacheable等注解上面加參數，比如像表達式之類的，讓數據放進緩存的時候按照表達式/參數值定義過期時間。

因此我們先查看原有的RedisCache是怎么樣的put邏輯

RedisCacheManager創建Cache

protected RedisCache createRedisCache(String name, @Nullable RedisCacheConfiguration cacheConfig) {return new RedisCache(name, this.cacheWriter, cacheConfig != null ? cacheConfig : this.defaultCacheConfig);}

打開RedisCache.class，查看put 方法如下：

public void put(Object key, @Nullable Object value) {Object cacheValue = this.preProcessCacheValue(value);if (!this.isAllowNullValues() && cacheValue == null) {throw new IllegalArgumentException(String.format("Cache '%s' does not allow 'null' values. Avoid storing null via '@Cacheable(unless=\"#result == null\")' or configure RedisCache to allow 'null' via RedisCacheConfiguration.", this.name));} else {this.cacheWriter.put(this.name, this.createAndConvertCacheKey(key), this.serializeCacheValue(cacheValue), this.cacheConfig.getTtl());}}

這里this.cacheConfig.getTtl() 就是緩存的過期時間，可以看到數據的緩存過期時間是從全局緩存配置里面獲取的過期時間配置的，而我需要實現的是讓某個cache下每個key隨機時間過期，因此我們需要改動這里 this.cacheConfig.getTtl(),我們在createRedisCache的時候改變這個值就行了。

1. 基于java動態執行字符串代碼，返回過期時間。

實現基于Spring.expression的ExpressService

/*** @title: ExpressUtil* @projectName redisdemo* @description: 動態執行字符串代碼* @author lps* @date 2020/3/912:01*/ @Slf4j public class ExpressService {private ExpressionParser spelExpressionParser;private ParserContext parserContext;// 表達式解析上下文private StandardEvaluationContext evaluationContext;public static enum ExpressType {/*** ${}表達式格式*/TYPE_FIRST,/*** #{}表達式格式*/TYPE_SECOND}private static final String PRE_TYPE_1 = "${";private static final String PRE_TYPE_2 = "#{";private static final String SUF_STR = "}";private ExpressService(String pre, String suf) {spelExpressionParser = new SpelExpressionParser();log.debug("表達式前綴={},表達式后綴={}", pre, suf);evaluationContext = new StandardEvaluationContext();// 增加map解析方案evaluationContext.addPropertyAccessor(new MapAccessor());parserContext = new TemplateParserContext(pre, suf);}/**** <p>* 創建表達式處理服務對象 默認為創建#{}格式表達式 通過ExpressType指定表達式格式，現有兩種${}和#{}* </p>*** @param type* 表達式格式類型* @return 表達式解析對象*/public static ExpressService createExpressService(ExpressType type) {if (type == ExpressType.TYPE_FIRST) {log.debug("生成表達式，表達式前綴={}", PRE_TYPE_1);return new ExpressService(PRE_TYPE_1, SUF_STR);} else {return new ExpressService(PRE_TYPE_2, SUF_STR);}}public Object expressParse(String express, Object data) throws Exception {log.debug("解析表達式信息={}", express);Expression expression = spelExpressionParser.parseExpression(express, this.parserContext);return expression.getValue(evaluationContext, data);}}

測試調用：

@Testpublic void testExpress(){ExpressService express = ExpressService.createExpressService(null);try {//固定超時時間System.out.println("ttl="+express.expressParse("#{60}", null));//調用方法生成隨機過期時間System.out.println("ttl="+express.expressParse("#{T(org.apache.commons.lang3.RandomUtils).nextInt(60,200)}", null));} catch (Exception e) {e.printStackTrace();}}

2. 設計name拼接ttl規則

由于createRedisCache只有兩個參數name以及cacheConfig，而只有name是對于單個cache來說的，cacheConfig是對于全局cache來說，因此我們需要設計name參數中指定cache的name以及過期時間的規則。

name賦值規則：name|ttlFun

eg: @Cacheable(cacheName="test|#{T(org.apache.commons.lang3.RandomUtils).nextInt(60,200)}")@Cacheable(cacheName="test|#{60}")

3. 編寫解析name代碼

??

/*** 分隔符|*/private static final String SEPERATE_LINE = "|";public MyRedisCacheManager(RedisCacheWriter cacheWriter, RedisCacheConfiguration defaultCacheConfiguration) {super(cacheWriter, defaultCacheConfiguration);}protected RedisCache createRedisCache(String name, @Nullable RedisCacheConfiguration cacheConfig) { // ``name賦值規則：name|ttlFun ``if(name.contains(SEPERATE_LINE)){String cacheName = name.substring(0,name.indexOf(SEPERATE_LINE));String expression = name.substring(name.indexOf(SEPERATE_LINE)+1);try{ExpressService express = ExpressService.createExpressService(null);long ttl = Long.parseLong(express.expressParse(expression, null).toString());cacheConfig = cacheConfig.entryTtl(Duration.ofSeconds(ttl));return super.createRedisCache(cacheName, cacheConfig);}catch (Exception e){e.printStackTrace();return super.createRedisCache(name, cacheConfig);}}return super.createRedisCache(name, cacheConfig);}

4. 修改CacheConfig

將原本的RedisManager替換成#3編寫的MyRedisManager

/*** 配置緩存管理器*/@Beanpublic CacheManager cacheManager(RedisConnectionFactory factory) {//關鍵點，spring cache 的注解使用的序列化都從這來，沒有這個配置的話使用的jdk自己的序列化，實際上不影響使用，只是打印出來不適合人眼識別RedisCacheConfiguration cacheConfig = RedisCacheConfiguration.defaultCacheConfig()// 將 key 序列化成字符串.serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(new StringRedisSerializer()))// 將 value 序列化成 json.serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(new GenericJackson2JsonRedisSerializer()))// 設置緩存過期時間，單位秒.entryTtl(Duration.ofSeconds(cacheExpireTime))// 不緩存空值.disableCachingNullValues(); /* RedisCacheManager redisCacheManager = RedisCacheManager.builder(factory).cacheDefaults(cacheConfig).build();*///修改RedisCacheManager 為MyRedisCacheManager MyRedisCacheManager redisCacheManager = new MyRedisCacheManager(RedisCacheWriter.nonLockingRedisCacheWriter(factory), cacheConfig);return redisCacheManager;}

5. 測試

編寫單元測試

@Testpublic void testQueryIdWithExpress(){Assert.assertNotNull(userService.queryById(3333));}

重新定義查詢的cache定義

@Override@Cacheable( value = "ca1|#{60}",key = "#id" ,unless="#result == null") // @Cacheable( value = "ca1|#{T(org.apache.commons.lang3.RandomUtils).nextInt(100,200)}",key = "#id" ,unless="#result == null")public User queryById(int id) {return this.userDao.queryById(id);}

當value=ca1|#{60}的時候，通過查看Redis的TTL 剩余為58s

當value=ca1|#{T(org.apache.commons.lang3.RandomUtils).nextInt(100,200)}的時候，隨機100-220范圍內秒數，通過查看Redis的TTL 剩余為107s

這時候使用random的方式就可以實現隨機過期時間了，隨機數最好選擇符合高斯(正態)分布的會比較好。

new Random().nextGaussian()

2.3.1.2 互斥鎖排隊

業界比價普遍的一種做法，即根據key獲取value值為空時，鎖上，從數據庫中load數據后再釋放鎖。若其它線程獲取鎖失敗，則等待一段時間后重試。這里要注意，分布式環境中要使用分布式鎖，單機的話用普通的鎖（synchronized、Lock）就夠了。

這樣做思路比較清晰，也從一定程度上減輕數據庫壓力，但是鎖機制使得邏輯的復雜度增加，吞吐量也降低了，有點治標不治本。

1.使用setnx的方式設置互斥鎖

??

public User queryById(int id) {try {if (redisTemplate.hasKey(id+"")) {return (User) redisTemplate.opsForValue().get(id+"");} else {//獲取鎖if(lock(id+"")){// 數據庫查詢User user = userDao.queryById(id);redisTemplate.opsForValue().set(id+"",user, Duration.ofSeconds(3000));//釋放鎖redisTemplate.delete(LOCK_PREFIX + "id");}}} catch (Exception e) {e.printStackTrace();}return (User) redisTemplate.opsForValue().get(""+id);}private static String LOCK_PREFIX = "prefix";private static long LOCK_EXPIRE =3000;/*** 互斥鎖實現*/public boolean lock(String key) {String lock = LOCK_PREFIX + key;return (Boolean) redisTemplate.execute((RedisCallback) connection -> {long expireAt = System.currentTimeMillis() + LOCK_EXPIRE + 1;//SETNXBoolean acquire = connection.setNX(lock.getBytes(), String.valueOf(expireAt).getBytes());if (acquire) {return true;} else {byte[] value = connection.get(lock.getBytes());if (Objects.nonNull(value) && value.length > 0) {long expireTime = Long.parseLong(new String(value)); //判斷鎖是否過期 if (expireTime < System.currentTimeMillis()) {byte[] oldValue = connection.getSet(lock.getBytes(), String.valueOf(System.currentTimeMillis() + LOCK_EXPIRE + 1).getBytes());return Long.parseLong(new String(oldValue)) < System.currentTimeMillis();}}}return false;});}

2.3.1.3 設置過期標志更新緩存

定時更新緩存，阻塞部分請求，達到緩沖作用，也可以設置key永不過期

2.3.1.4 多級緩存

這個方案主要在redis宕機，或者key在更新進緩存的中間，可以響應業務應用，減輕壓力

2.3.1.5 熔斷降級限流

這個方案是直接在業務應用之上進行請求流量控制，減輕下層壓力

原文鏈接：https://blog.csdn.net/qq_28540443/article/details/104746655

同時，歡迎所有開發者掃描下方二維碼填寫《開發者與AI大調研》，只需2分鐘，便可收獲價值299元的「AI開發者萬人大會」在線直播門票!

推薦閱讀：小網站的容器化(下)：網站容器化的各種姿勢，先跟著擼一波代碼再說！ 你知道嗎？其實 Oracle 直方圖自動統計算法存在這些缺陷！（附驗證步驟） 詳解以太坊虛擬機（EVM）的數據存儲機制 比特幣當贖金，WannaRen 勒索病毒二度來襲！平臺抗住日訪問量 7 億次，研發品控流程全公開“手把手撕LeetCode題目，扒各種算法套路的褲子”北京四環堵車引發的智能交通大構想從Ngin到Pandownload，程序員如何避免面向監獄編程？ 真香，朕在看了！

總結

以上是生活随笔為你收集整理的Redis 学习之一招击穿自己的系统，附送 N 个击穿解决大礼包 | 原力计划的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。