1.0 什么是缓存
2.0 项目中具体如何添加缓存
3.0 添加缓存后所带来的问题
3.1 读写不一致问题
3.1.1 缓存更新策略
3.1.2 具体实现缓存与数据库的双写一致
3.2 缓存穿透问题
3.2.1 具体解决缓存穿透问题
3.3 缓存雪崩问题
3.4 缓存击穿问题
3.4.1 利用互斥锁解决缓存击穿问题
3.4.2 利用逻辑过期解决缓存击穿问题
4.0 封装 Redis 工具类
缓存就是数据交换的缓冲器,称作为 Cache,是存放数据的零时地方,一般读写性能较高。缓存的作用可以降低后端负载,提高读写效率、降低响应时间。缓存的成本包括数据一致性成本、代码维护成本、运维成本等。
举例子,在实现根据用户 id 来查询用户信息的功能中,添加缓存的步骤:
首先,提交用户 id ,先从缓存中查找是否命中目标,就是是否有相同的 id 关键字 key 。如果命中,直接返回该 key 对应的 value 即可;如果没有命中,就需要来到数据库中查询用户信息,继续判断数据库中是否存在该用户 id ,如果不存在,那么返回报错信息;如果存在,那么返回该用户信息的同时,将用户信息写回到 Redis 缓存中。
缓存作用模型图:
代码实现:
@Autowired StringRedisTemplate stringRedisTemplate; @Override public String getUserNameById(Integer userId) throws Exception { //先判断userId是否为空 if (userId == null){ throw new Exception("userId is null"); } //先从缓存中查看是否存在该key String s = stringRedisTemplate.opsForValue().get("user:" + userId); if (s != null){ //如果缓存中不为null,则成功从缓存中获取值 return s; } //如果从缓存中获取不到,则需要到数据库中获取数据 String userName = adminMapper.getUserNameById(userId); //如果数据返回为null,那么数据库中查找不到数据 if (userName == null){ //直接抛出异常 throw new Exception("根据该用户id查找不到用户信息"); } //判断数据不为null之后,则需要将该用户信息写到redis中 stringRedisTemplate.opsForValue().set("user:"+userId,userName); //最后返回值即可 return userName; }
运行结果:
在第一次查询的时候,redis 第一次时找不到该用户信息,那么就会到数据库中查询,查询完毕之后,将数据写回到 redis 中,再到第二次查询的时候,就可以直接到 redis 中获取数据了。
发送的请求:
第一次获取数据:
到数据库中获取了
此时 redis 中:
已经存在该用户信息了
添加缓存之后,会带来一些问题,比如说:数据库更新之后,缓存还没来得及更新所带来的缓存与数据库数据不一致问题,还有缓存穿透、缓存雪崩、缓存击穿等问题给数据库带来的沉重的“打击”。
顾名思义,数据库与缓存中的数据两者不一致,为了解决这个问题,就有了缓存更新策略,可以极大可能维护缓存中的数据和数据库中的数据一致性。
通常的方法有三种:
1)内存淘汰:不用自己维护,利用 redis 的内存淘汰机制,当内存不足自动淘汰部分数据,下次查询时更新缓存。该方法一致性比较差,无维护成本。
2)超时剔除:给缓存数据添加 TTL 时间,到期后自动删除缓存,下次查询时更新缓存。该方法一致性一般,维护成本低。
3)主动更新:
编写业务逻辑,在修改数据库的同时,更新缓存。该方法一致性比较好,维护成本高。主动更新包含三种常见的策略:
第一种:Cache Aside Pattern:由缓存的调用者,在更新数据库的同时更新缓存。
第二种:Read/Write Through Pattern:缓存与数据库整合一个服务,由服务来维护一致性。调用者调用该服务,无需关心缓存一致性问题。
第三种:Write Behind Caching Pattern:调用者只操作缓存,由其他线程异步的将缓存数据持久化到数据库,保证最终一致。
在主动更新中,第一种方式比较常见,实现比较简单。但是在操作缓存和数据库时有三个问题需要考虑:
第一个问题:删除缓存还是更新缓存?
更新缓存:每次更新数据库都更新缓存,无效写操作较多。
删除缓存:更新数据库时让缓存失效,查询时在更新缓存。
因此,一般来说,选择删除缓存。
第二个问题:如何保证缓存与数据库的操作的同时成功或失败?
将缓存与数据库操作放在同一个事务即可,保证其原子性。
第三个问题:先操作缓存还是先操作数据库?
先写数据库,然后删除缓存。
缓存更新策略的最佳实践方案:
1)低一致性需求:使用内存淘汰机制。例如店铺类型的查询缓存。
2)高一致性需求:主动更新,并以超时剔除作为兜底方案。
读操作:
缓存命中则直接返回,缓存未命中则查询数据库,并写入缓存,设定超时时间。
写操作:
先写数据库,然后再删除缓存,要确保数据库与缓存操作的原子性。
实现高一致性需求:主动更新策略代码:
1)读操作:缓存命中则直接返回,缓存未命中则查询数据库,并写入缓存,设定超时时间。
@Autowired StringRedisTemplate stringRedisTemplate; @Override public String getUserNameById(Integer userId) throws Exception { //先判断userId是否为空 if (userId == null){ throw new Exception("userId is null"); } //先从缓存中查看是否存在该key String s = stringRedisTemplate.opsForValue().get("user:" + userId); if (s != null){ //如果缓存中不为null,则成功从缓存中获取值 return s; } //如果从缓存中获取不到,则需要到数据库中获取数据 String userName = adminMapper.getUserNameById(userId); //如果数据返回为null,那么数据库中查找不到数据 if (userName == null){ //直接抛出异常 throw new Exception("根据该用户id查找不到用户信息"); } //判断数据不为null之后,则需要将该用户信息写到redis中,且设定超时时间 stringRedisTemplate.opsForValue().set("user:"+userId,userName,100, TimeUnit.SECONDS); //最后返回值即可 return userName; }
这里的重点是:设置超时时间。
2)写操作:先写数据库,然后再删除缓存,要确保数据库与缓存操作的原子性。
为了保证原子性,需要加上 @Transactional 注解
@Override @Transactional public void modifyUser(UserDTO userDTO) throws Exception { //先判断userDTO是否为null if (userDTO == null){ throw new Exception("userDTO is null"); } //先更新数据库 adminMapper.modifyUser(userDTO); //再删除redis缓存 Integer userId = userDTO.getUserId(); stringRedisTemplate.delete("user"+userId); }
运行结果:
先查询用户信息,因为第一次 redis 不存在该用户信息,因此需要到数据库中获取该用户信息。
从数据库中查询信息:
redis 缓存情况:
接着去更新用户信息:
此时,redis 中的用户信息就被删除掉了:
下一次查询就需要到数据库中查询了。
再一次查询:
会到数据库中查询用户信息。
是指客户端请求的数据在缓存中和数据库中都不存在,这样缓存永远都不会生效,这些请求都会打到数据库。则会给数据库的压力非常大,因此需要解决这种情况发生。
常见的解决方案有四种:
1)增强 id 的复杂度,避免被猜测 id 规律。
2)做好数据的基础格式校验。
3)缓存空对象:实现简单,维护方便。
该方法的缺点:额外的内存消耗,因为设置 key 对应的 value 为 null ,占用了一定的缓存空间,因此为了减少内存浪费,会设置缓存时间 TTL ;还可能造成短期的不一致,当数据库中 key 有对应的 value 了,当前的 key 还在缓存中,value 还是为 null ,所以造成一定的不一致性。
4)布隆过滤:内存占用较少,没有多余 key ,该方法的缺点为实现复杂,存在误判的可能。
使用缓存空对象来解决缓存穿透问题步骤:
首先,从缓存中查询用户,判断缓存是否命中,如果命中,则直接返回用户信息;如果没有命中,根据用户 id 到数据库中查询用户信息,如果用户信息不为 null ,则说明用户信息是存在的,那么将用户信息写回到缓存中,方便下一次查询可以直接从缓存中获取用户信息;如果用户信息为 null ,则说明数据库中也不存在该用户信息,那么下一次就不需要继续查询该用户信息了,让其在缓存中查询,再抛出异常即可。
具体的流程图:
代码如下:
@Autowired StringRedisTemplate stringRedisTemplate; @Override public String getUserNameById(Integer userId) throws Exception { //先判断userId是否为空 if (userId == null){ throw new Exception("userId is null"); } //先从缓存中查看是否存在该key String s = stringRedisTemplate.opsForValue().get("user:" + userId); if (StrUtil.isNotBlank(s)){ //如果缓存中不为null,则成功从缓存中获取值 return s; } if (s != null){ //直接抛出异常 throw new Exception("该用户信息不存在!"); } //如果从缓存中获取不到,则需要到数据库中获取数据 String userName = adminMapper.getUserNameById(userId); //如果数据返回为null,那么数据库中查找不到数据 if (userName == null){ //如果在数据库中找不到该信息,则将该 key 值对应的 value 为 "" 写到缓存中 stringRedisTemplate.opsForValue().set("user:"+userId,"",100,TimeUnit.SECONDS); } //判断数据不为null之后,则需要将该用户信息写到redis中,且设定超时时间 stringRedisTemplate.opsForValue().set("user:"+userId,userName,100, TimeUnit.SECONDS); //最后返回值即可 return userName; }
运行结果:
查询数据库不存在的用户信息:
第一次会到数据库查询该用户信息,当该用户信息不存在时,则会在 redis 中设置空值,这样的好处,下一次的查询该用户,就不会打到数据库中了,减少了数据库的压力。
是指在同一时间段大量的缓存 key 同时失效或者 Redis 服务宕机,导致大量请求到达数据库,带来巨大压力。
3.3.1 解决缓存雪崩方案
1)给不同的 key 的 TTL 添加随机值。
2)利用 Redis 集群提高服务的可用性。
3)给缓存业务添加降级限流策略。
4)给业务添加多级缓存。
缓存击穿问题也叫热点 Key 问题,就是一个被高并发访问并且缓存重建业务交复杂的 Key 突然失效了,无数的请求访问会在瞬间给数据库带来巨大的冲击。
如图:
常见的解决方法:
1)利用互斥锁解决击穿问题
没有额外的内存消耗,保证一致性,实现简单。该方法的缺点:线程需要等待,性能受影响,可能有死锁的风险。
2)利用逻辑过期解决缓存击穿问题
线程无需等待,性能较好。该方法的缺点,不保证一致性,有额外的内存消耗,实现复杂。
利用互斥锁解决的步骤:
首先,查询缓存是否命中,如果命中,直接返回;如果没有命中,则需要判断是否能获取互斥锁,如果获取到了互斥锁,则查询数据库重建缓冲数据,最后释放锁,再返回数据;如果没有获取互斥锁,则休眠一段时间,再重试,直到从缓冲中获取到数据返回。
流程图:
代码如下:
解决缓存穿透与缓存击穿:
//解决缓存穿透与缓存击穿 public String getUserNameById2(Integer userId) throws Exception { //先判断userId是否为空 if (userId == null){ throw new Exception("userId is null"); } //先从缓存中查看是否存在该key String s = stringRedisTemplate.opsForValue().get("user:" + userId); if (StrUtil.isNotBlank(s)){ //如果缓存中不为null,则成功从缓存中获取值 return s; } if (s != null){ //直接抛出异常 throw new Exception("该用户信息不存在!"); } //如果从缓存中获取不到,则需要到数据库中获取数据 //判断释放可以获取到锁 String lock = "getLock"; String userName = null; try { boolean b = tryLock(lock); if (!b) { //如果没有获取到锁,休眠一会,再重新从缓存中获取数据 Thread.sleep(50); return getUserNameById2(userId); } userName = adminMapper.getUserNameById(userId); //如果数据返回为null,那么数据库中查找不到数据 if (userName == null){ //如果在数据库中找不到该信息,则将该 key 值对应的 value 为 "" 写到缓存中 stringRedisTemplate.opsForValue().set("user:"+userId,"",100,TimeUnit.SECONDS); } //判断数据不为null之后,则需要将该用户信息写到redis中,且设定超时时间 stringRedisTemplate.opsForValue().set("user:"+userId,userName,100, TimeUnit.SECONDS); } catch (Exception e) { throw new RuntimeException(e); } finally { //释放锁 unlock(lock); } //返回值即可 return userName; } //获取锁 private boolean tryLock(String key){ Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 200, TimeUnit.SECONDS); return BooleanUtil.isTrue(aBoolean); } //释放锁 private void unlock(String key){ stringRedisTemplate.delete(key); }
设置缓存中 key 的逻辑过期,顾名思义:在实际上,缓存中的 key 是设置永远不过期,将其添加过期字段,通过查看该字段,来判断该 key 在缓存中是否已经过期了。
利用逻辑过期解决缓存击穿问题步骤:
首先,判断缓存是否命中,如果没有命中,则返回空;如果命中,继续判断该字段是否过期,如果没有过期,则直接获取并且返回该值;如果已经过期,再继续判断能否获取锁,如果获取锁失败,则直接返回已经过期的值;如果获取锁成功,创建一个线程来做查询数据库,并且写入到缓存中,对于主线程来说,仍然返回旧的数据。
流程图:
代码实现:
利用逻辑过期实现解决缓存击穿问题:
//获取锁 private boolean tryLock(String key){ Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 200, TimeUnit.SECONDS); return BooleanUtil.isTrue(aBoolean); } //释放锁 private void unlock(String key){ stringRedisTemplate.delete(key); } //解决缓存穿透 public String getUserNameById(Integer userId) throws Exception { //先判断userId是否为空 if (userId == null){ throw new Exception("userId is null"); } //先从缓存中查看是否存在该key String s = stringRedisTemplate.opsForValue().get("user:" + userId); //如果从缓存中没有获取到数据,则直接抛出异常 if (s == null){ throw new Exception("该用户不存在!!!"); } //反序列化 RedisData redisData = JSON.parseObject(s, RedisData.class); String data = (String) redisData.getData(); LocalDateTime localDateTime = redisData.getLocalDateTime(); //判断是否过期 if (localDateTime.isAfter(LocalDateTime.now())){ //如果没有过期,则直接返回数据 return data; } //创建线程池 ExecutorService pool = Executors.newFixedThreadPool(10); //如果过期了 //判断能否获取到互斥锁 String lock = "getLock"; boolean b = tryLock(lock); if (b) { //获取到锁,从线程池中获取一个线程来从数据库获取信息,再将信息写入到缓存中 pool.submit(() -> { try { //先从数据库中获取到数据 String userName = adminMapper.getUserNameById(userId); //再将数据写入到缓存中 RedisData red = new RedisData(); //设置过期时间 red.setLocalDateTime(LocalDateTime.now().plusSeconds(100L)); red.setData(userName); //将其序列化 String jsonString = JSON.toJSONString(red); stringRedisTemplate.opsForValue().set("user:"+userId,jsonString); } catch (Exception e) { throw new RuntimeException(e); } finally { //释放锁 unlock(lock); } }); } //最后返回 return data; }
基于 StringRedisTemplate 封装一个缓存工具类,满足下列需要:
1)方法1:将任意 Java 对象序列化为 json 并存储在 string 类型的 key 中,并且可以设置 TTL 过期时间。
代码如下:
public void set(String key, Object value, Long time, TimeUnit timeUnit){ stringRedisTemplate.opsForValue().set(key, JSON.toJSONString(value),time,timeUnit); }
2)方法2:将任意 Java 对象序列化为 json 并存储在 string 类型的 key 中,并且可以设置逻辑过期时间,用于处理缓存击穿问题。
代码如下:
public void setWithLogicalExpire(String key,Object value,Long time,TimeUnit timeUnit){ RedisData redisData = new RedisData(); redisData.setData(value); redisData.setLocalDateTime(LocalDateTime.now().plusSeconds(timeUnit.toSeconds(time))); String jsonString = JSON.toJSONString(redisData); stringRedisTemplate.opsForValue().set(key,jsonString); }
3)方法3:根据指定的 key 查询缓存,并反序列化为指定类型,利用缓存空值的方式解决穿透问题。
//利用缓存空值解决缓存穿透 public <R,ID> R queryWithPassThrough(String prefix,ID id,Class<R> type,Function<ID,R> function,Long time,TimeUnit unit ){ String key = prefix + id; //判断在缓存中是否能命中 String jsonString = stringRedisTemplate.opsForValue().get(key); if (StrUtil.isNotBlank(jsonString)){ //反序列化 return JSON.parseObject(jsonString, type); } if (jsonString != null){ return null; } //查询数据库,且将数据信息写入到缓存中 R apply = function.apply(id); //判断是否为空值 if (apply == null){ //如果为空 //将其写进缓存中 stringRedisTemplate.opsForValue().set(key,"",50,TimeUnit.SECONDS); return null; } //序列化 String json = JSON.toJSONString(apply); //如果不为空 stringRedisTemplate.opsForValue().set(key,json,time,unit); return apply; }
4)方法4:根据指定的 key 查询缓存,并反序列为指定类型,需要利用逻辑过期解决缓存击穿问题。
//利用逻辑过期解决缓存击穿 public <R,ID> R queryWithLogicalExpire(String prefix,ID id,Class<R> type,Function<ID,R> function,Long time,TimeUnit unit){ String key = prefix + id; //判断在缓存中是否命中 String s = stringRedisTemplate.opsForValue().get(key); //如果不存在,直接返回null if (s == null){ return null; } //如果存在,还得判断是否过期 //反序列化 RedisData redisData = JSONUtil.toBean(s, RedisData.class); JSONObject d = (JSONObject) redisData.getData(); R data = JSONUtil.toBean(d, type); LocalDateTime localDateTime = redisData.getLocalDateTime(); if (localDateTime.isAfter(LocalDateTime.now())){ //如果没有过期 //直接返回数据 return data; } //创建线程池 ExecutorService pool = Executors.newFixedThreadPool(10); //过期了,判断是否可以获取锁 String lock = "getLock"; boolean b = tryLock(lock); if (b){ //如果获取锁成功, pool.submit(() -> { //从数据库中获取数据,再将数据写回缓存中 try { R apply = function.apply(id); setWithLogicalExpire(key,apply,time,unit); } catch (Exception e) { throw new RuntimeException(e); } finally { //释放锁 unlock(lock); } }); } return data; }
5)完整 Redis 的工具类
import cn.hutool.core.util.BooleanUtil; import cn.hutool.core.util.StrUtil; import cn.hutool.json.JSONObject; import cn.hutool.json.JSONUtil; import com.alibaba.fastjson.JSON; import com.example.bookproject20.pojo.RedisData; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Component; import java.time.LocalDateTime; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.function.Function; @Component public class CacheClient { private final StringRedisTemplate stringRedisTemplate; public CacheClient(StringRedisTemplate stringRedisTemplate){ this.stringRedisTemplate = stringRedisTemplate; } public void set(String key, Object value, Long time, TimeUnit timeUnit){ stringRedisTemplate.opsForValue().set(key, JSON.toJSONString(value),time,timeUnit); } public void setWithLogicalExpire(String key,Object value,Long time,TimeUnit timeUnit){ RedisData redisData = new RedisData(); redisData.setData(value); redisData.setLocalDateTime(LocalDateTime.now().plusSeconds(timeUnit.toSeconds(time))); String jsonString = JSON.toJSONString(redisData); stringRedisTemplate.opsForValue().set(key,jsonString); } //利用缓存空值解决缓存穿透 public <R,ID> R queryWithPassThrough(String prefix,ID id,Class<R> type,Function<ID,R> function,Long time,TimeUnit unit ){ String key = prefix + id; //判断在缓存中是否能命中 String jsonString = stringRedisTemplate.opsForValue().get(key); if (StrUtil.isNotBlank(jsonString)){ //反序列化 return JSON.parseObject(jsonString, type); } if (jsonString != null){ return null; } //查询数据库,且将数据信息写入到缓存中 R apply = function.apply(id); //判断是否为空值 if (apply == null){ //如果为空 //将其写进缓存中 stringRedisTemplate.opsForValue().set(key,"",50,TimeUnit.SECONDS); return null; } //序列化 String json = JSON.toJSONString(apply); //如果不为空 stringRedisTemplate.opsForValue().set(key,json,time,unit); return apply; } //利用逻辑过期解决缓存击穿 public <R,ID> R queryWithLogicalExpire(String prefix,ID id,Class<R> type,Function<ID,R> function,Long time,TimeUnit unit){ String key = prefix + id; //判断在缓存中是否命中 String s = stringRedisTemplate.opsForValue().get(key); //如果不存在,直接返回null if (s == null){ return null; } //如果存在,还得判断是否过期 //反序列化 RedisData redisData = JSONUtil.toBean(s, RedisData.class); JSONObject d = (JSONObject) redisData.getData(); R data = JSONUtil.toBean(d, type); LocalDateTime localDateTime = redisData.getLocalDateTime(); if (localDateTime.isAfter(LocalDateTime.now())){ //如果没有过期 //直接返回数据 return data; } //创建线程池 ExecutorService pool = Executors.newFixedThreadPool(10); //过期了,判断是否可以获取锁 String lock = "getLock"; boolean b = tryLock(lock); if (b){ //如果获取锁成功, pool.submit(() -> { //从数据库中获取数据,再将数据写回缓存中 try { R apply = function.apply(id); setWithLogicalExpire(key,apply,time,unit); } catch (Exception e) { throw new RuntimeException(e); } finally { //释放锁 unlock(lock); } }); } return data; } //获取锁 private boolean tryLock(String key){ Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 200, TimeUnit.SECONDS); return BooleanUtil.isTrue(aBoolean); } //释放锁 private void unlock(String key){ stringRedisTemplate.delete(key); } }
6)依赖:
<!--fastJSON--> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.76</version> </dependency> <!--redis、redis连接池依赖--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-pool2</artifactId> </dependency> <!--hutool--> <dependency> <groupId>cn.hutool</groupId> <artifactId>hutool-all</artifactId> <version>5.7.17</version> </dependency>
7)Redis 配置:
data: redis: password: 你的redis密码 host: 你的redis主机号,IP地址 lettuce: pool: max-active: 10 max-idle: 10 min-idle: 1 time-between-eviction-runs: 10s database: 0