1、多级缓存架构

2、缓存设计

2.1、缓存穿透

    缓存穿透是指查询一个根本不存在的数据， 缓存层和存储层都不会命中， 通常出于容错的考虑， 如果从存储层查不到数据则不写入缓存层。

    缓存穿透将导致不存在的数据每次请求都要到存储层去查询， 失去了缓存保护后端存储的意义。

    造成缓存穿透的基本原因有两个：

• 自身业务代码或者数据出现问题。
• 一些恶意攻击、 爬虫等造成大量空命中。

缓存穿透问题解决方案

1）、缓存空对象

String get(String key) {// 从缓存中获取数据 String cacheValue = cache.get(key); // 缓存为空if (StringUtils.isBlank(cacheValue)) {// 从存储中获取String storageValue = storage.get(key); cache.set(key, storageValue); // 如果存储数据为空， 需要设置一个过期时间(300秒)if (storageValue == null) { cache.expire(key, 60 * 5); }return storageValue;} else {// 缓存非空return cacheValue; }
}

2）、布隆过滤器

    对于恶意攻击，向服务器请求大量不存在的数据造成的缓存穿透，还可以用布隆过滤器先做一次过滤，对于不存在的数据布隆过滤器一般都能够过滤掉，不让请求再往后端发送。当布隆过滤器说某个值存在时，这个值可能不存在；当它说不存在时，那就肯定不存在。

    布隆过滤器就是一个大型的位数组和几个不一样的无偏 hash 函数。所谓无偏就是能够把元素的 hash 值算得比较均匀。向布隆过滤器中添加 key 时，会使用多个 hash 函数对 key 进行 hash 算得一个整数索引值然后对位数组长度进行取模运算得到一个位置，每个 hash 函数都会算得一个不同的位置。再把位数组的这几个位置都置为 1 就完成了 add 操作。

    向布隆过滤器询问 key 是否存在时，跟 add 一样，也会把 hash 的几个位置都算出来，看看位数组中这几个位置是否都为 1，只要有一个位为 0，那么说明布隆过滤器中这个key 不存在。如果都是 1，这并不能说明这个key 就一定存在，只是极有可能存在，因为这些位被置为 1 可能是因为其它的 key 存在所致。如果这个位数组比较稀疏，这个概率就会很大，如果这个位数组比较拥挤，这个概率就会降低。

    这种方法适用于数据命中不高、 数据相对固定、 实时性低（通常是数据集较大） 的应用场景， 代码维护较为复杂， 但是缓存空间占用很少。可以用guvua包自带的布隆过滤器。

<dependency><groupId>com.google.guava</groupId><artifactId>guava</artifactId><version>22.0</version>
</dependency>import com.google.common.hash.BloomFilter; //初始化布隆过滤器
//1000：期望存入的数据个数，0.001：期望的误差率
BloomFilter<String> bloomFilter = BloomFilter.create(Funnels.stringFunnel(Charset.forName("utf‐8")), 1000, 0.001); 
//把所有数据存入布隆过滤器
void init(){for (String key: keys) {bloomFilter.put(key); }
}String get(String key) { // 从布隆过滤器这一级缓存判断下key是否存在Boolean exist = bloomFilter.mightContain(key);if(!exist){ return ""; }// 从缓存中获取数据String cacheValue = cache.get(key);// 缓存为空if (StringUtils.isBlank(cacheValue)) {// 从存储中获取String storageValue = storage.get(key);cache.set(key, storageValue);// 如果存储数据为空， 需要设置一个过期时间(300秒)if (storageValue == null) {cache.expire(key, 60 * 5); } return storageValue; } else { // 缓存非空return cacheValue; } 
}

2.2、缓存失效

    由于大批量缓存在同一时间失效可能导致大量请求同时穿透缓存直达数据库，可能会造成数据库瞬间压力过大甚至挂掉，对于这种情况我们在批量增加缓存时最好将这一批数据的缓存过期时间设置为一个时间段内的不同时间。

2.3、缓存雪崩

    缓存雪崩指的是缓存层支撑不住或宕掉后， 流量会像奔逃的野牛一样， 打向后端存储层。由于缓存层承载着大量请求， 有效地保护了存储层， 但是如果缓存层由于某些原因不能提供服务(比如超大并发过来，缓存层支撑不住，或者由于缓存设计不好，类似大量请求访问bigkey，导致缓存能支撑的并发急剧下降)， 于是大量请求都会达到存储层， 存储层的调用量会暴增， 造成存储层也会级联宕机的情况。

    预防和解决缓存雪崩问题， 可以从以下三个方面进行着手。

• 保证缓存层服务高可用性，比如使用Redis Sentinel或Redis Cluster。
• 依赖隔离组件为后端限流并降级。比如使用Hystrix限流降级组件。
• 提前演练。 在项目上线前， 演练缓存层宕掉后， 应用以及后端的负载情况以及可能出现的问题， 在此基础上做一些预案设定。在正式部署之前，我先把可能的数据先预先访问一遍，这样部分可能大量访问的数据就会加载到缓存中。在即将发生大并发访问前手动触发加载缓存不同的key，设置不同的过期时间，让缓存失效的时间点尽量均匀。

2.4、缓存击穿

    缓存击穿是指缓存中没有但数据库中有的数据（一般是缓存时间到期），这时由于并发用户特别多，同时读缓存没读到数据，又同时去数据库去取数据，引起数据库压力瞬间增大，造成过大压力。

    解决方式

• 设置热点数据永远不过期。
• 加互斥锁：缓存中没有数据，第1个进入的线程，获取锁并从数据库去取数据，没释放锁之前，其他并行进入的线程会等待100ms，再重新去缓存取数据。这样就防止都去数据库重复取数据，重复往缓存中更新数据情况出现。
• 失效时间打散，方式大量缓存同一时间失效。

2.5、热点缓存key重建优化。

    开发人员使用“缓存+过期时间”的策略既可以加速数据读写， 又保证数据的定期更新， 这种模式基本能够满足绝大部分需求。 但是有两个问题如果同时出现， 可能就会对应用造成致命的危害：

• 当前key是一个热点key（例如一个热门的娱乐新闻），并发量非常大。
• 重建缓存不能在短时间完成， 可能是一个复杂计算， 例如复杂的SQL、 多次IO、 多个依赖等。

    在缓存失效的瞬间， 有大量线程来重建缓存， 造成后端负载加大， 甚至可能会让应用崩溃。要解决这个问题主要就是要避免大量线程同时重建缓存。我们可以利用互斥锁来解决，此方法只允许一个线程重建缓存， 其他线程等待重建缓存的线程执行完， 重新从缓存获取数据即可。

String get(String key) {// 从Redis中获取数据String value = redis.get(key);// 如果value为空， 则开始重构缓存if (value == null) {// 只允许一个线程重建缓存， 使用nx， 并设置过期时间exString mutexKey = "mutext:key:" + key;if (redis.set(mutexKey, "1", "ex 180", "nx")) {// 从数据源获取数据value = db.get(key); // 回写Redis， 并设置过期时间redis.setex(key, timeout, value);// 删除key_mutexredis.delete(mutexKey);}// 其他线程休息50毫秒后重试else {Thread.sleep(50);get(key);}}return value;
}

3、开发规范与性能优化

3.1、键值设计

3.1.1、key名设计

3.1.2、value设计

3.2、bigkey的危害

3.3、bigkey的产生

3.4、如何优化bigkey

3.4.1、拆

3.4.2、bigkey是都不可避免？

3.4.3、【推荐】：控制key的生命周期，redis不是垃圾桶。

4、命令使用

4.1、【推荐】 O(N)命令关注N的数量

4.2、【推荐】：禁用命令

4.3、【推荐】合理使用select

4.4、【推荐】使用批量操作提高效率

4.5、【建议】Redis事务功能较弱，不建议过多使用，可以用lua替代

5、客户端使用

5.1、【推荐】避免多个应用使用一个Redis实例

5.2、【推荐】使用带有连接池的数据库，可以有效控制连接，同时提高效率，标准使用方式：

JedisPoolConfig jedisPoolConfig = new JedisPoolConfig(); 
jedisPoolConfig.setMaxTotal(5);
jedisPoolConfig.setMaxIdle(2);
jedisPoolConfig.setTestOnBorrow(true);
JedisPool jedisPool = new JedisPool(jedisPoolConfig, "192.168.0.60", 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();
}

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 { } 
}