一线大厂高并发Redis缓存架构
Posted 胡尚
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了一线大厂高并发Redis缓存架构相关的知识，希望对你有一定的参考价值。
文章目录
高并发缓存架构设计
架构设计思路
首先是一个基础的缓存架构，对于新增、修改操作set会对缓存更新，对于查询操作先去查询redis缓存，没有再去查询DB，然后把数据在Redis中也缓存一份
但如果数据量很大，redis中不能存储所有的数据，而且很大一部分数据都是冷门数据，可能就插入时用了一次。对于这种情况我们就可以在新增或修改时对缓存set时加过期时间，在查询时将查询到的数据在缓存中更新过期时间，保证热点数据不过期，这样就简单的实现了冷热分离
考虑解决缓存击穿的情况，大量的数据同时间过期导致大量请求直接落到了DB上。在业务上肯定会有一些批量操作，那么对于批量新增的场景，多个数据设置了同一时间过期，那么就可能会出现缓存击穿的情况，解决方法是再加个随机方法，在原有过期时间上再增加一段随机时间
考虑解决缓存穿透的情况，管理员误删热点数据或恶意攻击，一直访问缓存和数据库中都不存在的数据，进而拖垮数据库。
我们可以选择往缓存中存一个空值+过期时间；或者是布隆过滤器。
这里最好是要区分一下，写一个方法区查询缓存，如果查询缓存没有查询到数据返回null，如果查询到了数据但是这个数据是我们上一步存的空值那么这里返回什么，最好是和前一步返回的null区分开
考虑解决热点数据缓存重建问题，热点数据过期或者是冷门数据忽然变成热点数据，这时缓存中没数据，DB中有数据，一瞬间大量的请求要查询这条数据就都落到了DB上。
解决方案是加锁，对查询DB操作加锁，这里需要加分布式锁，锁的粒度要小，比如电商中使用锁前缀+商品id。之所以要使用分布式锁而不是synchronized也是因为锁的粒度，使用synchronized(this)方式那么查询其他商品的操作也会被阻塞住，使用synchronized(锁前缀 + product.getId()) 这种方式生成的锁对象又不是同一个对象。为了性能，同时还需要参考单例模式的双重检测机制，将DCL(Double-Checked-Locking)机制也加上。
伪代码如下：
// 查询缓存
Object o = queryCache(keyStr);
if(o != null)
return o;
// 热点数据缓存重建 缓存没有查询到就加锁
lock.lock();
try
// 再查一遍
Object o = queryCache(keyStr);
if(o != null)
return o;
// 还没有查询到就查询DB并往Redis中写
queryDatabase(...);
finally
lock.unlock();
考虑解决缓存与DB双写不一致问题
双写不一致问题
读写不一致问题，假如更新操作时我不更新缓存，而是删除缓存嘞？如下所示也是有问题的。
要解决上面这些问题的方法还是加锁，在查DB更新缓存、更新DB更新缓存这两个地方加同一把分布式锁。此时就会有人提问了：
-
这么写一个简单的增删改查逻辑代码会变得很复杂
如果并发不高或者能容忍这些问题的话，那么你可以不加这些校验代码，如果你想要解决这些小概率问题那么就必须要加相应的代码。 -
这么写都是串行执行，效率不会很低吗?
在上一步中我们对查询方法就加了DCL机制，绝大部分请求在两次校验时就从缓存中拿到数据返回了，根本不会进入到加锁的位置
Object o = queryCache(keyStr);
if(o != null)
return o;
// 解决热点数据缓存重建问题 缓存没有查询到就加锁
lock.lock();
try
Object o = queryCache(keyStr);
if(o != null)
return o;
// 这个方法中才是查询DB更新缓存，这里面才会加锁，但是大部分请求在上面两次查缓存中就已经拿到数据并返回了
queryDatabase(...);
finally
lock.unlock();
接下来尝试对上面两种分布式锁做优化
首先是为了解决热点数据缓存重建问题而加的分布式锁，其实现在这种方式也能用了，因为只会有一个请求去查询DB然后写入缓存，其他阻塞的线程就会通过第二次验证从缓存中拿到数据返回。但是可能会有几万个请求都阻塞在了加锁的那一行中，我们可以使用trylock()方法来优化，指定一个最大等待时间，这样所有等待的线程一到时间就直接去执行第二个从缓存中取数据的代码了。
接下来是为了解决缓存与DB数据不一致问题而加的分布式锁，大部分的情况下我们业务都是读多写少，我们可以在这里使用读写锁来提高性能，更新DB更新缓存的地方使用写锁，缓存无数据 查询DB更新缓存的地方用读锁。
接下来考虑解决缓存雪崩的问题，我们都是请求–>web服务–>redis，如果并发量很大，多个web服务的请求到发到了一个redis节点上，redis处理不过来一秒十几万或更多的请求，那么redis就可能宕机或者是用户线程一直等待redis响应
如果redis宕机，那么业务代码trycatch后一般就去查数据库了，大量的请求就会让数据库挂掉，接着再整个服务都不能访问了。
如果redis没宕机，用户线程一直等待redis数据返回，此时并发很高，web服务中线程得不到释放，又不断有新的请求进来，又导致微服务可能宕机，进而影响到整个系统宕机。
解决方法：
-
限流，但万一中间件部分限流没有拦住，我们在代码层面也要做相应的处理，使用下面这种解决方案
-
多级缓存，我们可以在JVM层面也加一层缓存，在访问Redis前先访问JVM层面的缓存。因为JVM是本机内存，并发访问可以达到每秒百万
我们还需要一个单独的系统去维护JVM层面的缓存。确定什么样的数据能放缓存中；某个微服务节点更新了数据，其他节点的JVM层面的缓存也要更新等等情况。
完整代码
package com.hs.distributlock.service;
import com.alibaba.fastjson.JSON;
import com.hs.distributlock.entity.ProductEntity;
import com.hs.distributlock.mapper.ProductMapper;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import java.util.Objects;
import java.util.Random;
import java.util.concurrent.TimeUnit;
/**
* @Description: 高并发缓存架构
* @Author 胡尚
* @Date: 2023/3/25 20:10
*/
@Service
public class ProductService
@Autowired
StringRedisTemplate redisTemplate;
@Autowired
Redisson redisson;
@Autowired
ProductMapper productMapper;
/**
* 缓存穿透默认值
*/
public static final String PRODUCT_PENETRATE_DEFAULT = "";
/**
* 突发性热点缓存重建时加的锁
*/
public static final String LOCK_PRODUCT_HOT_CACHE_CREATE_PREFIX = "lock:product:hot_cache_create:";
/**
* 突发性热点缓存重建时加的锁
*/
public static final String LOCK_CACHE_DB_UNLIKE_PREFIX = "lock:cache_db_unlike:";
public void updateProduct(ProductEntity entity)
// 解决缓存与数据库双写 数据不一致问题 而加写锁
RReadWriteLock readWriteLock = redisson.getReadWriteLock(LOCK_CACHE_DB_UNLIKE_PREFIX + entity.getId());
RLock wLock = readWriteLock.writeLock();
wLock.lock();
try
// 更新DB 更新缓存
productMapper.update(entity);
String json = JSON.toJSONString(entity);
redisTemplate.opsForValue().set("product:id:" + entity.getId(), json, 12*60*60+getRandomTime(), TimeUnit.SECONDS);
finally
wLock.unlock();
public void insertProduct(ProductEntity entity)
// 解决缓存与数据库双写 数据不一致问题 而加写锁
RReadWriteLock readWriteLock = redisson.getReadWriteLock(LOCK_CACHE_DB_UNLIKE_PREFIX + entity.getId());
RLock wLock = readWriteLock.writeLock();
wLock.lock();
try
// 更新DB 更新缓存
productMapper.insert(entity);
String json = JSON.toJSONString(entity);
redisTemplate.opsForValue().set("product:id:" + entity.getId(), json, 12*60*60+getRandomTime(), TimeUnit.SECONDS);
finally
wLock.unlock();
/**
* 重点方法，这其中使用了双重检测去查询缓存，还加了读锁去解决缓存和数据库双写导致的数据不一致问题
*/
public ProductEntity queryProduct(Long id) throws InterruptedException
String productKey = "product:id:" + id;
// 从缓存取数据
ProductEntity entity = queryCache(productKey);
if (entity != null)
return entity;
// 突发性热点缓存重建问题，避免大量请求直接去请求DB，进而加锁拦截
RLock lock = redisson.getLock(LOCK_PRODUCT_HOT_CACHE_CREATE_PREFIX + id);
lock.tryLock(3, 30, TimeUnit.SECONDS);
try
// 第二次验证 从缓存取数据
entity = queryCache(productKey);
if (entity != null)
return entity;
// 缓存与DB数据双写不一致问题 加锁
RReadWriteLock readWriteLock = redisson.getReadWriteLock(LOCK_CACHE_DB_UNLIKE_PREFIX + id);
RLock rLock = readWriteLock.readLock();
rLock.lock();
try
// 查询数据库 更新缓存
entity = queryDatabase(productKey, id);
finally
rLock.unlock();
finally
lock.unlock();
return entity;
/**
* 从缓存中查询
*/
private ProductEntity queryCache(String key)
ProductEntity entity = null;
String json = redisTemplate.opsForValue().get(key);
if (StringUtils.hasLength(json))
// 判断是否为解决缓存穿透而手动存储的值，如果是则直接返回一个新对象，并和前端约定好错误提示
if (Objects.equals(PRODUCT_PENETRATE_DEFAULT, json))
return new ProductEntity();
entity = JSON.parseObject(json, ProductEntity.class);
// 延期
redisTemplate.expire(key, 12*60*60+getRandomTime(), TimeUnit.SECONDS);
return entity;
/**
* 从数据库中查询，如果查询到了就将数据在缓存中保存一份，如果没有查询到则往缓存中存一个默认值来解决缓存击穿问题
*/
private ProductEntity queryDatabase(String productKey, long id)
ProductEntity entity = productMapper.get(id);
// 如果数据库中也没有查询到，那么就往缓存中存一个默认值，去解决缓存击穿问题
if (entity == null)
redisTemplate.opsForValue().set(productKey, PRODUCT_PENETRATE_DEFAULT, 60*1000, TimeUnit.SECONDS);
else
redisTemplate.opsForValue().set(productKey, JSON.toJSONString(entity), 12*60*60+getRandomTime(), TimeUnit.SECONDS);
return entity;
private Integer getRandomTime()
return new Random().nextInt(5) * 60 * 60;
开发规范与优化建议
键值设计
Key的设计
-
可读性和可管理性。以业务名为前缀，用冒号分割。避免直接使用id作为key导致数据覆盖的情况。
-
简洁性。保证语义的前提下，控制key的长度，不要太长
-
不要包含空格、换行、单双引号以及其他特殊字符
value的设计
-
避免bigkey
字符串类型，最大512MB，但是一般超过10KB就认为是bigkey
非字符串类型，hash、list、set、zset，一般它们的元素超过5000就任务是bigkey
bigkey的危害：导致redis阻塞、网络拥堵
一般产生bigkey的原因：按天统计某些功能或用户集合、将数据库查询到的数据序列化后存入缓存，我们要判断是否是所有字段都需要缓存
优化bigkey：将一个bigkey拆分为多个数据、如果不可拆分则不要每次取所有的数据，比如
hmget key field [field ...]只读取其中某些元素 -
选择合适的数据类型
反例：
set user:1:name tom set user:1:age 19 set user:1:favor football正例:
hmset user:1 name tom age 19 favor football -
控制key的生命周期，添加过期时间
命令使用
-
O(N)命令关注N的数量
例如hgeall、lrange、smembers、zrange等并非不能用，但需要关注N的数量。
有遍历的需求可以使用hscan、sscan、zscan代替。
-
禁用命令：keys * 、flushall 、flushdb等
-
使用批量操作提高效率，比如pipeline管道等
-
redis的事务不要过多使用，可以使用lua脚本来代替
客户端的使用
-
避免所有业务使用同一个redis实例
可以搭建多个redis集群，不同的微服务调用不同的redis集群，不相干的业务拆分，公共数据做服务化。
-
使用带有连接池，有效控制连接提高效率
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig(); jedisPoolConfig.setMaxTotal(5); jedisPoolConfig.setMaxIdle(2); jedisPoolConfig.setTestOnBorrow(true); JedisPool jedisPool = new JedisPool(jedisPoolConfig, "127.0.0.1", 6379, 3000, null); Jedis jedis = null; try jedis = jedisPool.getResource(); //具体的命令 jedis.executeCommand() catch (Exception e) logger.error("op key error: " + e.getMessage(), key, e); finally //注意这里不是关闭连接，在JedisPool模式下，Jedis会被归还给资源池。 if (jedis != null) jedis.close();连接池参数含义：
序号 参数名 含义 默认值 使用建议 1 maxTotal 资源池中最大连接数 8 设置建议见下面 2 maxIdle 资源池允许最大空闲的连接数 8 设置建议见下面 3 minIdle 资源池确保最少空闲的连接数 0 设置建议见下面 4 blockWhenExhausted 当资源池用尽后，调用者是否要等待。只有当为true时，下面的maxWaitMillis才会生效 true 建议使用默认值 5 maxWaitMillis 当资源池连接用尽后，调用者的最大等待时间(单位为毫秒) -1：表示永不超时 不建议使用默认值 6 testOnBorrow 向资源池借用连接时是否做连接有效性检测(ping)，无效连接会被移除 false 业务量很大时候建议设置为false(多一次ping的开销)。 7 testOnReturn 向资源池归还连接时是否做连接有效性检测(ping)，无效连接会被移除 false 业务量很大时候建议设置为false(多一次ping的开销)。 8 jmxEnabled 是否开启jmx监控，可用于监控 true 建议开启，但应用本身也要开启 一般我们控制最大连接数和资源池允许的最大空闲连接数这两个配置就行了
Redis连接池的连接对象是懒加载的，连接池对象刚创建时不会初始化创建连接对象，是真正使用时才会去创建，我们可以做连接池预热
// 先将创建的连接对象保存起来，然后再统一调用close()方法放回连接池中 List<Jedis> minIdleJedisList = new ArrayList<Jedis>(jedisPoolConfig.getMinIdle()); for (int i = 0; i < jedisPoolConfig.getMinIdle(); i++) Jedis jedis = null; try jedis = pool.getResource(); minIdleJedisList.add(jedis); jedis.ping(); catch (Exception e) logger.error(e.getMessage(), e); finally //注意，这里不能马上close将连接还回连接池，否则最后连接池里只会建立1个连接 //jedis.close(); //统一将预热的连接还回连接池 for (int i = 0; i < jedisPoolConfig.getMinIdle(); i++) Jedis jedis = null; try jedis = minIdleJedisList.get(i); //将连接归还回连接池 jedis.close(); catch (Exception e) logger.error(e.getMessage(), e); finally
扩展
布隆过滤器
当布隆过滤器说某个值存在时，这个值可能不存在；当它说不存在时，那就肯定不存在。
布隆过滤器不能删除数据，如果要删除得重新初始化数据。
底层原理如下图所示：
使用Redisson实现一个简单的案例
package com.redisson;
import org.redisson.Redisson;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
public class RedissonBloomFilter
public static void main(String[] args)
// 构建一个redisson对象
Config config = new Config();
config.useSingleServer().setAddress("redis://localhost:6379");
RedissonClient redisson = Redisson.create(config);
RBloomFilter<String> bloomFilter = redisson.getBloomFilter("nameList");
//初始化布隆过滤器：预计元素为100000000L,误差率为3%,根据这两个参数会计算出底层的bit数组大小
bloomFilter.tryInit(100000000L,0.03);
//将zhuge插入到布隆过滤器中
bloomFilter.add("hushang");
//判断下面号码是否在布隆过滤器中
System.out.println(bloomFilter.contains("aabbcc"));//false
System.out.println(bloomFilter.contains("eeeee"));//false
System.out.println(bloomFilter.contains("hushang"));//true
实际开发中的伪代码如下
//初始化布隆过滤器
RBloomFilter<String> bloomFilter = redisson.getBloomFilter("nameList");
//初始化布隆过滤器：预计元素为100000000L,误差率为3%
bloomFilter.tryInit(100000000L,0.03);
//把所有数据存入布隆过滤器
void init()
for (String key: keys
☞
☞
☞
☞
如果您觉得好看，请点个“在看”↓↓↓
以上是关于一线大厂高并发Redis缓存架构的主要内容，如果未能解决你的问题，请参考以下文章