• 点评项目核心内容

    目录

    拦截器设置

    集群的session共享问题

     基于redis实现共享session登录

     创建bean对象技巧

    什么是缓存

    使用缓存来处理对象

    使用String类型缓存来处理集合

    缓存更新策略

    主动更新策略

    缓存穿透

     空串""和null的区别

    缓存null值解决穿透问题

    缓存雪崩

     缓存击穿

    互斥锁和逻辑过期介绍 

    基于互斥锁解决缓存穿透问题

    ​编辑

     下载JMeter模拟线程测试

    redis缓存工具类封装

    优惠卷秒杀

    全局唯一id

    优惠券添加

     优惠券秒杀下单

     JMeter线程测试遇到401错误

    超卖问题分析

    乐观锁

    版本号法

     CAS法

    一人一单的并发安全问题

    分布式锁

    ​编辑 什么是分布式锁

     分布式锁的实现

    基于Redis实现分布式锁的初级版本 

    线程存在问题

    线程阻塞超时自动删除后，线程完成释放别的线程的锁

     改进分布式锁（判断线程和存的是否一致）

    有并发安全分析 

     Redis的Lua脚本

     基于Redis的分布式锁实现思路

     还存在的问题

    Redission实现分布式锁

    Redisson可重入锁原理

    获取锁的Lua脚本 

     释放锁的Lua脚本

    Redisson分布式锁的原理

    Redis秒杀优化（暂未实现）

     达人探店

    点赞功能

     实现点赞排行榜功能

    关注和取关

    查看共同关注

    关注推送

    feed流模式

    Redis最佳实践

    Redis键值设计

    拦截器设置

    第一步，定义拦截器

    1. package com.hmdp.utils;
    2.  
    3. import com.hmdp.dto.UserDTO;
    4. import com.hmdp.entity.User;
    5. import org.springframework.web.servlet.HandlerInterceptor;
    6.  
    7. import javax.servlet.http.HttpServletRequest;
    8. import javax.servlet.http.HttpServletResponse;
    9. import javax.servlet.http.HttpSession;
    10.  
    11. public class LoginInterceptor implements HandlerInterceptor {
    12.     @Override
    13.     public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
    14.        //1、获取session
    15.         HttpSession session = request.getSession();
    16.         //2、获取session中的用户
    17.         Object user = session.getAttribute("user");
    18.         //3、判断用户是否存在
    19.         if (user == null) {
    20.             //4、不存在，拦截，返回401状态码
    21.             response.setStatus(401);
    22.             return false;
    23.         }
    24.         //5、存在，保存用户信息到ThreadLocal
    25.         UserHolder.saveUser((UserDTO) user);
    26.         //6、放行
    27.         return true;
    28.     }
    29.  
    30.     @Override
    31.     public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
    32.         //移除用户，避免内存泄漏
    33.         UserHolder.removeUser();
    34.     }
    35. }

    userhold类下

    1. package com.hmdp.utils;
    2.  
    3. import com.hmdp.dto.UserDTO;
    4.  
    5. public class UserHolder {
    6.     private static final ThreadLocal tl = new ThreadLocal<>();
    7.  
    8.     public static void saveUser(UserDTO user){
    9.         tl.set(user);
    10.     }
    11.  
    12.     public static UserDTO getUser(){
    13.         return tl.get();
    14.     }
    15.  
    16.     public static void removeUser(){
    17.         tl.remove();
    18.     }
    19. }

    第二步：配置文件（让拦截器生效）

    1. package com.hmdp.config;
    2.  
    3. import com.hmdp.utils.LoginInterceptor;
    4. import org.springframework.context.annotation.Configuration;
    5. import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
    6. import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
    7.  
    8. @Configuration
    9. public class MvcConfig  implements WebMvcConfigurer {
    10. //配置，添加拦截器，让之前的拦截器生效
    11.     @Override
    12.     public void addInterceptors(InterceptorRegistry registry) {
    13.         registry.addInterceptor(new LoginInterceptor())
    14.              //排除掉不需要拦截的路径
    15.                 .excludePathPatterns(
    16.                       "/shop/**",
    17.                         "/voucher/**",
    18.                         "/shop-type/**",
    19.                         "/upload/**",
    20.                         "/blog/hot",
    21.                         "/usr/code",
    22.                         "/usr/login"
    23.                 );
    24.     }
    25. }

    集群的session共享问题

    session共享问题：多态Tomcat并不共享session存储空间，当请求切换到不同tomcat服务是导致数据丢失问题

    session的代替为redis，满足数据共享，内存存储key、value结构

     基于redis实现共享session登录

    保存登录的用户信息，可以使用String结构，以JSON字符串来保存，比较直观：

     Hash结构可以将对象中的每个字段独立存储，可以针对单个字段做CRUD，并且内存占用更少:

     

     

     创建bean对象技巧

    自己创建的类，无法直接用@Autowire方式注入，因为他不属于 spring容器管理的。

    需要在创建的加入一个构造方法，然后在由其他由spring管理的类调用，然后在注入传入这个属性即可

    1. @Configuration
    2. public class MvcConfig  implements WebMvcConfigurer {
    3. //由spring管理的类注册
    4.     @Resource
    5.     private StringRedisTemplate stringRedisTemplate;
    6.  
    7. //配置，添加拦截器，让之前的拦截器生效
    8.     @Override
    9.     public void addInterceptors(InterceptorRegistry registry) {
    10.        //调用时传入这个即可
    11.         registry.addInterceptor(new LoginInterceptor(stringRedisTemplate))
    12.         
    13.     }
    1. public class LoginInterceptor implements HandlerInterceptor {
    2.  
    3. private StringRedisTemplate stringRedisTemplate;
    4.  
    5.     public LoginInterceptor(StringRedisTemplate stringRedisTemplate) {
    6.         this.stringRedisTemplate = stringRedisTemplate;
    7.     }
    8. }

    什么是缓存

    缓存就是数据交换的缓冲区（称作Cache，是存储数据的临时笛梵，一般读写性能较高）

    使用缓存来处理对象

    1.  //1、从redis查询商铺缓存，使用opsForValue接受对象，返回的是json串
    2.         String shopJson = stringRedisTemplate.opsForValue().get(key);
    3.  
    4.         //2、判断是否存在
    5.         if (StrUtil.isNotBlank(shopJson)){
    6.             //3、存在，直接返回缓存中的
    7.                //是json数据，则需要通过JSONUtil返回指定的对象即可
    8.             Shop shop= JSONUtil.toBean(shopJson, Shop.class);
    9.             return  Result.ok(shop);
    10.         }
    11.         //4、缓存中不存在，根据id查询数据库
    12.         Shop shop = this.getById(id);
    13.         //5、数据库中不存在，返回错误
    14.         if (shop==null){
    15.             return Result.fail("商铺不存在");
    16.         }
    17.         //6、存在，写入缓存
    18.         stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop));
    19.        //7、返回

    使用String类型缓存来处理集合

    1. @Resource
    2. private StringRedisTemplate stringRedisTemplate;
    3.     @Override
    4.     public Result shopTypeList() {
    5.         String key="shop_Type_List";
    6.         //查询缓存
    7.         String shopTypeJson = stringRedisTemplate.opsForValue().get(key);
    8.  
    9.         if (StrUtil.isNotBlank(shopTypeJson)){
    10.             //查到了，直接返回，json转list集合
    11.             List shopTypes = JSONUtil.toList(shopTypeJson, ShopType.class);
    12.             return Result.ok(shopTypes);
    13.         }
    14.         //缓存没查到，查数据库
    15.         List typeList = this.query().orderByAsc("sort").list();
    16.         //数据库没查到，返回错误
    17.         if (typeList==null){
    18.             return Result.fail("没有列表信息");
    19.         }
    20.         //数据库查到，缓存下
    21.         //list集合转json
    22.         String json = JSONUtil.toJsonStr(typeList);
    23.         stringRedisTemplate.opsForValue().set(key, json);

    缓存更新策略

     业务场景

    低一致性需求：使用内存淘汰机制，例如店铺类型的查询

    高一致性需求；主动更新，并以超时剔除作为兜底方案，例如店铺的详细信息

    主动更新策略

    操作缓存和数据库的三个问题需要考虑：

    1，删除缓存还是更新缓存？

    • 更新缓存：每次更新数据都会更新缓存，无效写操作较多（×）
    • 删除缓存：更新数据库时让缓存失效，查询时在更新缓存（√）

    2、如何保证缓存与数据库的操作同时成功或者失败？

    • 单体系统，将缓存与数据库操作放在一个事务
    • 分布式系统，利用TCC操作等分布式事务方案

    3、先操作缓存还是先操作数据库

    • 先操作数据库，在删除缓存

     缓存更新策略的最佳实践方案：

    1、低一致性需求：使用Redis自带的内存淘汰机制

    2、高一致性需求：主动更新，并以超时提出作为兜底方案

    读操作：

    • 缓存命中则直接返回
    • 缓存未命中则查询数据库，并写入缓存，设置超时时间

    写操作::

    • 先写数据库，然后在删除缓存
    • 要确保数据库与缓存操作的原子性

    缓存穿透

    我们使用Redis大部分情况都是通过Key查询对应的值，假如发送的请求传进来的key是不存在Redis中的，那么就查不到缓存，查不到缓存就会去数据库查询。假如有大量这样的请求，这些请求像“穿透”了缓存一样直接打在数据库上，这种现象就叫做缓存穿透。

    解决方案：

    1、缓存空对象（把无效的Key存进Redis中）。如果Redis查不到数据，数据库也查不到，我们把这个Key值保存进Redis，设置value="null"，当下次再通过这个Key查询时就不需要再查询数据库。这种处理方式肯定是有问题的，假如传进来的这个不存在的Key值每次都是随机的，那存进Redis也没有意义，会占用redis的内存空间，所以设置过期时间是有必要的，其次，当这个值一开始没有内容，我们查询数据库后 ，将null赋值给这个值，并存在redis中，而之后我们数据库新增了这个值，但是缓存中还是为null，这就会导致短期数据不一致，可以使用更新数据库删除那个缓存就可以解决。

    2、使用布隆过滤器。布隆过滤器的作用是某个 key 不存在，那么就一定不存在，它说某个 key 存在，那么很大可能是存在(存在一定的误判率)。于是我们可以在缓存之前再加一层布隆过滤器，在查询的时候先去布隆过滤器查询 key 是否存在，如果不存在就直接返回，这个布隆过滤器也存在一定的穿透风险。

     项目中解决缓存穿透的思路

     空串""和null的区别

    null表示的是一个对象的值，而非一个字符串。例如声明一个对象的引用，String aaa = null ;
    ""表示的是一个长度为0的空字符串。例如声明一个字符串String bbb = "" ;
    所以：null不指向任何对象，相当于没有任何值;而""代表一个长度为0的字符串。

    缓存null值解决穿透问题

    1.  public Result queryById(Long id) {
    2.         String key=CACHE_SHOP_KEY+id;
    3.         //1、从redis查询商铺缓存，使用opsForValue接受对象，返回的是json串
    4.         String shopJson = stringRedisTemplate.opsForValue().get(key);
    5.  
    6.         //2、判断是否存在
    7.         if (StrUtil.isNotBlank(shopJson)){
    8.             //3、存在，直接返回缓存中的
    9.                //是json数据，则需要通过JSONUtil返回指定的对象即可
    10.             Shop shop= JSONUtil.toBean(shopJson, Shop.class);
    11.             return  Result.ok(shop);
    12.         }
    13.         //4、缓存中不存在，判断是否命中的为空串""
    14.         if (shopJson!=null){
    15.             return Result.fail("商铺不存在");
    16.         }
    17.         Shop shop = this.getById(id);
    18.         //5、数据库中不存在，并返回错误
    19.         if (shop==null){
    20.             //插入一个空串,设置过期时间
    21.             stringRedisTemplate.opsForValue().set(key,"", CACHE_NULL_TTL, TimeUnit.MINUTES);
    22.             return Result.fail("商铺不存在");
    23.         }
    24.         //6、存在，写入缓存
    25.         stringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.MINUTES);
    26.        //7、返回
    27.         return  Result.ok(shop);
    28.     }

    缓存穿透产生的原因是什么?

    用户请求的数据在缓存中和数据库中都不存在，不断发起这样的请求给数据库带来巨大压力

    缓存穿透的解决方案有?

    • 缓存null值
    • 布隆过滤
    • 增强id的复杂度，避免被猜测id规律
    • 做好数据的基础格式校验
    • 加强用户权限校验
    • 做好热点参数的限流

    缓存雪崩

    当某一个时刻出现大规模的redis缓存失效的情况，就会导致大量的请求直接打在数据库上面，导致数据库压力巨大，如果在高并发的情况下，可能瞬间就会导致数据库宕机。这时候如果运维马上又重启数据库，马上又会有新的流量把数据库打死。这就是缓存雪崩。缓存雪崩是指在同一时段大量的缓存key同时失效或者Redis服务宕机，导致大量请求到达数据库，带来巨大压力。

    解决方案:

    • 给不同的Key的TTL添加随机值
    • 利用Redis集群提高服务的可用性
    • 给缓存业务添加降级限流策略
    • 给业务添加多级缓存

    1、在原有的失效时间上加上一个随机值，这样就避免了因为采用相同的过期时间导致的缓存雪崩。

    如果真的发生了缓存雪崩，有没有什么兜底的措施？

    2、使用熔断机制。当流量到达一定的阈值时，就直接返回“系统拥挤”之类的提示，防止过多的请求打在数据库上。至少能保证一部分用户是可以正常使用，其他用户多刷新几次也能得到结果。

    3、提高数据库的容灾能力，可以使用分库分表，读写分离的策略。

    4、为了防止Redis宕机导致缓存雪崩的问题，可以搭建Redis集群，提高Redis的容灾性

     缓存击穿

    其实跟缓存雪崩有点类似，缓存雪崩是大规模的key失效，而缓存击穿是一个热点的Key，有大并发集中对其进行访问，突然间这个Key失效了，导致大并发全部打在数据库上，导致数据库压力剧增。这种现象就叫做缓存击穿。

    解决方案：

    • 互斥锁
    • 逻辑过期（不过期）

    1、业务允许的话，对于热点的key可以设置永不过期的key。

    2、使用互斥锁。如果缓存失效的情况，只有拿到锁才可以查询数据库，降低了在同一时刻打在数据库上的请求，防止数据库打死。当然这样会导致系统的性能变差。

    多条线程同时访问数据库 

    互斥锁和逻辑过期介绍 

     

    基于互斥锁解决缓存穿透问题

    修改id查询店铺 ，基于互斥锁来解决缓存击穿问题

    1.     /**
    2.      * 互斥锁解决缓存击穿
    3.      * @param id
    4.      * @return
    5.      */
    6.     public Shop queryWithMutex(Long id){
    7.         String key=CACHE_SHOP_KEY+id;
    8.         //1、从redis查询商铺缓存，使用opsForValue接受对象，返回的是json串
    9.         String shopJson = stringRedisTemplate.opsForValue().get(key);
    10.  
    11.         //2、判断是否存在
    12.         if (StrUtil.isNotBlank(shopJson)){
    13.             //3、存在，直接返回缓存中的
    14.             //是json数据，则需要通过JSONUtil返回指定的对象即可
    15.             Shop shop= JSONUtil.toBean(shopJson, Shop.class);
    16.             return  shop;
    17.         }
    18.         //4、缓存中不存在，判断是否命中的为空串""
    19.         if (shopJson!=null){
    20.             return null;
    21.         }
    22.  
    23.         //4实现缓存重建
    24.         //4.1获取互斥锁
    25.         String lockKey=LOCK_SHOP_KEY+id;
    26.         Shop shop = null;
    27.         try {
    28.             boolean isLock = tryLock(lockKey);
    29.             //4.2判断是否获取成功
    30.             if (!isLock){
    31.                 //4.3失败，则休眠并重试
    32.                 Thread.sleep(99);
    33.                return queryWithMutex(id);
    34.             }
    35.             //4.4成功，根据id查询数据库
    36.             shop = this.getById(id);
    37.             //模拟重建延时
    38.             Thread.sleep(366);
    39.             //5、数据库中不存在，并返回错误
    40.             if (shop==null){
    41.                 //插入一个空串,设置过期时间
    42.                 stringRedisTemplate.opsForValue().set(key,"", CACHE_NULL_TTL, TimeUnit.MINUTES);
    43.                 return null;
    44.             }
    45.             //6、存在，写入缓存
    46.             stringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.MINUTES);
    47.         } catch (InterruptedException e) {
    48.            throw new RuntimeException(e);
    49.         } finally {
    50.             //7、释放互斥锁
    51.             unLock(lockKey);
    52.         }
    53.         //8、返回
    54.         return shop;
    55.     }

    基于逻辑过期解决缓存击穿问题

    1.   /**
    2.      * 逻辑辑过期解决缓存击穿
    3.      * @param id
    4.      * @return
    5.      */
    6.  //创建一个线程池
    7.     private static final ExecutorService CACHE_REBUILD_EXECUTOR= Executors.newFixedThreadPool(10);
    8.  
    9.     public Shop queryWithLogicalExpire(Long id){
    10.         String key=CACHE_SHOP_KEY+id;
    11.         //1、从redis查询商铺缓存，使用opsForValue接受对象，返回的是json串
    12.         String shopJson = stringRedisTemplate.opsForValue().get(key);
    13.  
    14.         //2、判断是否存在
    15.         if (StrUtil.isBlank(shopJson)){
    16.             //3不存在，直接返回null
    17.            return  null;
    18.         }
    19.         //4命中，需要把json反序列化为对象
    20.         RedisData redisData = JSONUtil.toBean(shopJson, RedisData.class);
    21.         JSONObject data = (JSONObject) redisData.getData();
    22.         Shop shop = JSONUtil.toBean(data, Shop.class);
    23.         LocalDateTime expireTime = redisData.getExpireTime();
    24.         //5、判断是否过期，是否在当前时间之后
    25.         if (expireTime.isAfter(LocalDateTime.now())){
    26.             //5。1没过期。直接返回店铺信息
    27.             return  shop;
    28.         }
    29.         //5.2已过期，需要缓存
    30.         //6、缓存重建
    31.         //6/1获取互斥锁
    32.         String lockKey= LOCK_SHOP_KEY+id;
    33.         boolean isLock = tryLock(lockKey);
    34.         //6.2判断是否获取锁成功
    35.         if (isLock){
    36.            //6.3成功，开启独立线程，实现缓存重建
    37.             CACHE_REBUILD_EXECUTOR.submit(()->{
    38.                 try {
    39.                     //重建缓存
    40.                     this.saveShop2Redis(id,20L);
    41.                 } catch (Exception e) {
    42.                     e.printStackTrace();
    43.                 } finally {
    44.                     //释放锁
    45.                     unLock(lockKey);
    46.                 }
    47.  
    48.             });
    49.         }
    50.  
    51.         return shop;
    52.     }

     下载JMeter模拟线程测试

     修改数据据库中的一点信息，会发现某一时刻重建前是旧数据，完成后是新数据

     

     控制台中只有一数据重建，一次是查询旧数据，一次为新数据重建

    redis缓存工具类封装

    基于stringRedisTemplate封装一个缓存工具类，满足下列需求:

    方法1:将任意va对象序列化为json并存储在string类型的key中，并且可以设置TTL过期时间

    方法2:将任意/ava对象序列化为json并存储在string类型的key中，并且可以设置逻辑过期时间，用于处理缓存击穿问题

    方法3:根据指定的key查询缓存，并反序列化为指定类型，利用缓存空值的方式解决缓存穿透问题

    方法4:根据指定的key查询缓存，并反序列化为指定类型，需要利用逻辑过期解决缓存击穿问题

    1. package com.hmdp.utils;
    2.  
    3. import cn.hutool.core.util.BooleanUtil;
    4. import cn.hutool.core.util.StrUtil;
    5. import cn.hutool.json.JSONObject;
    6. import cn.hutool.json.JSONUtil;
    7. import lombok.extern.slf4j.Slf4j;
    8. import org.springframework.data.redis.core.StringRedisTemplate;
    9. import org.springframework.stereotype.Component;
    10.  
    11.  
    12. import java.time.LocalDateTime;
    13. import java.util.concurrent.ExecutorService;
    14. import java.util.concurrent.Executors;
    15. import java.util.concurrent.TimeUnit;
    16. import java.util.function.Function;
    17.  
    18. import static com.hmdp.utils.RedisConstants.LOCK_SHOP_KEY;
    19.  
    20. @Slf4j
    21. @Component
    22.  
    23. /**
    24.  * 缓存工具类1
    25.  */
    26. public class CacheClient {
    27.     private final StringRedisTemplate stringRedisTemplate;
    28.  
    29.     public CacheClient(StringRedisTemplate stringRedisTemplate){
    30.         this.stringRedisTemplate = stringRedisTemplate;
    31.     }
    32.  
    33.     public void  set(String key, Object value, Long time, TimeUnit unit){
    34.         stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(value),time,unit);
    35.     }
    36.  
    37.     public void setWithLogicalExpire(String key, Object value, Long time, TimeUnit unit){
    38.         //设置逻辑过期
    39.         RedisData redisData = new RedisData();
    40.         redisData.setData(value);
    41.         redisData.setExpireTime(LocalDateTime.now().plusSeconds(unit.toSeconds(time)));
    42.         //写入Redis
    43.         stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(redisData));
    44.     }
    45.  
    46.     public  R queryWithPassThrough(
    47.             String keyPrefix, ID id, Class type, Function dbFallback, Long time, TimeUnit unit){
    48.         String key = keyPrefix + id;
    49.         //1.从redis查询商铺缓存
    50.         String json = stringRedisTemplate.opsForValue().get(key);
    51.         //2.判断是否存在
    52.         if (StrUtil.isNotBlank(json)){
    53.             //3.存在，直接返回
    54.             return JSONUtil.toBean(json, type);
    55.         }
    56.         //判断命中的是否是空值
    57.         if (json != null){
    58.             //返回一个错误信息
    59.             return null;
    60.         }
    61.         //4.不存在，根据id查询数据库
    62.         R r = dbFallback.apply(id);
    63.         //5.不存在，返回错误
    64.         if (r == null){
    65.             //将空值写入redis
    66.             stringRedisTemplate.opsForValue().set(key,"",2,TimeUnit.MINUTES);
    67.             //返回错误信息
    68.             return null;
    69.         }
    70.         //6.存在写入redis
    71.         this.set(key,r,time,unit);
    72.         return r;
    73.     }
    74.  
    75.     /**
    76.      * 逻辑删除解决缓存击穿
    77.      */
    78.  
    79.     private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);
    80.  
    81.     public  R queryWithLogicalExpire(
    82.             String keyPrefix, ID id, Class type, Function dbFallback, Long time, TimeUnit unit) {
    83.         String key = keyPrefix + id;
    84.         //1.从redis查商铺缓存
    85.         String json = stringRedisTemplate.opsForValue().get(key);
    86.         //2.判断是否存在
    87.         if (StrUtil.isBlank(json)) {
    88.             //3.存在，缓存中存的null
    89.             return null;
    90.         }
    91.         //4.命中，先把json反序列化为对象
    92.         RedisData redisData = JSONUtil.toBean(json, RedisData.class);
    93.         R r = JSONUtil.toBean((JSONObject) redisData.getData(), type);
    94.         LocalDateTime expireTime = redisData.getExpireTime();
    95.         //5.判断是否过期
    96.         if (expireTime.isAfter(LocalDateTime.now())) {
    97.             //5.1未过期，直接返回店铺信息
    98.             return r;
    99.         }
    100.         //5.2已过期，需要缓存重建
    101.         //6.缓存重建
    102.         //6.1获取互斥锁
    103.         String lockKey = LOCK_SHOP_KEY + id;
    104.         boolean isLock = tryLock(lockKey);
    105.         //6.2判断是否获取锁成功
    106.         if (isLock) {
    107.             //6.3 成功，再进行二次判断，查看缓存中是否有数据，因为有可能是别人刚刚重建完释放锁，刚好获取到了
    108.  
    109.             //6.4 开启独立线程，实现缓存重建
    110.             CACHE_REBUILD_EXECUTOR.submit(() -> {
    111.                 try {
    112.                     //查询数据库
    113.                     R r1 = dbFallback.apply(id);
    114.                     //写入redis
    115.                     this.setWithLogicalExpire(key, r1, time, unit);
    116.                 } catch (Exception e) {
    117.                     throw new RuntimeException(e);
    118.                 } finally {
    119.                     //释放锁
    120.                     unlock(lockKey);
    121.                 }
    122.             });
    123.         }
    124.         //6.6返回过期的商铺信息
    125.         return r;
    126.     }
    127.  
    128.  
    129.     //获取锁和开锁
    130.     private boolean tryLock(String key){
    131.         Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);
    132.         return BooleanUtil.isTrue(flag);
    133.     }
    134.     private void unlock(String key){
    135.         stringRedisTemplate.delete(key);
    136.     }
    137. }
    138.  
    139.

    优惠卷秒杀

    全局唯一id

    全局id生成器

    全局ID生成器，是一种在分布式系统下用来生成全局唯一id的工具，一般满足：

    • 唯一性
    • 高可用
    • 高性能
    • 递增性
    • 安全性

    为了增加ID的安全性，我们可以不直接使用Redis自增的数值，而是拼接一些其它信息：

     ID的组成部分:
    符号位:1bit，永远为0
    时间戳:31bit，以秒为单位，可以使用69年
    序列号:32bit，秒内的计数器，支持每秒产生2^32个不同ID

    1. @Component
    2. public class RedisWorker {
    3.  
    4.     private static final long BEGIN_TIMESTAMP=1640995000L;
    5.     private static final long COUNT_BITS=32;
    6.     @Resource
    7.     private StringRedisTemplate stringRedisTemplate;
    8.  
    9.   public long nextId(String keyPrefix){
    10.       //1、生成时间戳
    11.       LocalDateTime now=LocalDateTime.now();
    12.       long nowSecond= now.toEpochSecond(ZoneOffset.UTC);
    13.       long timestamp = nowSecond - BEGIN_TIMESTAMP;
    14.  
    15.       //2、生成序列号
    16.       //2、1获取当前时间精确到天
    17.       String date = now.format(DateTimeFormatter.ofPattern("yyyy:MM:dd"));
    18.       //2、2自增长
    19.       Long count = stringRedisTemplate.opsForValue().increment("icr:" + keyPrefix + ":" + date);
    20.  
    21.       //拼接并返回(先向左移把右边空出，然后在或，相当于加上)
    22.       return timestamp<
    23.   }
    24.  
    25. }

     测试类中

    1.     private ExecutorService es= Executors.newFixedThreadPool(500);
    2.     @Test
    3.     void testIdWordker() throws InterruptedException {
    4.         CountDownLatch latch = new CountDownLatch(300);
    5.  
    6.         Runnable task=()->{
    7.             for (int i=0;i<100;i++){
    8.                 long id = redisWorker.nextId("order");
    9.                 System.out.println("id="+id);
    10.             }
    11.             latch.countDown();
    12.         };
    13.  
    14.         long begin = System.currentTimeMillis();
    15.         for (int i = 0; i <100 ; i++) {
    16.             es.submit(task);
    17.         }
    18.         latch.await();
    19.         long end = System.currentTimeMillis();
    20.         System.out.println("time="+(end-begin));
    21.     }

    全局唯一id生成策略

    • UUID
    • redis自增
    • snowflake算法
    • 数据库自增 

    redis自增id策略

    • 每天一个key，方便统计订单量
    • id构造是 时间戳+计数器

    优惠券添加

    没有后台只能通过postman添加

    1. {
    2. "shopId": 1,
    3. "title": "200元代金券",
    4. "subTitle": "周六周末可用",
    5. "rules": "全场通用\\n可以叠加\\n仅限制堂食",
    6. "payValue": 16000,
    7. "actualValue": 20000,
    8. "type": 1,
    9. "stock": 100,
    10. "beginTime": "2023-08-01T00:00:00",
    11. "endTime": "2024-08-01T00:00:00"
    12. }

     

     优惠券秒杀下单

     下单时需要判断两点:
    秒杀是否开始或结束，如果尚未开始或已经结束则无法下单

    库存是否充足，不足则无法下单

    1. @Service
    2. public class VoucherOrderServiceImpl extends ServiceImpl implements IVoucherOrderService {
    3. @Resource
    4.     private ISeckillVoucherService seckillVoucherService;
    5. @Resource
    6. private RedisWorker redisWorker;
    7.     /**
    8.      * 优惠券下单
    9.      * @param voucherId
    10.      * @return
    11.      */
    12.     @Override
    13.     @Transactional
    14.     public Result seckillVoucher(Long voucherId) {
    15.         //1、查询优惠券
    16.         SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
    17.         //2、判断秒杀是否开始
    18.         if (voucher.getBeginTime().isAfter(LocalDateTime.now())){
    19.             //尚未开始
    20.             return Result.fail("秒杀还未开始");
    21.         }
    22.         //3、判断秒杀是否结束
    23.         if (voucher.getEndTime().isBefore(LocalDateTime.now())){
    24.             //已经结束
    25.             return Result.fail("秒杀已经结束");
    26.         }
    27.         //4、判断库存是否充足
    28.         if (voucher.getStock()<1){
    29.             //库存不足
    30.             return Result.fail("已经被抢完");
    31.         }
    32.         //5、扣减库存
    33.         boolean success = seckillVoucherService.update()
    34.                 .setSql("stock=stock-1")
    35.                 .eq("voucher_id", voucherId).update();
    36.        if (!success)return Result.fail("库存不足");
    37.         //6创建秒杀券订单
    38.         VoucherOrder voucherOrder = new VoucherOrder();
    39.         //6.1订单id
    40.         long orderId = redisWorker.nextId("order");
    41.         voucherOrder.setId(orderId);
    42.         //6.2用户id
    43.         Long userId = UserHolder.getUser().getId();
    44.         voucherOrder.setUserId(userId);
    45.         //6.3代金券id
    46.         voucherOrder.setVoucherId(voucherId);
    47.         //添加订单
    48.         save(voucherOrder);
    49.         //7返回订单id
    50.         return Result.ok(orderId);
    51.     }
    52. }

     

     JMeter线程测试遇到401错误

    这是未授权问题
    用 F12 打开开发者工具
    在network网络 里寻找相关信息:

    添加一个信息管理器

     订单出现了超卖

    超卖问题分析

    超卖问题是多线程安全问题，即在一个线程还没执行完，其他线程抢先执行，对同一个数据进行修改

    乐观锁

     乐观锁的关键是判断之前查询到的数据是否被修改，常见的方式有

    版本号法

    在修改之前查询一次版本号，若版本号不变则说明没有被其他线程修改，则正常进行数据修改，并让版本加一，若是版本号不一致，则不会执行

     

     CAS法

    直接比较数据是否发生了改变，若不变则说明安全

    如果弄数据是否和之前一致，会导致成功率低

    设置200个线程只卖出23个 

    1. @Service
    2. public class VoucherOrderServiceImpl extends ServiceImpl implements IVoucherOrderService {
    3. @Resource
    4.     private ISeckillVoucherService seckillVoucherService;
    5. @Resource
    6. private RedisWorker redisWorker;
    7.     /**
    8.      * 优惠券下单
    9.      * @param voucherId
    10.      * @return
    11.      */
    12.     @Override
    13.     @Transactional
    14.     public Result seckillVoucher(Long voucherId) {
    15.         //1、查询优惠券
    16.         SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
    17.         //2、判断秒杀是否开始
    18.         if (voucher.getBeginTime().isAfter(LocalDateTime.now())){
    19.             //尚未开始
    20.             return Result.fail("秒杀还未开始！");
    21.         }
    22.         //3、判断秒杀是否结束
    23.         if (voucher.getEndTime().isBefore(LocalDateTime.now())){
    24.             //已经结束
    25.             return Result.fail("秒杀已经结束");
    26.         }
    27.         //4、判断库存是否充足
    28.         if (voucher.getStock()<1){
    29.             //库存不足
    30.             return Result.fail("已经被抢完");
    31.         }
    32.         //5、扣减库存
    33.         boolean success = seckillVoucherService.update()
    34.                 .setSql("stock=stock-1") //set stock=stock-1
    35.                 .eq("voucher_id", voucherId).gt("stock", 0)//where id=? and stock=?只要大于0即可
    36.                 .update();
    37.        if (!success)return Result.fail("库存不足");
    38.         //6创建秒杀券订单
    39.         VoucherOrder voucherOrder = new VoucherOrder();
    40.         //6.1订单id
    41.         long orderId = redisWorker.nextId("order");
    42.         voucherOrder.setId(orderId);
    43.         //6.2用户id
    44.         Long userId = UserHolder.getUser().getId();
    45.         voucherOrder.setUserId(userId);
    46.         //6.3代金券id
    47.         voucherOrder.setVoucherId(voucherId);
    48.         //添加订单
    49.         save(voucherOrder);
    50.         //7返回订单id
    51.         return Result.ok(orderId);
    52.     }
    53. }

     超卖这样的线程安全问题，解决方案有哪些?


    1.悲观锁:添加同步锁，让线程串行执行

    • 优点:简单粗暴
    • 缺点:性能一般

    乐观锁:不加锁，在更新时判断是否有其它线程在修改

    • 优点:性能好
    • 缺点:存在成功率低的问题

    给整个this对象上锁

    优点是简单安全

    缺点是性能低，因为这样所有用户都被锁上了，我们的初衷是，单个用户中，只能单卖，这样就会导致其他用户也会受到影响

    1. @Service
    2. public class VoucherOrderServiceImpl extends ServiceImpl implements IVoucherOrderService {
    3. @Resource
    4.     private ISeckillVoucherService seckillVoucherService;
    5. @Resource
    6. private RedisWorker redisWorker;
    7.     /**
    8.      * 优惠券下单
    9.      * @param voucherId
    10.      * @return
    11.      */
    12.     @Override
    13.     public Result seckillVoucher(Long voucherId) {
    14.         //1、查询优惠券
    15.         SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
    16.         //2、判断秒杀是否开始
    17.         if (voucher.getBeginTime().isAfter(LocalDateTime.now())){
    18.             //尚未开始
    19.             return Result.fail("秒杀还未开始！");
    20.         }
    21.         //3、判断秒杀是否结束
    22.         if (voucher.getEndTime().isBefore(LocalDateTime.now())){
    23.             //已经结束
    24.             return Result.fail("秒杀已经结束");
    25.         }
    26.         //4、判断库存是否充足
    27.         if (voucher.getStock()<1){
    28.             //库存不足
    29.             return Result.fail("已经被抢完");
    30.         }
    31.  
    32.  
    33.         //7返回订单id
    34.         return createVoucherOrder(voucherId);
    35.     }
    36.  
    37.     @Transactional
    38.     public  synchronized Result createVoucherOrder(Long voucherId){
    39.         //5、一人一单
    40.         Long userId = UserHolder.getUser().getId();
    41.         //5.1查询订单
    42.         Integer count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
    43.         //判断是否已经存在
    44.         if (count>0){
    45.             //用户已经购买过
    46.             return Result.fail("你已经购买过这个券了");
    47.         }
    48.  
    49.  
    50.         //6、扣减库存
    51.         boolean success = seckillVoucherService.update()
    52.                 .setSql("stock=stock-1") //set stock=stock-1
    53.                 .eq("voucher_id", voucherId).gt("stock", 0)//where id=? and stock=?只要大于0即可
    54.                 .update();
    55.         if (!success)return Result.fail("库存不足");
    56.         //6创建秒杀券订单
    57.         VoucherOrder voucherOrder = new VoucherOrder();
    58.         //6.1订单id
    59.         long orderId = redisWorker.nextId("order");
    60.         voucherOrder.setId(orderId);
    61.         //6.2用户id
    62.  
    63.         voucherOrder.setUserId(userId);
    64.         //6.3代金券id
    65.         voucherOrder.setVoucherId(voucherId);
    66.         //添加订单
    67.         save(voucherOrder);
    68.         //返回订单id
    69.         return Result.ok(orderId);
    70.     }
    71. }

    悲观锁升级后 

    1. @Override
    2.     public Result seckillVoucher(Long voucherId) {
    3.         //1、查询优惠券
    4.         SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
    5.         //2、判断秒杀是否开始
    6.         if (voucher.getBeginTime().isAfter(LocalDateTime.now())){
    7.             //尚未开始
    8.             return Result.fail("秒杀还未开始！");
    9.         }
    10.         //3、判断秒杀是否结束
    11.         if (voucher.getEndTime().isBefore(LocalDateTime.now())){
    12.             //已经结束
    13.             return Result.fail("秒杀已经结束");
    14.         }
    15.         //4、判断库存是否充足
    16.         if (voucher.getStock()<1){
    17.             //库存不足
    18.             return Result.fail("已经被抢完");
    19.         }
    20.         Long userId = UserHolder.getUser().getId();
    21.  
    22.        synchronized (userId.toString().intern()){//intern获取源对象，new多少次都只是从常量池中寻找
    23.            //拿到当前对象的代理对象(事务的对象)
    24.            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
    25.            return proxy.createVoucherOrder(voucherId);//要想让事务生效，必须要有代理对象
    26.     }
    27. }
    28.     @Transactional
    29.     public   Result createVoucherOrder(Long voucherId){
    30.         //5、一人一单
    31.         Long userId = UserHolder.getUser().getId();
    32.         //5.1查询订单
    33.         Integer count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
    34.         //判断是否已经存在
    35.         if (count>0){
    36.             //用户已经购买过
    37.             return Result.fail("你已经购买过这个券了");
    38.         }
    39.  
    40.  
    41.         //6、扣减库存
    42.         boolean success = seckillVoucherService.update()
    43.                 .setSql("stock=stock-1") //set stock=stock-1
    44.                 .eq("voucher_id", voucherId).gt("stock", 0)//where id=? and stock=?只要大于0即可
    45.                 .update();
    46.         if (!success)return Result.fail("库存不足");
    47.         //6创建秒杀券订单
    48.         VoucherOrder voucherOrder = new VoucherOrder();
    49.         //6.1订单id
    50.         long orderId = redisWorker.nextId("order");
    51.         voucherOrder.setId(orderId);
    52.         //6.2用户id
    53.  
    54.         voucherOrder.setUserId(userId);
    55.         //6.3代金券id
    56.         voucherOrder.setVoucherId(voucherId);
    57.         //添加订单
    58.         save(voucherOrder);
    59.         //返回订单id
    60.         return Result.ok(orderId);
    61.     }

    一人一单的并发安全问题

    通过加锁可以解决在单机情况下的一人一单安全问题，但是在集群模式下就不行了。
    1.我们将服务启动两份，端口分别为8081和8082:

    2.然后修改nginx的conf目录下的nginx.conf文件，配置反向代理和负载均衡

    nginx配置实现了反代理和负载均衡 

     

     发现线程不安全

    集群下的锁监听器tomcat等不是同一个

    分布式锁

     什么是分布式锁

    分布式锁：满足分布式系统或集群模式下多进程可见并且互斥的锁

     分布式锁的实现

    分布式锁的核心是是实现多进程之间互斥，常见的有三种 

     实现分布式锁时需要实现两个基本方法

    获取锁：

    • 互斥:确保只能有一个线程获取锁
    • 非阻塞：尝试一次，成功返回true，失败返回false

    添加锁，nx是互斥（当存在则不执行），ex是设置超时时间

    local是键 thread1是值

    释放锁：

    • 手动释放
    • 超时释放：获取锁时添加一个超时时间

    释放锁，直接删除即可

     业务流程

    基于Redis实现分布式锁的初级版本 

    锁的类 

    1. public class SimpleRedisLock  implements ILock{
    2.  
    3.     private StringRedisTemplate stringRedisTemplate;
    4.     private String name;
    5.  
    6.     public SimpleRedisLock(StringRedisTemplate stringRedisTemplate, String name) {
    7.         this.stringRedisTemplate = stringRedisTemplate;
    8.         this.name = name;
    9.     }
    10.  
    11.     private static final String KEY_PREFIX="lock:";
    12.  
    13.     //获取锁
    14.     @Override
    15.     public boolean tryLock(long timeoutSec) {
    16.         //获取线程标识
    17.         long threadId = Thread.currentThread().getId();
    18.         //获取锁，这里的返回值是一个布尔型的包装类，直接返回有时会出现空指针异常
    19.         Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId + "", timeoutSec, TimeUnit.SECONDS);
    20.         //和真比较，就不会出现异常
    21.         return Boolean.TRUE.equals(success) ;
    22.     }
    23.  
    24.     //释放锁
    25.     @Override
    26.     public void unlock() {
    27.   stringRedisTemplate.delete(KEY_PREFIX+name);
    28.     }
    29. }

    1.     Long userId = UserHolder.getUser().getId();
    2.  
    3.         //创建锁对象
    4.         SimpleRedisLock lock = new SimpleRedisLock(stringRedisTemplate, "order:" + userId);
    5.         //获取锁
    6.         boolean isLock = lock.tryLock(12);
    7.  
    8.         //判断是否获取锁成功
    9.         if (!isLock){
    10.             //获取锁失败
    11.             return Result.fail("一个人只能下一单");
    12.         }
    13.         try {
    14.             //拿到当前对象的代理对象(事务的对象)
    15.             IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
    16.             return proxy.createVoucherOrder(voucherId);//要想让事务生效，必须要有代理对象
    17.         }  finally {
    18.             //释放锁
    19.             lock.unlock();
    20.         }
    线程存在问题
    线程阻塞超时自动删除后，线程完成释放别的线程的锁

    存在的线程阻塞超时自动删除后，线程释放别的线程的锁

     改进分布式锁（判断线程和存的是否一致）

    需求:修改之前的分布式锁实现，满足:

    1.在获取锁时存入线程标示(可以用UUID表示)

    2.在释放锁时先获取锁中的线程标示，判断是否与当前线程标示一致如果一致则释放锁
    如果不一致则不释放锁

    1.   public void unlock() {
    2.         //获取线程标识
    3.       String threadId=  ID_PREFIX+Thread.currentThread().getId();
    4.       //获取锁中的标识
    5.         String id=stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
    6.         //判断两个是否一致，从而判断是否为同一线程
    7.         if (threadId.equals(id)){
    8.             stringRedisTemplate.delete(KEY_PREFIX+name);
    9.         }
    有并发安全分析 

     这里有个并发问题，即当判断完相同时，发生了阻塞，没来得及删除锁，被redis超时释放后，下一个线程来获取后，之前那个线程阻塞完成，就会释放掉锁，但是此时这把锁的拥有者不是他。

    所以我们改进的是时候，应该保证，判断和删除在同一条语句中，即使用lua脚本可以保证原子性

     Redis的Lua脚本

    Redis提供了Lua脚本功能，在一个脚本中编写多条Redis命令，确保多条命令执行时的原子性。Lua是一种编程语言，它的基本语法大家可以参考网站: https://www.runoob.com/lua/lua-tutorial.html语法如下:

     如，我们要执行set name jack则脚本是这样

     列如。我们要先执行set name Rose，在执行get name ，则脚本如下

     需要用Redis命令来调用脚本，调用脚本的常见命令如下:

    例如，我们要执行 redis.call('set','name','jack') 这个脚本，语法如下: 

    脚本中可以从KEYS和ARGV数组获取这些参数:
    如果脚本中的key、value不想写死，可以作为参数传递。key类型参数会放入KEYS数组，其它参数会放入ARGV数组，在

     resoure下创建unlock.lua

    1. --比较线程标识与锁中是否一致
    2. if (redis.call('get',KEYS[1])==ARGV[1]) then
    3.     -- 释放锁 del key
    4.     return redis.call('del',KEYS[1])
    5. end
    6. return 0

    1.     private static final String KEY_PREFIX = "lock:";
    2.     private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";
    3.     private static final DefaultRedisScript UNLOCK_SCRIPT;
    4.  
    5.     static {
    6.         UNLOCK_SCRIPT = new DefaultRedisScript<>();
    7.         UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
    8.         UNLOCK_SCRIPT.setResultType(Long.class);
    9.     }
    10.  
    11.     //获取锁
    12.     @Override
    13.     public boolean tryLock(long timeoutSec) {
    14.         //获取线程标识
    15.         String threadId = ID_PREFIX + Thread.currentThread().getId();
    16.         //获取锁，这里的返回值是一个布尔型的包装类，直接返回有时会出现空指针异常
    17.         Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);
    18.         //和真比较，就不会出现异常
    19.         return Boolean.TRUE.equals(success);
    20.     }
    21.  
    22.     //释放锁
    23.     @Override
    24.     public void unlock() {
    25.         //调用Lia
    26.         stringRedisTemplate.execute(UNLOCK_SCRIPT,
    27.                 Collections.singletonList(KEY_PREFIX+name),
    28.                 ID_PREFIX+Thread.currentThread().getId());
    29.     }
     基于Redis的分布式锁实现思路
    • 利用set nxex获取锁，并设置过期时间，保存线程标示
    • 释放锁时先判断线程标示是否与自己一致，一致则删除锁

    特性:

    • 利用setnx满足互斥性
    • 利用set ex保证故障时锁依然能释放，避免死锁，提高安全性
    • 利用Redis集群保证高可用和高并发特性
     还存在的问题

    基于setnx实现的分布式锁还存在下面的问题

    不可重入：同一个线程无法多次获取同一把锁

    不可重试：获取锁只尝试一次就返回false，没有重试机制

    超时释放：锁超时释放虽然可以避免死锁，但如果是业务执行耗时较长，也会导致锁释放，存在安全隐患

    主从一致性：如果Redis提供了主从集群主从同步存在延迟，当主宕机时，如果从并同步主中的
    锁数据，则会出现锁实现

    Redission实现分布式锁

    Redisson是一个在Redis的基础上实现的ava驻内存数据网格 (In-Memory Data Grid)。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务，其中就包含了各种分布式锁的实现。

    官网地址: https://redisson.org
    GitHub地址: https://github.com/redisson/redisso 

    Redisson入门

    导入maven地址

    1.      
    2.         
    3.             org.redisson
    4.             redisson
    5.             3.13.6
    6.

    创建配置文件

    1. package com.hmdp.config;
    2.  
    3. import org.redisson.Redisson;
    4. import org.redisson.api.RedissonClient;
    5. import org.redisson.config.Config;
    6. import org.springframework.context.annotation.Bean;
    7. import org.springframework.context.annotation.Configuration;
    8.  
    9. @Configuration
    10. public class RedissonConfig {
    11.  
    12. @Bean
    13.     public RedissonClient redissonClient(){
    14.     //配置
    15.     Config config=new Config();
    16.     config.useSingleServer().setAddress("redis://localhost:6379");
    17.     //创建RedissonClient对象
    18.     return Redisson.create(config);
    19. }
    20. }

    1.        //创建锁对象
    2. //        SimpleRedisLock lock = new SimpleRedisLock(stringRedisTemplate, "order:" + userId);
    3.         RLock lock = redissonClient.getLock("lock:order:" + userId);
    4.  
    5.         //获取锁，数量分别是：获取锁的最大等待时间（期间会重试），锁自动释放时间，时间单位
    6.         boolean isLock = lock.tryLock(1,10,TimeUnit.SECONDS);
    7.         if(isLock){
    8.           try {
    9.              System.out.println("执行业务");}
    10.           finally {
    11.        // 释放锁
    12.            lock.unlock();}
    Redisson可重入锁原理

    获取锁的Lua脚本 

     释放锁的Lua脚本

    Redisson分布式锁的原理

     Redisson分布式锁原理：

    • 可重入：利用hash结构记录线程id和重入次数
    • 可重试：利用信号量和PubSub功能实现等待、唤醒，获取锁失败的重试机制
    • 超时续约：用于watchDog.每隔一段时间（releaseTime/3），重置超时时间

    总结 

    1)不可重入Redis分布式锁
    原理:利用setnx的互斥性;利用ex避免死锁;释放锁时判断线程标示
    缺陷:不可重入、无法重试、锁超时失效
    2)可重入的Redis分布式锁:
    原理:利用hash结构，记录线程标示和重入次数;利用watchDog延续锁时间;利用信号量控制锁重试等待
    缺陷: redis宕机引起锁失效问题
    3)Redisson的multiLock:
    原理:多个独立的Redis节点，必须在所有节点都获取重入锁，才算获取锁成功
    缺陷:运维成本高、实现复杂

    Redis秒杀优化（暂未实现）

     

     达人探店

    点赞功能

    1. /**
    2.      * 博客点赞
    3.      *
    4.      * @param id
    5.      * @return
    6.      */
    7.     @Override
    8.     public Result likeBlog(Long id) {
    9.         //获取blog实体
    10.         Blog blog = getById(id);
    11.         //获取博客id
    12.         Long blogId = blog.getId();
    13.         //获取登录当前用户id
    14.         Long userId = UserHolder.getUser().getId();
    15.  
    16.         //拼接key
    17.         String key = BLOG_LIKED_KEY + blogId;
    18.         //去redis中看有没有点赞过，查询这个key是否存在
    19.         Boolean isMember = stringRedisTemplate.opsForSet().isMember(key, userId.toString());
    20.         //不存在，即没点赞过
    21.         if (BooleanUtil.isFalse(isMember)) {
    22.             //修改数据库，让liked加1
    23.             boolean isSuccess = update().setSql("liked=liked+1").eq("id", id).update();
    24.             if (isSuccess) {
    25.                 //修改成功后，将自定义的blog的isLike(是否点赞过)改为true
    26.                // blog.setIsLike(true);不能，因为这不是数据库中的字段存不了
    27.                 //将这个点赞信息加到redis缓存中
    28.                 stringRedisTemplate.opsForSet().add(key, userId.toString());
    29.             }
    30.         } else {
    31.             //存在，即点赞过,则取消赞
    32.             boolean isSuccess = update().setSql("liked=liked-1").eq("id", id).update();
    33.             if (isSuccess) {
    34.                 //修改，删除
    35.                 stringRedisTemplate.opsForSet().remove(key, userId.toString());
    36.             }
    37.         }
    38.         return Result.ok();
    39.     }

     实现点赞排行榜功能

    1.   /**
    2.      * 博客点赞排序
    3.      * @param id
    4.      * @return
    5.      */
    6.     @Override
    7.     public Result queryBlogLikes(long id) {
    8.         String key =BLOG_LIKED_KEY+id;
    9.         //1查询top5的点赞用户 zrange key 0 4
    10.         Set topRange = stringRedisTemplate.opsForZSet().range(key, 0, 4);
    11.         if (topRange==null||topRange.isEmpty()){
    12.             return Result.ok(Collections.emptyList());
    13.         }
    14.         //2解析出其中的用户id
    15.         List ids = topRange.stream().map(Long::valueOf).collect(Collectors.toList());
    16.         String idStr = StrUtil.join(",", ids);
    17.         //3根据用户id查询用户where id in (5,1) order by field (id,5,1)
    18.         List userDTOS = userService.query().in("id", ids).last("order by field (id," + idStr + ")").list().stream()
    19.                 .map(user -> BeanUtil.copyProperties(user, UserDTO.class))
    20.                 .collect(Collectors.toList());
    21.         return Result.ok(userDTOS);
    22.     }

    关注和取关

    1.  @Override
    2.     public Result follow(Long followUserId, boolean isFollow) {
    3.         //1获取当前登录用户
    4.         Long userId = UserHolder.getUser().getId();
    5.         if (userId==null)return Result.fail("还没有登录");
    6.         Follow follow = new Follow();
    7.         //2判断为关注还是取关
    8.         if (isFollow){
    9.             //为关注，新增follow数据
    10.             follow.setUserId(userId);
    11.             follow.setFollowUserId(followUserId);
    12.             save(follow);
    13.         }else {
    14.             //为取关，删除follow数据
    15.             LambdaUpdateWrapper updateWrapper=new LambdaUpdateWrapper();
    16.             updateWrapper.eq(Follow::getUserId, userId)
    17.                     .eq(Follow::getFollowUserId,followUserId);
    18.             //删除数据
    19.             this.remove(updateWrapper);
    20.  
    21.         }
    22.         return Result.ok();
    23.     }
    24.  
    25.     @Override
    26.     public Result isFollow(Long followUserId) {
    27.         //1获取当前登录用户
    28.         Long userId = UserHolder.getUser().getId();
    29.         if (userId==null)return Result.fail("还没有登录");
    30.         Integer count = query().eq("user_id", userId).eq("follow_user_id", followUserId).count();
    31.         //判断是否有数据
    32.         return Result.ok(count>0);
    33.     }

    查看共同关注

    1. /**
    2.      * 被查看的人和我的共同关注
    3.      *
    4.      * @param checkedUserId
    5.      * @return
    6.      */
    7.     @Override
    8.     public Result commonConcernPerson(Long checkedUserId) {
    9.         //1获取当前登录用户
    10.         Long userId = UserHolder.getUser().getId();
    11.         if (userId == null) return Result.fail("还没有登录");
    12.         String key = "follows:" + userId;
    13.         //2求交集
    14.         String key2 = "follows:" + checkedUserId;
    15.         Set intersect = stringRedisTemplate.opsForSet().intersect(key, key2);
    16.       if (intersect==null||intersect.isEmpty()){
    17.           //没有交集
    18.           return Result.ok(Collections.emptyList());
    19.       }
    20.         //解析出id集合
    21.         List ids = intersect.stream().map(Long::valueOf).collect(Collectors.toList());
    22.         //4、根据id查询用户，转为userDto
    23.         List userDTOS = userService.listByIds(ids)
    24.                 .stream()
    25.                 .map(user -> BeanUtil.copyProperties(user, UserDTO.class ))
    26.                 .collect(Collectors.toList());
    27.         return Result.ok(userDTOS);
    28.     }

    关注推送

    关注推送也叫做Feed流，直译为投喂。为用户持续的提供“沉浸式”的体验，通过无限下拉刷新获取新的信息。

    feed流模式

    Feed流产品有两种常见模式:
    Timeline:不做内容筛选，简单的按照内容发布时间排序，常用于好友或关注。例如朋友圈优点:信息全面，不会有缺失。并且实现也相对简单
    缺点:信息噪音较多，用户不一定感兴趣，内容获取效率低

    智能排序:利用智能算法屏蔽掉违规的、用户不感兴趣的内容。推送用户感兴趣信息来吸引用户> 优点: 投喂用户感兴趣信息，用户粘度很高，容易沉迷
    缺点:如果算法不精准，可能起到反作用

     

    对于普通人：直接发送到他的粉丝下 

    对于大v：活跃粉丝直接给他推送，不活跃粉丝放在收件箱，等他要读的时候推

    feed流实现方案

     需求

     修改新增探店笔记的业务，在保存blog到数据库的同时，推送到粉丝的收件箱

    收件箱满足可以根据时间戳排序，必须用Redis的数据结构实现

    查询收件箱数据时，可以实现分页查询

    不能使用传统的分页，得使用滚动分页

    1.     
    2. @GetMapping("/of/follow")
    3. public Result queryBlogOfFollow(
    4. @RequestParam("lastId") Long max,@RequestParam(value = "offset",defaultValue = "0") Integer offset){
    5.         return blogService.queryBlogOfFollow(max,offset);
    6. }

    1.     @Override
    2.     public Result queryBlogOfFollow(Long max, Integer offset) {
    3.         //1获取当前用户
    4.         Long userId = UserHolder.getUser().getId();
    5.         //2查询收件箱 ZREVRANGEBYSCORE key Max Min LIMiT offset count
    6.         String key =FEED_KEY+userId;
    7.         Set> typedTuples = stringRedisTemplate.opsForZSet()
    8.                 .reverseRangeByScoreWithScores(key, 0, max, offset, 2);
    9.         //3判断是否为空
    10.         if (typedTuples==null||typedTuples.isEmpty()){
    11.             return Result.ok();
    12.         }
    13.         //4解析数据：blogId，minTime(时间戳)，offset
    14.     List ids=new ArrayList<>(typedTuples.size());
    15.         long minTime=0;
    16.         int offNum=1;
    17.         //统计有多次offNum（和最小的相同的个数）
    18.         for (ZSetOperations.TypedTuple tuple : typedTuples) {
    19.             //获取id
    20.             ids.add(Long.valueOf(tuple.getValue()));
    21.             //获取分数（时间戳）
    22.             long time = tuple.getScore().longValue();
    23.             if (time==minTime){
    24.                 offNum++;
    25.             }else {
    26.                 minTime=time;
    27.                 offNum=1;
    28.             }
    29.         }
    30.         //5,根据id查询blog
    31.         String idStr =StrUtil.join(",", ids);
    32.         List blogs = query().in("id", ids).last("order by field (id," + idStr + ")").list();
    33.  
    34.         //把博客相关信息点赞填充
    35.         for (Blog blog : blogs) {
    36.             //查询blog有关用户
    37.             queryBlogUser(blog);
    38.             isLikeBlog(blog);
    39.         }
    40.  
    41.         //5封装并返回
    42.         ScrollResult scrollResult = new ScrollResult();
    43.         scrollResult.setList(blogs);
    44.         scrollResult.setOffset(offNum);
    45.         scrollResult.setMinTime(minTime);
    46.         return Result.ok(scrollResult);
    47.     }

    Redis最佳实践

    Redis键值设计

    优雅的key结构

    Redis的key虽然可以自定义，但最好遵循下面结构最佳实践的约定：

    • 遵循基本格式：[业务名称]:[数据名]:[id]
    • 长度不超过44字节
    • 不包含特殊字符

    例如：在登录业务，保存用户信息，key是这样：login:user:1

    优点：

    • 可读强
    • 避免key冲突
    • 方便管理

    更节省内存：key是string类型，底层编码包含int、embstr、和raw三种，embstr在小于44字节使用，采用连续的空间，内存占用更小        

  • 相关阅读:
    60 条 rsync 常用命令及其说明
    HSRP协议（思科私有）/VRRP协议（公有）
    最新清理删除Mac电脑内存空间方法教程
    JVM第一话 -- JVM入门详解以及运行时数据区分析
    亲测解决no module named ‘PyQt5.QtCore‘
    C++ 学习宝藏网站分享
    Opncv 实现拍照、颜色识别和阈值选取
    工业控制系统安全标准
    web前端工程师面试之路
    java 基于微信小程序的饭店外卖点餐系统 uniapp小程序
  • 原文地址：https://blog.csdn.net/weixin_60719453/article/details/132125271