缓存与数据库双写不一致小概率事件
//线程1 写数据库stock = 5 ---------------》更新缓存
//线程2 写数据库stock = 4 -----》更新缓存
//线程1 ------》写数据库stock = 10 -----》删除缓存
//线程2 -----------------------------------------------------------------------------------------------》写数据库stock = 9 -----》删除缓存
//线程3 -------------------------------------------------》查缓存（空）----》查数据库stock = 10------------------------------------------------------------------------》写缓存
使用 redisson 的 RReadWriteLock，让修改 stock 的地方串行执行，源码还是使用 lua 脚本实现。
开发规范与性能优化
key名设计
- 可读性和可管理性: 以业务名(或数据库名)为前缀(防止key冲突)，用冒号分隔，比如业务名:表名:id
- 简洁性: 控制key的长度，不要包含特殊字符
vlaue设计原则
- 拒绝bigkey
- 字符串类型：它的big体现在单个value值很大，一般认为超过10KB就是bigkey
- 非字符串类型：哈希、列表、集合、有序集合，它们的big体现在元素个数太多。一般来说，string类型控制在10KB以内，hash、list、set、zset元素个数不要超过5000。
- 非字符串的bigkey，不要使用del删除，使用hscan、sscan、zscan方式渐进式删除，同时要注意防止bigkey过期时间自动删除问题(例如一个200万的zset设置1小时过期，会触发del操作，造成阻塞）
-
bigkey的危害
- 导致redis阻塞
- 网络拥塞
- 过期删除
- 默认异步删除
bigkey的产生
一般来说，bigkey的产生都是由于程序设计不当，或者对于数据规模预料不清楚造成的，来看几个例子：
(1) 社交类：粉丝列表，如果某些明星或者大v不精心设计下，必是bigkey。
(2) 统计类：例如按天存储某项功能或者网站的用户集合，除非没几个人用，否则必是bigkey。
(3) 缓存类：将数据从数据库load出来序列化放到Redis里，这个方式非常常用，但有两个地方需要注意，第一，是不是有必要把所有字段都缓存；第二，有没有相关关联的数据，有的同学为了图方便把相关数据都存一个key下，产生bigkey。
如何优化bigkey
- 拆
- big list： list1、list2、…listN
- big hash：可以讲数据分段存储，比如一个大的key，假设存了1百万的用户数据，可以拆分成200个key，每个key下面存放5000个用户数据
- 如果bigkey不可避免，也要思考一下要不要每次把所有元素都取出来(例如有时候仅仅需要hmget，而不是hgetall)
- 选择适合的数据类型
- 控制key的生命周期
命令使用
- hgetall、lrange、smembers、zrange、sinter等并非不能使用，但是需要明确N的值。有遍历的需求可以使用hscan、sscan、zscan代替。
- 禁止线上使用keys、flushall、flushdb等，通过redis的rename机制禁掉命令，或者使用scan的方式渐进式处理。
- redis的多数据库较弱，使用数字进行区分，很多客户端支持较差，同时多业务用多数据库实际还是单线程处理，会有干扰。
- 使用批量操作提高效率
- 原生命令：例如mget、mset。
- 非原生命令：可以使用pipeline提高效率。
- 但要注意控制一次批量操作的元素个数(例如500以内，实际也和元素字节数有关)。
- 原生命令是原子操作，pipeline是非原子操作。
- Redis事务功能较弱，不建议过多使用，可以用lua替代
客户端使用
- 避免多个应用使用一个Redis实例
- 使用带有连接池的数据库，可以有效控制连接，同时提高效率
- 高并发下建议客户端添加熔断功能(例如sentinel、hystrix)
- 设置合理的密码，如有必要可以使用SSL加密访问
- redis的三种清除策略
- 被动删除：当读/写一个已经过期的key时，会触发惰性删除策略，直接删除掉这个过期key
- 主动删除：由于惰性删除策略无法保证冷数据被及时删掉，所以Redis会定期(默认每100ms)主动淘汰一批已过期的key，这里的一批只是部分过期key，所以可能会出现部分key已经过期但还没有被清理掉的情况，导致内存并没有被释放
- 当前已用内存超过maxmemory限定时，触发主动清理策略，八种淘汰策略
-
针对设置了过期时间的key做处理
- volatile-ttl：在筛选时，会针对设置了过期时间的键值对，根据过期时间的先后进行删除，越早过期的越先被删除
- volatile-random：就像它的名称一样，在设置了过期时间的键值对中，进行随机删除。
- volatile-lru：会使用 LRU 算法筛选设置了过期时间的键值对删除。
- volatile-lfu：会使用 LFU 算法筛选设置了过期时间的键值对删除。
-
针对所有的key做处理
- allkeys-random：从所有键值对中随机选择并删除数据。
- allkeys-lru：使用 LRU 算法在所有数据中进行筛选删除。
- allkeys-lfu：使用 LFU 算法在所有数据中进行筛选删除。
-
不处理
- noeviction：不会剔除任何数据，拒绝所有写入操作并返回客户端错误信息"(error) OOM command not allowed when used memory"，此时Redis只响应读操作
-
针对设置了过期时间的key做处理
LRU 算法（Least Recently Used，最近最少使用）
淘汰很久没被访问过的数据，以最近一次访问时间作为参考。
LFU 算法（Least Frequently Used，最不经常使用）
淘汰最近一段时间被访问次数最少的数据，以次数作为参考。
maxmemory-policy(默认是noeviction)，推荐使用volatile-lru
当Redis运行在主从模式时，只有主结点才会执行过期删除策略，然后把删除操作”del key”同步到从结点删除数据。
使用带有连接池的数据库，可以有效控制连接，同时提高效率，标准使用方式
- 连接池的最佳性能是maxTotal = maxIdle，这样就避免连接池伸缩带来的性能干扰。但是如果并发量不大或者maxTotal设置过高，会导致不必要的连接资源浪费。
- 计算资源池大小
- 一次命令时间（borrow|return resource + Jedis执行命令(含网络) ）的平均耗时约为1ms，一个连接的QPS大约是1000
- 业务期望的QPS是50000
- 理论上需要的资源池大小是50000 / 1000 = 50个。但事实上这是个理论值，还要考虑到要比理论值预留一些资源，通常来讲maxTotal可以比理论值大一些。
- 可以给redis连接池做预热
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
JedisPool jedisPool;
{
System.out.println("=====初始化连接池=====");
//资源池中最大连接数
jedisPoolConfig.setMaxTotal(10);
//资源池允许最大空闲的连接数
jedisPoolConfig.setMaxIdle(10);
//资源池确保最少空闲的连接数
jedisPoolConfig.setMinIdle(2);
//向资源池借用连接时是否做连接有效性检测(ping)，无效连接会被移除。业务量很大时候建议设置为false(多一次ping的开销)
jedisPoolConfig.setTestOnBorrow(true);
jedisPool = new JedisPool(jedisPoolConfig, "127.0.0.1", 6379, 3000, null);
//连接池预热示例代码
List<Jedis> minIdleJedisList = new ArrayList<Jedis>(jedisPoolConfig.getMinIdle());
for (int i = 0; i < jedisPoolConfig.getMinIdle(); i++) {
Jedis jedis = null;
try {
jedis = jedisPool.getResource();
jedis.clientSetname("client:" + i);
minIdleJedisList.add(jedis);
jedis.ping();
} catch (Exception e) {
e.printStackTrace();
} finally {
//注意，这里不能马上close将连接还回连接池，否则最后连接池里只会建立1个连接。。
//jedis.close();
}
}
//统一将预热的连接还回连接池
for (int i = 0; i < jedisPoolConfig.getMinIdle(); i++) {
Jedis jedis = null;
try {
jedis = minIdleJedisList.get(i);
//将连接归还回连接池
jedis.close();
System.out.println("连接" + jedis.clientGetname() + "归还成功");
} catch (Exception e) {
e.printStackTrace();
} finally {
}
}
}
@GetMapping("pool")
public String pool() {
Jedis jedis = null;
try {
jedis = jedisPool.getResource();
//具体的命令
String set = jedis.set("pool:redis:", "1");
System.out.println("使用连接：" + jedis.clientGetname() + " 执行" + " 结果:" + set);
} catch (Exception e) {
e.printStackTrace();
} finally {
//注意这里不是关闭连接，在JedisPool模式下，Jedis会被归还给资源池。
if (jedis != null)
jedis.close();
}
return "success";
}
慢查询日志：slowlog
config get slow* #查询有关慢日志的配置信息
config set slowlog-log-slower-than 20000 #设置慢日志使时间阈值,单位微秒，此处为20毫秒，即超过20毫秒的操作都会记录下来，生产环境建议设置1000，也就是1ms，这样理论上redis并发至少达到1000，如果要求单机并发达到1万以上，这个值可以设置为100
config set slowlog-max-len 1024 #设置慢日志记录保存数量，如果保存数量已满，会删除最早的记录，最新的记录追加进来。记录慢查询日志时Redis会对长命令做截断操作，并不会占用大量内存，建议设置稍大些，防止丢失日志
config rewrite #将服务器当前所使用的配置保存到redis.conf
slowlog len #获取慢查询日志列表的当前长度
slowlog get 5 #获取最新的5条慢查询日志。慢查询日志由四个属性组成：标识ID，发生时间戳，命令耗时，执行命令和参数
slowlog reset #重置慢查询日志
布隆过滤器
布隆过滤器就是一个大型的位数组和几个不一样的无偏 hash 函数。所谓无偏就是能够把元素的 hash 值算得比较均匀。
布隆过滤器的实现原理
| 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| k1 | 1 | 1 | 1 | ||||||||||||||||
| k2 | 1 | 1 | 1 | ||||||||||||||||
| k3 | 1 | 1 | 1 |
对 k1 进行多个 hash 计算获得数组索引值，add
对 k2 进行多个 hash 计算获得数组索引值，add文章来源：https://www.toymoban.com/news/detail-717817.html
对 k3 进行多个 hash 计算获得数组索引值，exists，此时出现了 hash 碰撞，误判 k3 存在。文章来源地址https://www.toymoban.com/news/detail-717817.html
- 适用于数据命中不高、 数据相对固定、 实时性低（通常是数据集较大） 的应用场景， 代码维护较为复杂， 但是缓存空间占用很少。
- 根据预计元素、误差率生成数组的长度。
- 不能修改，只能重建。
布隆过滤器的代码示例
@GetMapping("bloom")
public String bloom() {
Config config = new Config();
List<String> nodes = Arrays.asList(
"redis://192.168.139.135:8001",
"redis://192.168.139.136:8002",
"redis://192.168.139.137:8003",
"redis://192.168.139.135:8004" ,
"redis://192.168.139.136:8005",
"redis://192.168.139.137:8006");
//集群配置
config.useClusterServers().setNodeAddresses(nodes);
config.useClusterServers().setPassword("yes");
RedissonClient redisson = Redisson.create(config);
RBloomFilter<String> nameFilter = redisson.getBloomFilter("nameFilter");
//初始化布隆过滤器：预计元素为100000000L,误差率为3%,根据这两个参数会计算出底层的bit数组大小
nameFilter.tryInit(100000000L,0.03);
String[] nameList = { "a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k"};
String[] checkList = { "a1", "b2", "c3", "d", "e4", "f5", "g6", "h7", "i8", "j", "k"};
for (String name : nameList) {
nameFilter.add(name);
}
//判断下面号码是否在布隆过滤器中
for (String name : checkList) {
boolean rs = nameFilter.contains(name);
System.out.println(name + " 检查结果：" + rs);
}
return "success";
}
到了这里，关于49.Redis缓存设计与性能优化的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
