redis高并发请求实战

通常来说，我们写代码的时候，都是优先返回缓存值，如果有，则返回缓存的值；如果没有，则查数据库，然后把数据放到缓存，然后再把数据返回。但本例子有很多问题，在高并发读的情况下，缓存失效了，会导致大量的请求查询数据库，导致数据库压力过大崩掉(也就是缓存击穿问题)。请求方一直在请求一个缓存没有且数据库也没有的数据，会导致大量的请求穿透到数据库（也就是缓存穿透问题,可以理解为缓存起不到保护后端持久层，就像被穿透了一样）。还有双写不一致等问题。
如下面的例子有各种各样的问题：

public class ShopService {@Autowiredprivate ShopDao shopDao;@Autowiredprivate RedisUtil redisUtil;@Autowiredprivate RedissonClient redissonClient;/*** 商品的redis key前缀*/private String shopKey = "shop:";/*** 商品的redis 读写锁 key前缀*/private String SHOP_UPDATE_KEY = "shop:update:";/*** 商品为空时的空字符*/private String EMP_SHOP = "{}";/*** 过期时间，秒*/private Integer expireTime = 30 * 60;/*** 获取商品信息* 普遍写法* @param shopId 商品Id* @return*/public Shop get(Long shopId) {Shop shop = null;String redisKey = shopKey + shopId;String shopStr = redisUtil.get(redisKey);if (StringUtils.isNotEmpty(shopStr)) {shop = JacksonUtil.jsonStr2Bean(shopStr, Shop.class);return shop;}shop = shopDao.getById(shopId);if (shop != null) {redisUtil.set(redisKey, JacksonUtil.bean2JsonStr(shop));}return shop;}
}

缓存击穿与缓存雪崩，指的都是缓存失效，然后请求到了数据库。缓存击穿指某个热点 key，在缓存过期的一瞬间，同时有大量的请求打进来，最终请求到了数据库。缓存雪崩指的是同一时间，大量的热点key集中过期了。
对于缓存击穿问题如何解决：
1.对于热点数据缓存同时失效，可以用热点数据的方案，让热点数据永不过期，即每次访问热点数据时都延长热点数据的过期时间即可。
2.用分布式锁对key加锁，就是让并发请求改为串行请求，同时只让一个请求到达数据库，然后把查询结果放到缓存，其他请求从缓存取值。

如下例子为热点数据永不过期的方案：

public class ShopService {@Autowiredprivate ShopDao shopDao;@Autowiredprivate RedisUtil redisUtil;@Autowiredprivate RedissonClient redissonClient;/*** 商品的redis key前缀*/private String shopKey = "shop:";/*** 商品的redis 读写锁 key前缀*/private String SHOP_UPDATE_KEY = "shop:update:";/*** 商品为空时的空字符*/private String EMP_SHOP = "{}";/*** 过期时间，秒*/private Integer expireTime = 30 * 60;/*** 获取商品信息* 保留热点数据，且对过期时间生成随机数，避免出现缓存击穿问题(同一时间缓存大规模失效，大量请求到数据库)* @param shopId 商品Id* @return*/public Shop get2(Long shopId) {Shop shop = null;String redisKey = shopKey + shopId;String shopStr = redisUtil.get(redisKey);if (StringUtils.isNotEmpty(shopStr)) {// 设置缓存过期时间，即保留热点数据，避免大量的商品信息数据把redis内存撑爆。// getExpireTime() 随机过期时间 和热点数据 也能解决缓存击穿问题，如果用户一直访问热点数据，热点数据一直不过期redisUtil.expire(redisKey, getExpireTime());shop = JacksonUtil.jsonStr2Bean(shopStr, Shop.class);return shop;}shop = shopDao.getById(shopId);if (shop != null) {redisUtil.set(redisKey, JacksonUtil.bean2JsonStr(shop));redisUtil.expire(redisKey, getExpireTime());}return shop;}
}    

缓存穿透，指的是redis缓存没有，数据库也没有。请求一直在请求这种数据，导致请求穿透到数据库。
缓存穿透问题如何解决？
1.布隆过滤器，布隆过滤器是一种数据结构，对所有可能查询到的参数都是以 hash 的方式存储，如果有则表示可能有，也可能没有。如果没有则真的是没有。利用这个特性，对所有的数据先放到布隆过滤器，如果布隆过滤器都查询不到，则不用再查询数据库了。
2.如果数据库查询后发现没有数据，就可以缓存一个空对象，然后设置过期时间。

public class ShopService {@Autowiredprivate ShopDao shopDao;@Autowiredprivate RedisUtil redisUtil;@Autowiredprivate RedissonClient redissonClient;/*** 商品的redis key前缀*/private String shopKey = "shop:";/*** 商品的redis 读写锁 key前缀*/private String SHOP_UPDATE_KEY = "shop:update:";/*** 商品为空时的空字符*/private String EMP_SHOP = "{}";/*** 过期时间，秒*/private Integer expireTime = 30 * 60;/*** 获取商品信息* 解决缓存穿透问题，避免客户端一直访问不存在的key* @param shopId 商品Id* @return*/public Shop get3(Long shopId) {Shop shop = null;String redisKey = shopKey + shopId;String shopStr = redisUtil.get(redisKey);if (StringUtils.isNotEmpty(shopStr)) {if (Objects.equals(EMP_SHOP, shopStr)) {redisUtil.expire(redisKey, getExpireTime());return shop;}// 设置缓存过期时间，即保留热点数据，避免大量的商品信息数据把redis内存撑爆。// 且也能解决缓存击穿问题，如果用户一直访问热点数据，热点数据一直不过期redisUtil.expire(redisKey, getExpireTime());shop = JacksonUtil.jsonStr2Bean(shopStr, Shop.class);return shop;}shop = shopDao.getById(shopId);if (shop != null) {redisUtil.set(redisKey, JacksonUtil.bean2JsonStr(shop));redisUtil.expire(redisKey, getExpireTime());} else {// 即使是数据库里面查询id不存在，也创建一个空对象放到redis缓存里面，避免缓存穿透问题// （缓存穿透是指缓存和数据库中都没有的数据，而用户不断发起请求。由于缓存是不命中时被动写的，并且出于容错考虑，如果从存储层查不到数据则不写入缓存，这将导致这个不存在的数据每次请求都要到存储层去查询，失去了缓存的意义）redisUtil.set(redisKey, EMP_SHOP);redisUtil.expire(redisKey, getExpireTime());}return shop;}
}    

如下例子为：解决缓存击穿问题的一种思路

public class ShopService {@Autowiredprivate ShopDao shopDao;@Autowiredprivate RedisUtil redisUtil;@Autowiredprivate RedissonClient redissonClient;/*** 商品的redis key前缀*/private String shopKey = "shop:";/*** 商品的redis 读写锁 key前缀*/private String SHOP_UPDATE_KEY = "shop:update:";/*** 商品为空时的空字符*/private String EMP_SHOP = "{}";/*** 过期时间，秒*/private Integer expireTime = 30 * 60;/*** 获取商品信息* 解决热点数据缓存重建问题（用户短时间内查询冷门商品，这时候大量的请求同时达到数据库）* @param shopId 商品Id* @return*/public Shop get4(Long shopId) {Shop shop = null;String redisKey = shopKey + shopId;shop = getShopFromCache(shopId);if (shop != null) {return shop;}// 利用分布式锁 + DCL双重检查锁方式来实现只能请求一次到数据库RLock lock = redissonClient.getLock(redisKey);lock.lock();try {// 第一次请求的时候是空，则继续向数据库请求；第二次请求过来发现非空，则直接返回，不再查数据库。shop = getShopFromCache(shopId);if (shop != null) {return shop;}shop = shopDao.getById(shopId);if (shop != null) {redisUtil.set(redisKey, JacksonUtil.bean2JsonStr(shop));redisUtil.expire(redisKey, getExpireTime());} else {// 即使是数据库里面查询id不存在，也创建一个空对象放到redis缓存里面，避免缓存穿透问题// （缓存穿透是指缓存和数据库中都没有的数据，而用户不断发起请求。由于缓存是不命中时被动写的，并且出于容错考虑，如果从存储层查不到数据则不写入缓存，这将导致这个不存在的数据每次请求都要到存储层去查询，失去了缓存的意义）redisUtil.set(redisKey, EMP_SHOP);redisUtil.expire(redisKey, getExpireTime());}} finally {lock.unlock();}return shop;}private Shop getShopFromCache(Long shopId) {Shop shop = null;String redisKey = shopKey + shopId;String shopStr = redisUtil.get(redisKey);if (StringUtils.isNotEmpty(shopStr)) {if (Objects.equals(EMP_SHOP, shopStr)) {redisUtil.expire(redisKey, getExpireTime());// 返回一个表示不存在的shop对象，shop = new Shop();return shop;}// 设置缓存过期时间，即保留热点数据，避免大量的商品信息数据把redis内存撑爆。// 且也能解决缓存击穿问题，如果用户一直访问热点数据，热点数据一直不过期redisUtil.expire(redisKey, getExpireTime());shop = JacksonUtil.jsonStr2Bean(shopStr, Shop.class);}return shop;}/*** 返回过期时间随机数* @return*/private int getExpireTime() {// new Random().nextInt(1 * 60);设置随机数是为了解决缓存击穿问题// 如果是批量上架商品，则这个随机时间对解决缓存击穿问题是有生效的，如果没有批量上架商品，则出现大量商品同一时间过期是小概率事件// 缓存击穿是指缓存中没有但数据库中有的数据（一般是缓存时间到期），这时由于并发用户特别多，同时读缓存没读到数据，又同时去数据库去取数据，引起数据库压力瞬间增大，造成过大压力。return expireTime + new Random().nextInt(1 * 60);}
} 

双写不一致问题的解决方案，有很多人都是先删除缓存再写数据库，延时双删等等方案，这些方案都是降低了出现双写不一致的概率，实际上还是有可能出现的。如果要彻底解决这个问题，就是要令读写串行化即可。读写串行化后，如果要写数据，就要等全部读完之后再写。如果要读数据就要写完之后再读。

如本例子在读多写少的场景下，用分布式读写锁，让读写串行化，解决双写不一致问题。

public class ShopService {@Autowiredprivate ShopDao shopDao;@Autowiredprivate RedisUtil redisUtil;@Autowiredprivate RedissonClient redissonClient;/*** 商品的redis key前缀*/private String shopKey = "shop:";/*** 商品的redis 读写锁 key前缀*/private String SHOP_UPDATE_KEY = "shop:update:";/*** 商品为空时的空字符*/private String EMP_SHOP = "{}";/*** 过期时间，秒*/private Integer expireTime = 30 * 60;/*** 获取商品信息* 解决缓存数据库双写不一致的问题* @param shopId 商品Id* @return*/public Shop get5(Long shopId) {Shop shop = null;String redisKey = shopKey + shopId;String updateKey = SHOP_UPDATE_KEY + shopId;shop = getShopFromCache(shopId);if (shop != null) {return shop;}// 利用分布式锁 + DCL双重检查锁方式来实现只能请求一次到数据库RLock lock = redissonClient.getLock(redisKey);lock.lock();try {// 第一次请求的时候是空，则继续向数据库请求；第二次请求过来发现非空，则直接返回，不再查数据库。shop = getShopFromCache(shopId);if (shop != null) {return shop;}// 在读的时候使用读写锁的读锁，然后在写的时候需要使用读写锁的写锁，解决双写不一致问题。// 用ReadWriteLock而不是ReentrantLock，提高了锁的效率/***  更新商品信息时的demo* RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(updateKey);* RLock rLock = readWriteLock.writeLock();* rLock.lock();* try {*    shopDao.update(shopObject)* } finally {*     rLock.unlock();*  }* rLock.unlock();*/RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(updateKey);RLock rLock = readWriteLock.readLock();rLock.lock();try {shop = shopDao.getById(shopId);if (shop != null) {redisUtil.set(redisKey, JacksonUtil.bean2JsonStr(shop));redisUtil.expire(redisKey, getExpireTime());} else {// 即使是数据库里面查询id不存在，也创建一个空对象放到redis缓存里面，避免缓存穿透问题// （缓存穿透是指缓存和数据库中都没有的数据，而用户不断发起请求。由于缓存是不命中时被动写的，并且出于容错考虑，如果从存储层查不到数据则不写入缓存，这将导致这个不存在的数据每次请求都要到存储层去查询，失去了缓存的意义）redisUtil.set(redisKey, EMP_SHOP);redisUtil.expire(redisKey, getExpireTime());}} finally {rLock.unlock();}} finally {lock.unlock();}return shop;}private Shop getShopFromCache(Long shopId) {Shop shop = null;String redisKey = shopKey + shopId;String shopStr = redisUtil.get(redisKey);if (StringUtils.isNotEmpty(shopStr)) {if (Objects.equals(EMP_SHOP, shopStr)) {redisUtil.expire(redisKey, getExpireTime());// 返回一个表示不存在的shop对象，shop = new Shop();return shop;}// 设置缓存过期时间，即保留热点数据，避免大量的商品信息数据把redis内存撑爆。// 且也能解决缓存击穿问题，如果用户一直访问热点数据，热点数据一直不过期redisUtil.expire(redisKey, getExpireTime());shop = JacksonUtil.jsonStr2Bean(shopStr, Shop.class);}return shop;}/*** 返回过期时间随机数* @return*/private int getExpireTime() {// new Random().nextInt(1 * 60);设置随机数是为了解决缓存击穿问题// 如果是批量上架商品，则这个随机时间对解决缓存击穿问题是有生效的，如果没有批量上架商品，则出现大量商品同一时间过期是小概率事件// 缓存击穿是指缓存中没有但数据库中有的数据（一般是缓存时间到期），这时由于并发用户特别多，同时读缓存没读到数据，又同时去数据库去取数据，引起数据库压力瞬间增大，造成过大压力。return expireTime + new Random().nextInt(1 * 60);}
} 

一般来说，写到了demo5，已经比较全了，但还要考虑一个问题超大并发读导致redis撑爆的问题出现缓存雪崩问题，比如每秒10万以上的请求打到redis。
这时候可以考虑用多级缓存来解决这个问题，Map，Guava，Caffeine等jvm缓存方案。比如先从Map获取数据，拿不到再去redis拿。然后再对服务器进行限流设置，比如每个机器只允许2万的请求过来，这样就不会把服务器弄挂了。

题外话:
缓存雪崩的解决方案
解决方案：
1.缓存数据的过期时间设置随机，防止同一时间大量数据过期现象发生。
2.如果缓存数据库是分布式部署，将热点数据均匀分布在不同搞得缓存数据库中。
3.设置热点数据永远不过期。
4.在缓存失效后，通过加锁或者队列来控制读数据库写缓存的线程数量。比如对某个key只允许一个线程查询数据和写缓存，其他线程等待。
具体可参考上述例子。

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！