## 1. Redis大Key,向redis插入100W数据 1. 将100W数据植入txt文档中 ```javascript for ((i=1;i<100*10000;i++));do echo "set k$i v$i" >> /usr/local/redis/redisTest.txt ;done; ``` 2. 查看一页的数据，不要用cat数据太多了，用more ```javascript more /usr/local/redis/redisTest.txt ``` 3. 向redis中开始植入数据 ```javascript cat /usr/local/redis/redisTest.txt | /usr/local/java/redis-7.0.8/src/redis-cli -h 127.0.0.1 -p 6379 -a 123456 --pipe ``` 4. 配置redis禁止使用`keys * ,fulshdb ,flushall`等敏感命令，在redis.conf文件开始配置（单机版），搜索`rename-command CONFIG`  5. 重启redis服务，会提示不存在这些命令  6. 由于要查询关于某个key的前缀查询可以使用`scan`命令 ```javascript scan 0 match k* count 15 ```  ## 2.Bigkey的删除策略 1. String类型控制在10kb以内，hash,list,set,zset元素个数不超过5000 2. 非`String`类型的不要使用del删除，使用hscan,sscan,zscan方式渐进式删除，同时要注意防护bigkey过期时间自动删除问题，例如一个200万的zset设置一个小时国企，会触发del操作，造成阻塞，而且该操作不会出现在慢查询中（latency可查） ### 2.1. 查询redis的大key命令 ```javascript ./redis-cli -p 6379 -a 123456 -c --bigkeys ``` ### 2.2. 进入redis客户端查询指定key的字节大小命令 ```javascript memory usage k10000 ``` ### 2.3.五种类型的删除 1. String类型可以直接用del命令，因为他本身就快 2. hash类型。使用渐进式删除，先删除map中的小key在删除整个key 3. list使用ltrim渐进式逐步删除，直到全部删除完成，分段删除，`ltrim list 0 2` ,只留0-2的位置数据，最后再del删除key 4. set使用sscan每次获取部分元素，再使用srem命令删除每个元素 5. zset使用zscan每次获取本分元素，在使用zremrangebyrank命令删除整个元素 ### 2.4. redis的调优，基于redis.conf的lazyfree调优,非阻塞型删除  ### 2.3. 如何避免大key 1. 对大key进行拆分 - 将一个Big Key拆分为多个key-value这样的小Key，并确保每个key的成员数量或者大小在合理范围内，然后再进行存储，通过get不同的key或者使用mget批量获取。 2. 对大key进行清理 - 对Redis中的大Key进行清理，从Redis中删除此类数据。Redis自4.0起提供了UNLINK命令，该命令能够以非阻塞的方式缓慢逐步的清理传入的Key，通过UNLINK，你可以安全的删除大Key甚至特大Key 3. 监控Redis内存、网络带宽、超时等指标 - 通过监控系统并设置合理的Redis内存报警阈值来提醒我们此时可能有大Key正在产生，如：Redis内存使用率超过70%，Redis内存1小时内增长率超过20%等。 4. 压缩value - 使用序列化、压缩算法将key的大小控制在合理范围内，但是需要注意序列化、反序列化都会带来一定的消耗。如果压缩后，value还是很大，那么可以进一步对key进行拆分。 ## 3.缓存数据库双写一致性 ### 3.1. 采用双检加锁策略 > 加锁前查询一下redsi缓存，加锁后查询一下redsi缓存，以防万一  ### ❌3.2.先更新数据库，在更新缓存（不保证绝对的不可用，都是分情况的） 1. 异常问题1： - 原本100库存，数据库更新99 - 更新redis在不确定因素更新失败，导致最终不一致性 2. 异常问题2： - 原本100库存，并发情况下，两个线程A更新数据库99 - 线程B更新数据库98 - 线程B更新缓存98 - 线程A更新缓存99，导致最终不一致性. ### ❌3.3.先更新缓存，在更新数据库 >不推荐，业务上一般把MySQL作为底单数据库，保证最后的解释权 1. 异常问题； - 并发情况下，两个线程，线程A更新缓存100 - 线程B更新缓存99 - 线程B更新数据库99 - 线程A更新数据库100 ### ❌3.4.先删除缓存，在更新数据库 1. 异常问题 1. 两个并发线程A删除缓存后，此时A去更新数据库， 2. B查询数据缓存没有直接查询数据库的旧数据（此时线程A还在更新数据库），B将旧数据回写到缓存中， 3. 等待A更新完数据库，缓存依然是之前的旧数据 2. 解决方案 1. 延时双删  3. 常见问题 1. 延时这个时间休眠多久 - 线程A休眠的时间需要大于线程B读取再写入缓存的时间，需要评估自己的业务耗时 - 新启动一个后台监控程序，比如看门狗监控程序 2. 这种`同步`（一条线的执行）的淘汰策略，吞吐量降低怎么办？ - 开启一个新的线程，新的线程监控更新成功后再次删除缓存数据 ### ⚠ 3.5.先个更新数据库，在删除缓存（消息队列兜底） 1. 更新数据库数据 2. 数据库会将操作信息写入binlog日志当中 3. 订阅程序提取所需要的数据以及key 4. 另起一段业务代码，获得该信息 5. 尝试删除/更新缓存操作，发现删除/更新失败的话 6. 将这个信息发送到消息队列 7. 重新从消息队列中获得该数据，重试操作 ## 4.布隆过滤器 ### 4.1. bitmap之布隆过滤器的使用场景 1. 现有50亿个电话号码，现有10W个电话号码，如果快速准确判断这些电话号码是否存在 2. 安全链接网址，全球数10亿的网址判断 3. 黑白名单，识别垃圾邮件等 4. 解决缓存穿透问题，redis集合bitmap使用 5. 抖音防止推荐重复视频，饿了么防止推荐重复优惠券，推荐过的就不要重复推荐（推荐时先去布隆过滤器判断是否存在） ### 4.2. 布隆过滤器特性 1. 高效插入和查询，占用空间少，返回结果是不确定性+不够完美 2. 布隆过滤器可以添加元素，但是不能删除元素，由于涉及hashcode判断依据，删掉元素会导致误判率增加（由于不同的值会出现哈希冲突，占用同一个槽位，导致误删） 3. 当实际元素数量超过初始化数量时，应该对布隆过滤器进行重建，重新分配一个size更大的过滤器，再将所有的历史元素批量add进行 4. 使用时最好不要让实际元素远大于初始化数量，一次给够！ ### 4.3. 布隆过滤器代码的简单实现  > 布隆过滤器有，redis有 > 布隆过滤器有，redis无 > 布隆过滤器无，直接返回，不再继续走下去 > 总结：布隆过滤器无，一定是没有的数据，布隆过滤器有，数据不一定存在（因为两个不同的值，哈希值有可能一样，也就是说他们的槽位也是一样的） 1. 将已知数据加入到布隆过滤器 ```java //布隆过滤器用到的详细数据，并非布隆过滤器的key private final static String CUSTOMER = "customer:"; @GetMapping("setUserInit/{userId}") @ApiOperation(value = "用户ID白名单初始化",notes = "用户ID白名单初始化") public void init(@PathVariable String userId) { //白名单客户加载到布隆过滤器 String key = CUSTOMER + userId; redisUtil.set(key, JSONObject.toJSONString(setUser())); //计算hashvalue,由于存在计算出来的负数可能，我们取绝对值 int hashValue = Math.abs(key.hashCode()); //通过hashvalue和2的32次方后取余，获得对应的下标坑位 long index = (long) (hashValue % Math.pow(2, 32)); //设置redis里面的bitmap对应类型白名单：whitelistCustomer的坑位，将该值设置为1 redisUtil.setBit("whitelistCustomer",index,true); } //模拟数据库数据 private User setUser() { User user = new User(); user.setAge(new Random().nextInt(100) + 1); user.setName("张"+new Random().nextInt(100) + 1); return user; } ``` 2. 查看布隆过滤器是否存在，不存在则一定不存在，存在有可能不存在 ```java /** * 检测key是否在槽位 * @param checkItem * @param key * @return */ private boolean checkWithBloomFilter(String checkItem,String key) { //计算hashvalue,由于存在计算出来的负数可能，我们取绝对值 int hashValue = Math.abs(key.hashCode()); //通过hashvalue和2的32次方后取余，获得对应的下标坑位 long index = (long) (hashValue % Math.pow(2, 32)); Boolean exitOk = redisUtil.getBit(checkItem, index); //可以在redis客户端命令为：getbit whitelistCustomer [index] log.info("布隆过滤器下标为：{},是否存在：{}",index,exitOk); return exitOk; } @GetMapping("getUser/{userId}") @ApiOperation(value = "根据用户ID查看布隆过滤器是否存在",notes = "根据用户ID查看布隆过滤器是否存在") public User getUser(@PathVariable String userId) { String key = CUSTOMER + userId; if (!checkWithBloomFilter("whitelistCustomer",key)) { throw new RuntimeException("白名单没有该用户数据" + userId); } //存在的话查询redis User user = JSONObject.parseObject(redisUtil.get(key),User.class); if (null == user) { //查询数据库兜底 user = setUser(); //加redis的string redisUtil.set(key,JSONObject.toJSONString(user)); } return user; } ``` ### 4.4. 布隆过滤器的误判率在0.03 1. 编写Java代码测试 > 下列将100万数据存入布隆过滤器中，故意出误判，将存100万+1起点加到10万数据，故意测误判率 ```java @GetMapping("errProbability") @ApiOperation(value = "测试误判率",notes = "测试误判率") public void errProbability() { //万级单位 int w = 10000; //存100W数据 int all = 100 * w; //误判率0.03 double fpp = 0.03; //布隆过滤器 BloomFilter<Integer> bloomFilter = BloomFilter.create(Funnels.integerFunnel(), all,fpp); //将数据加入布隆过滤器 for (int i = 0; i < all; i++) { bloomFilter.put(i); } //估计误判10万数据 ArrayList<Integer> integers = new ArrayList<>(10 * w); for (int i = all + 1; i < all + 10 * w; i++) { if (bloomFilter.mightContain(i)) { log.info("误判：{}",i); integers.add(i); } } log.info("误判总数：{}",integers.size()); } ```   ## 5. 分布式锁——自研 ### 5.1. setNx分布式锁--递归重试！高并发下严禁使用，容易内存溢出 1. 使用初级版本实现分布式锁代码 ```java @Autowired private StringRedisTemplate stringRedisTemplate; private final static String SHOP001 = "shop001"; /** * 初始化数量5000 */ @PostConstruct public void init() { stringRedisTemplate.opsForValue().set(SHOP001,"5000"); } /** * 递归重试 */ @GetMapping("getShop") @ApiOperation(value = "setNx分布式锁", notes = "setNx分布式锁") public String getShop() throws InterruptedException { String msg = ""; String key = "shop"; String uuidValue = IdUtil.simpleUUID() + ":" + Thread.currentThread().getId(); //获取锁 Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, uuidValue); if (Boolean.FALSE.equals(flag)) { log.warn(uuidValue+"：未拿到锁，等待20毫秒继续执行"); //暂停20s，进行递归重试 TimeUnit.MILLISECONDS.sleep(20); getShop(); }else { log.info(uuidValue+"：拿到锁，开始继续执行"); try { //查询库信息 String shop001 = stringRedisTemplate.opsForValue().get(SHOP001); //查看库存 int inventNum = shop001 == null ? 0 : Integer.parseInt(shop001); //扣减库存，每次-1 if (inventNum > 0) { stringRedisTemplate.opsForValue().set(SHOP001, String.valueOf(--inventNum)); log.info("成功卖出一个商品，库存剩余：{}",inventNum); msg = "成功卖出一个商品，库存剩余：" + inventNum; } else { log.info("商品卖完了"); msg = "商品卖完了"; } } finally { stringRedisTemplate.delete(key); } } return msg; } ``` ### 5.2. 第二版本，采用`while`自旋的方式避免高并发下的内存溢出，以及用if的虚假唤醒（以下存在锁过期问题） ```javascript /** * 自旋 */ @GetMapping("getShop1") @ApiOperation(value = "setNx分布式锁", notes = "setNx分布式锁") public String getShop1() throws InterruptedException { String msg = ""; String key = "shop"; String uuidValue = IdUtil.simpleUUID() + ":" + Thread.currentThread().getId(); //不用递归，高并发下会内存溢出，虚假唤醒，用自旋替代递归方法的重试，用while替换if while (Boolean.FALSE.equals(stringRedisTemplate.opsForValue().setIfAbsent(key, uuidValue))) { log.warn(uuidValue + "：未拿到锁，等待20毫秒继续执行"); //暂停20s，进行递归重试 TimeUnit.MILLISECONDS.sleep(20); } log.info(uuidValue + "：拿到锁，开始继续执行"); try { //查询库信息 String shop001 = stringRedisTemplate.opsForValue().get(SHOP001); //查看库存 int inventNum = shop001 == null ? 0 : Integer.parseInt(shop001); //扣减库存，每次-1 if (inventNum > 0) { stringRedisTemplate.opsForValue().set(SHOP001, String.valueOf(--inventNum)); log.info("成功卖出一个商品，库存剩余：{}", inventNum); msg = "成功卖出一个商品，库存剩余：" + inventNum; } else { log.info("商品卖完了"); msg = "商品卖完了"; } } finally { stringRedisTemplate.delete(key); } return msg; } ``` ### 5.3. 加过期时间(依然存在问题，finally中判断和删除`要`保持原子性！需要`lua`脚本) >注意：切记要自己删除自己的锁 >比如设置key（锁）得过期时间为30S >线程A优先获取锁，执行的程序30S的时候没有执行完，自动删除key(锁)了，此时线程A还在执行业务 >线程B过了30S获取锁，在正常执行，此时线程A执行完了，删除了key(锁)此时你要记住他删的是线程B的做啊！  ### 5.4. Lua脚本实现分布式锁（依然存在且问题，要考虑锁的可重入性） 1. redis客户端，对脚本语言的基本使用 - 传统的三个命令，并不是原子性，1. `set k1 v1`;2.`expire k1 30`;3.`expire k1 30` - 利用脚本组成一句，最后的0表示没有动态参数 ```javascript eval "redis.call('set','k1','k2') redis.call('expire','k1','30') return redis.call('get','k1')" 0 ```  - 关于动态参数的配置，后面2表示用到几个参数 ```javascript eval "return redis.call('mset',KEYS[1],ARGV[1],KEYS[2],ARGV[2])" 2 k1 k2 lua1 lua2 ```  2. 参考[Redis的官网的lua脚本](https://redis.io/docs/manual/patterns/distributed-locks/)  ```javascript if redis.call("get",KEYS[1]) == ARGV[1] then return redis.call("del",KEYS[1]) else return 0 end ``` - 预先设置一个锁的key，然后判断key是不是自己的，是的删除  - 利用redis客户端执行该命令 ```javascript eval "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end" 1 dadalock 111222 ```  - 关于lua的if练习  3. 代码的实现 ```javascript /** * LUA脚本删除锁 */ @GetMapping("getShop3") @ApiOperation(value = "setNx分布式锁3", notes = "setNx分布式锁3") public String getShop3() throws InterruptedException { String msg = ""; String key = "shop"; String uuidValue = IdUtil.simpleUUID() + ":" + Thread.currentThread().getId(); //不用递归，高并发下会内存溢出，虚假唤醒，用自旋替代递归方法的重试，用while替换if,设置过期时间30S，一定要是原子性操作啊！！！ while (Boolean.FALSE.equals(stringRedisTemplate.opsForValue().setIfAbsent(key, uuidValue,30L,TimeUnit.MINUTES))) { log.warn(uuidValue + "：未拿到锁，等待20毫秒继续执行"); //暂停20s，进行递归重试 TimeUnit.MILLISECONDS.sleep(20); } log.info(uuidValue + "：拿到锁，开始继续执行"); try { //查询库信息 String shop001 = stringRedisTemplate.opsForValue().get(SHOP001); //查看库存 int inventNum = shop001 == null ? 0 : Integer.parseInt(shop001); //扣减库存，每次-1 if (inventNum > 0) { stringRedisTemplate.opsForValue().set(SHOP001, String.valueOf(--inventNum)); log.info("成功卖出一个商品，库存剩余：{}", inventNum); msg = "成功卖出一个商品，库存剩余：" + inventNum; } else { log.info("商品卖完了"); msg = "商品卖完了"; } } finally { //利用lua脚本 String lauScript = "if redis.call('get',KEYS[1]) == ARGV[1] then " + "return redis.call('del',KEYS[1]) " + "else " + "return 0 " + "end"; //DefaultRedisScript参数1是lua脚本，参数2是参数返回类型 Object execute = stringRedisTemplate.execute(new DefaultRedisScript(lauScript, Boolean.class), Arrays.asList(key),uuidValue); log.info("利用lua脚本删除锁{}",execute); } return msg; } ``` ### 5.5. 针对上面四个锁进行总结 1. `setnx`，只能解决有无的问题，够用但是不够完美 2. `hset`（map类型解决可重入锁的问题）,还解决了有无 ### 5.6. 基于lua脚本连接redis的map结构进行计数器加锁，解锁编写可重入锁  1. lua脚本编写加锁代码 - v1版本 ```lua -- 判断key是否存在 if redis.call('exists','key') == 0 then -- 不存在添加key redis.call('hset','key','uuid;threadid',1) -- 设置key过期时间50 redis.call('expire','key',50) -- 1表示操作成功，返回 return 1 -- 判断key是否存在 elseif redis.call('hexists','key','uuid;threadid') == 1 then -- 存在对key进行+1 redis.call('hincrby','key','uuid:threadid',1) -- 设置key过期时间50 redis.call('expire','key',50) -- 1表示操作成功，返回 return 1 else -- 0表示没有执行任何数据，返回 return 0 end ``` - v2版本(经测试`hincrby`进行+1的操作，即使没有key也会加上数据) ```lua -- 判断key是否存在 if redis.call('exists','key') == 0 or redis.call('hexists','key','uuid;threadid') == 1 then -- 不存在添加key redis.call('hincrby','key','uuid;threadid',1) -- 设置key过期时间50 redis.call('expire','key',50) -- 1表示操作成功，返回 return 1 else -- 0表示没有执行任何数据，返回 return 0 end ``` - v3动态参数版本 ```lua -- 判断key是否存在 if redis.call('exists',KEYS[1]) == 0 or redis.call('hexists',KEYS[1],ARGV[1]) == 1 then -- 不存在添加key redis.call('hincrby',KEYS[1],ARGV[1],1) -- 设置key过期时间50 redis.call('expire',KEYS[1],ARGV[2]) -- 1表示操作成功，返回 return 1 else -- 0表示没有执行任何数据，返回 return 0 end ``` 2. 进入redis客户端进行测试lua脚本 ```javascript eval "if redis.call('exists',KEYS[1]) == 0 or redis.call('hexists',KEYS[1],ARGV[1]) == 1 then redis.call('hincrby',KEYS[1],ARGV[1],1) redis.call('expire',KEYS[1],ARGV[2]) return 1 else return 0 end" 1 redisLock 111222:1 50 ```  3. lua脚本编写解锁代码 - v1版本 ```lua -- 查看key是否存在 if redis.call('hexists',key,uuid:treeadid) == 0 then -- 不存在返回null return nil -- 进行数值-1.判断是不是=0 elseif redis.call('hincrby',key,uuid;thread,-1) == 0 then -- 等于0可以删除key return redis.call('del',key) else -- 不是自己的锁，什么也不操作 return 0 end ``` - v2动态参数版本 ```lua -- 查看key是否存在 if redis.call('hexists',KEYS[1],ARGV[1]) == 0 then -- 不存在返回null return nil -- 进行数值-1.判断是不是=0 elseif redis.call('hincrby',KEYS[1],ARGV[1],-1) == 0 then -- 等于0可以删除key return redis.call('del',KEYS[1]) else -- 计数器还没减为0，什么也不操作 return 0 end ``` 4. 加锁解锁redsi客户端全套测试（加三次锁，解三次锁后删除key） ```javascript eval "if redis.call('hexists',KEYS[1],ARGV[1]) == 0 then return nil elseif redis.call('hincrby',KEYS[1],ARGV[1],-1) == 0 then return redis.call('del',KEYS[1]) else return 0 end" 1 redisLock 111222:1 ```  ### 5.7.自研分布式锁加自动续期 > 谈到自研，就要考虑到通用性，适配性，可扩展性 1. 基于以上特点，编写一个工厂模式，方便加入其他模式的分布式锁 ```java @Configuration public class DistributedLockFactory { @Autowired private StringRedisTemplate stringRedisTemplate; private String lockName; private String uuid; public DistributedLockFactory() { this.uuid = IdUtil.simpleUUID(); } public Lock getDistributedLock(String lockType) { if (ObjectUtils.isEmpty(lockType)) return null; if (lockType.equalsIgnoreCase("redis")) { this.lockName = "redisLock"; return new RedisDistributedLock(stringRedisTemplate,lockName,uuid); }else if (lockType.equalsIgnoreCase("zk")) { this.lockName = "zkLock"; //TODO return null; }else if (lockType.equalsIgnoreCase("MySQL")) { this.lockName = "mysqlLock"; //TODO return null; } return null; } } ``` 2. redis分布式锁的Map基于lua脚本实现 ```javascript public class RedisDistributedLock implements Lock { private StringRedisTemplate stringRedisTemplate; private String lockName; private String uuidValue; private String expireTime; //构造方法注入 public RedisDistributedLock(StringRedisTemplate stringRedisTemplate, String lockName, String uuid) { this.stringRedisTemplate = stringRedisTemplate; this.lockName = lockName; //线程id不会重复 this.uuidValue = uuid + ":" + Thread.currentThread().getId(); this.expireTime = "50"; } @Override public void lock() { tryLock(); } @Override public boolean tryLock() { try { tryLock(-1L,TimeUnit.SECONDS); } catch (InterruptedException e) { throw new RuntimeException(e); } return false; } /** * 尝试获取锁 * @return * @throws InterruptedException */ @Override public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { if (time == -1L) { String luaScript = "if redis.call('exists',KEYS[1]) == 0 or redis.call('hexists',KEYS[1],ARGV[1]) == 1 then " + "redis.call('hincrby',KEYS[1],ARGV[1],1) " + "redis.call('expire',KEYS[1],ARGV[2]) " + "return 1 " + "else " + "return 0 " + "end"; while (Boolean.FALSE.equals(stringRedisTemplate.execute(new DefaultRedisScript<>(luaScript, Boolean.class), Arrays.asList(lockName), uuidValue, expireTime))) { TimeUnit.MILLISECONDS.sleep(20); } return true; } return false; } /** * 解锁 */ @Override public void unlock() { String luaScript = "if redis.call('hexists',KEYS[1],ARGV[1]) == 0 then " + "return nil " + "elseif redis.call('hincrby',KEYS[1],ARGV[1],-1) == 0 then " + "return redis.call('del',KEYS[1]) " + "else " + "return 0 " + "end"; // nil = false ; 1 = true ; 0 = false Long flag = stringRedisTemplate.execute(new DefaultRedisScript<>(luaScript, Long.class), Arrays.asList(lockName), uuidValue); if (null == flag) { throw new RuntimeException("锁不存在"); } } /** * 下面两个不用，也用不到 */ @Override public void lockInterruptibly() throws InterruptedException { } @Override public Condition newCondition() { return null; } } ``` 3. 接口中使用 ```javascript @RestController @RequestMapping("aqsLock") @Api(value = "自研可重入分布式锁") @Slf4j public class AQSLockController { @Autowired private StringRedisTemplate stringRedisTemplate; @Autowired private DistributedLockFactory distributedLockFactory; private final static String SHOP001 = "shop001"; @GetMapping("getData") @ApiOperation(value = "自研可重入分布式锁", notes = "自研可重入分布式锁") public String getData() { //加锁 Lock lock = distributedLockFactory.getDistributedLock("redis"); lock.lock(); String msg = ""; try { //查询库信息 String shop001 = stringRedisTemplate.opsForValue().get(SHOP001); //查看库存 int inventNum = shop001 == null ? 0 : Integer.parseInt(shop001); //扣减库存，每次-1 if (inventNum > 0) { stringRedisTemplate.opsForValue().set(SHOP001, String.valueOf(--inventNum)); log.info("成功卖出一个商品，库存剩余：{}", inventNum); msg = "成功卖出一个商品，库存剩余：" + inventNum; testReEntry(); } else { log.info("商品卖完了"); msg = "商品卖完了"; } } finally { lock.unlock(); } return msg; } //重入锁 private void testReEntry() { //加锁 Lock lock = distributedLockFactory.getDistributedLock("redis"); lock.lock(); try { log.info("==================================测试可重入锁================================="); } finally { lock.unlock(); } } } ``` 4. 在以上功能中加入自动续期功能，主要是定时（这个定时要小于定制的key过期时长）检测reids中的锁是否还存在  ```java /** * 相当于看门狗 */ private void renewExpire() { String luaScript = "if redis.call('hexists',KEYS[1],ARGV[1]) == 1 then " + "return redis.call('expire',KEYS[1],ARGV[2]) " + "else " + "return 0 " + "end "; new Timer().schedule(new TimerTask() { @Override public void run() { //返回两个值就用bollen类型 if(Boolean.TRUE.equals(stringRedisTemplate.execute(new DefaultRedisScript<>(luaScript, Boolean.class), Arrays.asList(lockName), uuidValue, expireTime))) { renewExpire(); } } //每隔十秒执行看门狗 },(Long.parseLong(this.expireTime) * 1000) / 3); } ``` ### 5.8. 关于自研分布式锁展开思路需要哪些步骤（总结） 1. 安装JUC里面的`java.util.concurrent.locks.Lock`接口会犯编写，里面已经有加锁解锁的接口，自己可以实现它并重写这两个方法 2. lock()加锁关键逻辑（要满足：`独占性`，`高可用`，`防死锁`，`不乱抢`，`重入性`） - 加锁：加锁实际上就是在redis中，给key（利用Map类型，加入加锁的次数）设置一个值，且给定一个过期时间，要保持原子性 - 自旋：加锁不成功，需要while进行重试并自旋，严谨while代码块里面是空的，最好加个延时20毫秒等，不加的话会导致cpu打满 - 续期：自动续期，加锁成功后，立即后台启动个线程监听redis的key是否存在，加钟 3. unlock解锁关键逻辑 - 自己的锁要删除自己的锁，不要删除别人的锁，避免张冠李戴 - 考虑可重入性的递减，加锁几次就要减锁几次 - 最后次数为0的时候，直接del删除 ## 6. 分布式锁RedLock——Redisson >理念：该方案也是基于set加锁，Lua脚本解锁进行改良的，所以redis之父只描述了差异的地方，多台redis实例都为master节点，大致方案如下 >1. 获取当前时间，以毫秒为单位 >2. 以此尝试从5个实例，使用相同的key和随机值（UUID订单号）获取锁，当向Redis请求获取锁时，客户端应该设置一个超时时间，这个超时时间要小于锁的失效时间，例如锁的失效时间为10S，则超时时间应该为5-50MS，这样可以防止一个Redis宕机导致长时间处于阻塞状态，如果不可用的话客户端应该尽快尝试另一个redis请求获取锁 > 3. 例如我们有3个客户端实例，那么最少2个实例加锁成功才算分布式锁加锁成功。如果没有满足该条件，就需要在向所有的Redis实例发送释放锁命令即可，不用关心之前有没有从Redis实例成功获取到锁. ### 6.1. 代码案例 1. Redisson配置类`RedissonConfig` ```java @Configuration public class RedissonConfig { @Bean public Redisson redisson(){ Config config = new Config(); config.useSingleServer() .setAddress("redis://139.196.111.66:6379") .setPassword("******") .setDatabase(6) .setConnectTimeout(20000) .setTimeout(5000) ; return (Redisson) Redisson.create(config); } } ``` 2. 业务代码 ```javascript public class RedissonController { @Autowired private Redisson redisson; @Autowired private StringRedisTemplate redisTemplate; //属性，方法 private int number = 10; @GetMapping("lock") @ApiOperation(value = "分布式锁解决超卖问题",notes = "分布式锁解决超卖问题") public void lock(String orderId){ //------------------------------------------------------// // 加分布式锁情况 // //-----------------------------------------------------// RLock lock = null; try { lock = redisson.getLock(orderId);//获取锁的对象 //lock.lock();//锁的过期时间(默认30s)，如果业务时间过长，锁会被自动续时长 lock.lock(10, TimeUnit.SECONDS); //设置锁的过期时间(不会续时长 ) if (number>0){ //限购商品为每人1个 Boolean aBoolean = redisTemplate.hasKey(Thread.currentThread().getName()); if (!Boolean.TRUE.equals(aBoolean)){ System.out.println(Thread.currentThread().getName()+"卖出了第"+number--+"张票,剩余："+number); //60s即可继续抢 redisTemplate.opsForValue().set( Thread.currentThread().getName() ,new SimpleDateFormat("yyyy-MM-dd HH:mm:ss SSS").format(new Date()) ,60,TimeUnit.SECONDS ); } } } finally { assert lock != null; //改进点，只能删除属于自己的key,不能删除别人的 if (lock.isLocked() && lock.isHeldByCurrentThread()) { lock.unlock();//释放锁 } } } } ``` ### 6.2. 源码解析 1. Redisson的看门狗，守护线程"续命"，在获取锁成功后，给锁加一个watchdog,watchdog会起一个定时任务，在锁没有被释放且快要过期的时候续期。锁的key默认的超时时间为30S 2. 加锁过程 - 通过exists判断，如果锁不存在，则设置值和过期时间，加锁成功（Map类型） - 通过hexists判断，如果锁已存在，并且锁的是当前线程，则证明是重入锁，加锁成功 - 如果锁已存在，但锁不是当前线程，则证明有其他线程持有锁。返回当前锁的过期时间（代表了锁key的剩余生存时间）。加锁失败 3. 加锁成功看门狗守护线程"续命"给锁加一个watchdog,watchdog会起一个定时任务，在锁没有被释放且快要过期的时候续期。锁的key默认的超时时间为30S，每10秒续命 - 同样的调用Lua脚本，key存在的话就缓存续命，不是自己的锁续不了 4. 解锁的步骤 - 如果释放锁的线程和已存在的线程不是同一个线程，返回null - 通过hincrby递减1，先释放一次锁，若剩余次数还大于0，则证明当前锁是重入锁，刷新过期时间 - 若剩余次数小于0，删除key并发布释放锁的消息，解锁成功 ### 6.3. 多节点分布式锁 1. 这个锁的算法实现了多个redis实例情况，相对于单个redis节点来说，优点在于防止了单节点故障造成整个服务停止运行的情况，且多节点中锁的设计，及多寄点同时崩溃等各种意外的情况 2. Redisson分布式锁支持MultiLock机制可以将多个锁合并一个大锁，对一个大锁进行统一的申请加锁以及解锁 3. 互斥：任何时刻只能由一个client获取锁， 4. 释放死锁：锁定的资源崩溃也要保证能够释放锁， 5. 容错性：只要多数redis节点（一半以上）在使用，client就可以获取和释放锁，当宕机的redis节点重新启动后，发现锁还存在的话，会先从默认30S来倒计时，然后与其他节点进行同步，如果其他节点为25S那就瞬间成25S ## 7.内存淘汰策略内存查看和打满OOM ### 7.1. 查看redis内存信息 1. 内存查看命令 ```java config get maxmemory ```  2. 查看机器内存详细信息 ```java info memory ``` 3. 电脑的操作系统64位，显示0是表示无限制，电脑实际的物理内存 4. 电脑的32位操作系统，默认最大为3G ### 7.2. 一般在生产上配置多大的内存空间呢 1. 一般配置redis内存为物理内存的四分之三 2. 可以再redis.conf文件中进行修改内存大小  3. 也可以通过命令的方式来设置内存大小（临时，重启后失效） ```java config set maxmemory 1048576 ``` ### 7.2. 一般在生产上配置多大的内存空间呢 1. 一般配置redis内存为物理内存的四分之三 2. 可以再redis.conf文件中进行修改内存大小  3. 也可以通过命令的方式来设置内存大小（临时，重启后失效） ```java config set maxmemory 1048576 ``` ## 8. reids底层的redisObject 1. Redis采用redisObject结构来统一物种不同的数据类型，这样所有的数据类型就都可以以相同的形式在函数间传递而不用使用特定的类型结构。 2. 为了识别不同的数据类型，redisObject种定义了type和encoding字段对不用的数据类型加以区别 3. redisObject就是String,hash,list,set,zset的父类，可以在函数间传递时隐藏具体的类型信息，所以作者抽象了redisObjetc结构来到达同样的目的 4. 以下为C语言代码 ```cpp typedef struct redisObject { unsigned type:4; /*对象类型，包括；obj_string,obj_list,obj_hash,obj_set,obj_zset*/ unsigned encoding:4; /*redis底层具体的数据结构，一共8种*/ unsigned lru:REDIS_LRU_BITS; /*24位，对象最后一次被命令程序访问的时间，与内存回收有关lru*/ int refcount; /*引用次数，当refcount位0的时候，表示该对象已经不被任何对象引用，则可以进行垃圾回收了*/ void *ptr; /*指向对象实际的数据结构，也就是用户传递的真实数据*/ } robj; ``` 5. 基于此，`unsigned encoding:4;`上面redis底层数据结构的编码位10个,常用的8个  ## 9.五大类型的底层存储结构 ### 9.1. redis6之前的数据类型与数据结构之间的关系 1. String = 动态字符串（SDS） 2. Set = 哈希表（hashtable） + 整数数组（intset） 3. Sorted Set = 跳表(skipList) + 压缩列表（zipList） 4. List = 双端链表（quicklist） + 压缩列表（zipList） 5. Hash = 哈希表（hashtable） + 压缩列表（zipList）  ### 9.2. redis7的数据类型与数据结构之间的关系(将`压缩列表`替换成`紧凑列表`) 1. String = 动态字符串（SDS） 2. Set = 哈希表（hashtable） + 整数数组（intset） 3. Sorted Set = 跳表(skipList) + 紧凑列表（listpack） 4. List = 双端链表（quicklist） 5. Hash = 哈希表（hashtable） + 紧凑列表（listpack）  ### 9.3. 黑窗口调试工具`预先：set age 18` 1. 查看key的类型 ```cpp type age ``` 2. 查看key内部存储类型 ```cpp object encoding age ``` 3. 调试查看key的详细信息，需要套打开debug模式，`redis7`默认是关闭的 ```cpp debug object age ```  4. 查看关闭状态 ```cpp config get enable-debug-command ```  5. 关闭redis服务，打开redis.cong,修改配置文件后重启redis服务  6. 打开后，查看详细  - Value：key的二进制存储位置 - refcount：被引用数 - encoding：底层存储的数据结构 - serializedlength：字符串长度 - lru：缓存淘汰策略 - lru_seconds_idle：空闲数：值越小越忙：说明频繁被调用 ### 9.4. String类型的底层结构（简答：`动态字符串`） 1. int - 保存long类型的64位的整数 - 数字长度最长19位 - 只有整数才会int，如果小数点那些是转成字符串 2. embstr - 代表embstr格式的SDS动态字符串，保存长度小于44字节的字符串 3. raw - 保存长度大于44字节的字符串 4. 底层自适应的选择较为优化的内部编码格式，而这一切对用户完全透明的 5. 流程步骤  ### 9.5. Hash类型的底层结构 - redis6 == ziplist(压缩列表),hashtable - redis7 == listpack(紧凑列表),hashtable #### 9.5.1. redis6的结构 1. 查看redis6的配置 ```cpp config get hash* ```  - HASH的小key超过512由压缩列表转换hashtable - 键值长度超过64字节（一个英文字母一个字节）由压缩列表转换hashtable 2. 修改hash结构的配置 ```cpp config set hash-max-ziplist-entries 3 config set hash-max-ziplist-value 8 ``` 3. 使用命令查看key在内部存储类型 ```cpp object encoding user01(key) ``` 4. 综上，hash结构的类型小key字段值（hash的大key下面有多个小key超过3就会转换）大于3的时候会由ziplist转换成hashtable;value值长度超过8也会由ziplist转换成hashtable；切记升级过后，反过来转换是不可以的 #### 9.5.2. redis7的结构 1. 与6相比，就是将ziplist压缩列表替换成了listpack紧凑列表，其他用法都一样的 2. listpack #### 9.5.3. 为什么要用listpack替换ziplist呢？ 1. ziplist 内存浪费：ziplist 存储的元素大小是固定的，即使元素大小不足最大限制，也会浪费内存空间。 2. ziplist 扩容效率低下：当 ziplist 容量不足时，需要扩容，但扩容时需要重新分配内存并复制原数据，效率较低。 3. ziplist 插入和删除效率低：在插入或删除元素时，ziplist需要进行元素的移动和内存的重新分配，效率较低。 4. listpack紧凑的存储结构：listpack 存储的元素大小是可变的，不会浪费内存空间。 5. listpack高效的扩容策略：listpack 采用类似于动态数组的扩容策略，可以在不需要重复分配和复制数据的情况下扩容。 ### 9.6.List类型底层结构 1. 底层是quickList结构，quicklist存储了一个双向链表，每个节点都是一个listpack 2. 每一个节点是一个quicklistNode，包含prev和next指针。 3. 每一个quicklistNode 包含 一个ziplist在redis7改成listpack，*zp 压缩链表里存储键值。 4. 所以quicklist是对ziplist(redis7用listpack)进行一次封装，使用小块的ziplist来既保证了少使用内存，也保证了性能。 ### 9.7.Set类型底层结构 >设置长度类型  1. set的两种编码格式：integer,hashtable 2. 超过设置的最大值（默认最大值是512）会由integer类型转变成hashtable ### 9.8.ZSet类型底层数据结构 - reids6为ziplist（压缩链表）+skiplist（跳表） - redsi7为listpack（紧凑列表）+skiplist（跳表）  1. 超过设置的最大值（默认最大值是512）会由listpack(紧凑列表)类型转变成skiplist跳表 ## 10. skiplist跳表 [复习链接](https://www.bilibili.com/video/BV13R4y1v7sP/?p=164&spm_id_from=..top_right_bar_window_history.content.click&vd_source=1b68b6ac864b7378d8ff1cee8769dd41)  ### 10.1. 跳表是什么 1. 跳表是可以实现二分查找的有序链表，它是一个空间换时间的结构，访问时间更快，由于链表无法进行二分查找，因此借鉴数据库索引的意思，提取出链表关键节点（索引），先在关键节点上查找，在进入下层链表查找，提取多层关键节点，形成了跳跃表。 2. 最终跳表就是链表+多级索引 ### 10.2. 跳表的优缺点 1. 跳表优点：跳表是一个最经典的空间换时间的解决方案（索引越来越多，但是查询时间少），而且只有在数据量较大的情况下才能体现出来又是。而且应该是读多写少的情况下才能使用，所以它的使用范围应该还是比较有限的 2. 跳表的缺点：维护成本相对要高，在单链表种一旦定位好要插入的位置，插入时间点的时间复杂度是很低的，就是O(1),但是新增或是删除时需要把所有索引都更新一边，为了保证原始链表中的数据有序性，我们需要先找到要动的位置，这个查找操作就会比奥耗时最多再新增和删除的过程种更新，时间复杂度是O(log n) ## 11. IO多路复用（主要是Linux的`epoll`函数搞定的） ### 11.1. IO多路复用的理念 1. 一个进程能够同时处理多个tcp链接，例如一个老师监场100个学生，哪位学生想交卷请举手，这就是一种高效机制，在很多连接中发现某个事件并快速定位 ### 11.2. 同步异步和阻塞非阻塞 1. 同步阻塞：服务员说快到你了，先别离开我后台看一眼马上通知你，客户在前台干等着啥也不干 2. 同步非阻塞：服务员说快到你了，先别离开。客户在前台边刷抖音边等着叫号 3. 异步阻塞：服务员说还要再等等，你先逛逛一会通知你，客户怕过号，拿着排号的小票等着，啥也不干，一直等着服务员通知 4. 异步非阻塞：服务员说还要再等等，你先逛逛一会通知你，客户拿着排号小票并且刷着抖音，等着服务员通知 ## 12. 年会抢红包 ### 12.1. 考虑红包设计 1. 发红包（List结构）： - redis的list结构，天生支持原子性  2. 抢红包： - 不加锁且原子性，还能支持高并发，每个人一次且有红包记录 3. 记红包（Hash结构）： - 盘点+汇总，防止作弊，同一个用户不可以抢两次 4. 拆红包算法（二倍均值法）： - 所有人抢到金额之和等于红包金额，不能超过，也不能少于 - 每个人至少抢到一分 - 要保证所有人抢到金额几率相等 ### 12.2. 二倍均值法 假设有10个人(N)，红包金额100元（M） `每次抢到的金额 = 随机区间（0，剩余红包金额M ➗ 剩余人数N） ×2 ）` 第一次： - 100➗10 × 2 = 20，所以第一个人的随即范围在（0，20）平均可以抢到10元。假设第一个人随机到10元，那么剩余金额100 - 10 = 90 第二次： - 90➗9 × 2 = 20，所有第二个人的随机范围同样是（0，20）平均可以抢到10元。假设第一个人随机到18元，那么剩余金额90- 18 = 72 第三次： - 72 ➗8 × 2 = 18，所有第二个人的随机范围同样是（0，18）平均可以抢到10元。假设第一个人随机到10元，那么剩余金额72- 10 = 62 ### 12.3. 发红包设计 1. 根据传入进来的总金额，以及包多少个包的两个参数 2. 利用二倍均值法拆成小红包 3. 将拆好的小红包保存到list结构中，并加入24小时过期，寓意是24小时将没有被领取的红包退回到发红包的用户中 ### 12.4. 抢红包设计 1. 首先验证某个用户是否抢过 红包 2. 没抢过则可以抢，利用list的随机弹出一个元素 3. 如果随机弹出来的为null，表示抢完了 ### 12.5. 最终的代码落地实现 ```java @RestController @RequestMapping("redPack") @Api(value = "抢红包") @Slf4j public class RedPackageController { //发送包预先将分好的金额存入List集合 public static final String RED_PACKAGE_KEY = "redpackage:"; //记录抢红包的用户 public static final String RED_PACKAGE_CONSUME_KEY = "redpackage:comsume:"; @Autowired private RedisTemplate redisTemplate; /** * 发送包 * @param totalMoney 一共多少金额 * @param redPackageNumber 拆分几个 * @return */ @GetMapping("send") @ApiOperation(value = "发送包",notes = "发送包") public String send(int totalMoney,int redPackageNumber) { //1. 拆红包，将总金额totalMoney拆分为redPackageNumber个子红包 Integer[] splitRedPackages = splitRedPackageAlgorithm(totalMoney,redPackageNumber); //拆分红包算法通过后获得多个子红包的数组 //2.发红包保存list结构,且放置过期事件 String key = RED_PACKAGE_KEY + IdUtil.simpleUUID(); redisTemplate.opsForList().leftPushAll(key,splitRedPackages); redisTemplate.expire(key,1, TimeUnit.DAYS); return key + "\t" + Ints.asList(Arrays.stream(splitRedPackages).mapToInt(Integer::valueOf).toArray()); } /** * 抢红包 * @param redPackkey 红包的redis的key * @param userId 用户信息ID * @return */ @GetMapping("rob") @ApiOperation(value = "抢红包",notes = "抢红包") public String robRedPackage(String redPackkey,String userId) { //1.验证某个用户是否强国红包，抢过不能重复抢 Object redPack = redisTemplate.opsForHash().get(RED_PACKAGE_CONSUME_KEY + redPackkey, userId); //2.没有抢过则可以抢 if (null == redPack) { //2.1. 从大红包（list）里面出队一个，作为该客户的红包，也就是抢到一个红包 Object userPack = redisTemplate.opsForList().leftPop(RED_PACKAGE_KEY + redPackkey); if (null != userPack) { //2.2. 抢到红包后需要记录进入hash结构，表示谁抢到了多少钱某个子红包的金额 redisTemplate.opsForHash().put(RED_PACKAGE_CONSUME_KEY + redPackkey,userId,userPack); log.info("用户：{}，抢到了{}红包",userId,userPack); //TODO 后续异步进入MySQL或是MQ进一步做统计处理，可以是每年发出红包个数抢到红包个数，年度总结等 return String.valueOf(userPack); } //2.2. 整个红包抢完了... return "errCodel-1,红包抢完了"; } //3. 某个用户抢过了，不可以重复抢 return "errCodel-2，你已经抢过了，不能重复抢"; } /** * 二倍均值算法 * @param totalMoney 一共多少金额 * @param redPackageNumber 拆分几个 * @return */ private Integer[] splitRedPackageAlgorithm(int totalMoney,int redPackageNumber) { Integer[] redPackageNumbers = new Integer[redPackageNumber]; //已经抢夺的红包金额，也就是被拆分塞进子红包的金额 int useMoney = 0; for (int i = 0; i < redPackageNumber; i++) { //判断是不是最后一次拆红包金额 if (i == redPackageNumber - 1) { redPackageNumbers[i] = totalMoney - useMoney; }else { //二倍均值法，每次拆分后塞进子红包的金额 = 随机取件（0，（剩余红包金额 ➗ 未被抢的剩余红包个数） × 2 ） int avgMoney = ((totalMoney - useMoney) / redPackageNumber - i) * 2; //取范围 redPackageNumbers[i] = 1 + new Random().nextInt(avgMoney - 1); } useMoney = useMoney + redPackageNumbers[i]; } return redPackageNumbers; } } ```