devbox
很楠不爱3
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

《黑马点评》简版笔记

作者：很楠不爱3 | 2024-06-16 01:29:08

赞

踩

《黑马点评》简版笔记

本套笔记基于黑马程序员《黑马点评》项目，供大家参考

目录

短信登入- Redis代替session的业务流程

Controller接口

Service层

拦截器

缓存

  Controller接口

缓存更新策略 

 Service层之更新

缓存穿透

 缓存雪崩 ​编辑

 缓存击穿

利用互斥锁解决缓存击穿问题

利用逻辑过期解决缓存击穿问题

优惠卷秒杀

 Redis实现全局唯一ID

 库存超卖问题

乐观锁

 一人一单问题

悲观锁

集群下的线程并发安全问题

分布式锁

 锁误删操作

锁释放过程发生阻塞

基于setnx所存在的问题

 基于Redission实现的分布式锁

基于Redis的秒杀优化

秒杀业务完整代码

达人探店

 点赞

好友关注

关注功能实现

共同关注

Feed流实现方案

​编辑

附件商铺

 导入店铺数据到GEO

实现附件商铺功能

用户签到

实现签到功能

实现统计连续签到功能

UV统计

短信登入- Redis代替session的业务流程

验证码我们使用手机号作为Key，value数据结构使用String；用户信息我们以随机token作为Key，Hash作为value的数据结构，并将随机token存储在浏览器前端每次请求都会携带这个随机的token.

Controller接口

主要实现客户向后端发送验证码，登入，根据/me接口查看当前用户是否有权限，进入界面

  1. /**
  2.  * 发送手机验证码
  3.  */
  4. @PostMapping("/code")
  5. public Result sendCode(@RequestParam("phone") String phone, HttpSession session) {
  6.     // 发送短信验证码并保存验证码
  7.     return userService.sendCode(phone, session);
  8. }
  9.  
  10. /**
  11.  * 登录功能
  12.  * @param loginForm 登录参数，包含手机号、验证码；或者手机号、密码
  13.  */
  14. @PostMapping("/login")
  15. public Result login(@RequestBody LoginFormDTO loginForm, HttpSession session){
  16.     // 实现登录功能
  17.     return userService.login(loginForm, session);
  18. }
  19.  
  20. /**
  21.  * 登入校验功能
  22.  * @return 返回当前登入的用户
  23.  */
  24. @GetMapping("/me")
  25. public Result me(){
  26.     // 获取当前登录的用户并返回
  27.     return Result.ok(ThreadLocalUtls.getUser());
  28. }
  29.

Service层

验证码匹配

  1. @Override
  2. public Result sendCode(String phone, HttpSession session) {
  3.     // 1、判断手机号是否合法
  4.     if (RegexUtils.isPhoneInvalid(phone)) {
  5.         return Result.fail("手机号格式不正确");
  6.     }
  7.     // 2、手机号合法，生成验证码，并保存到Redis中
  8.     String code = RandomUtil.randomNumbers(6);
  9.     stringRedisTemplate.opsForValue().set(LOGIN_CODE_KEY + phone, code,
  10.             RedisConstants.LOGIN_CODE_TTL, TimeUnit.MINUTES);
  11.     // 3、发送验证码
  12.     log.info("验证码:{}", code);
  13.     return Result.ok();
  14. }

实现用户登入功能：将UserDto对象存储在Redis中的时候，我们将对象转化为hash，其中hash的key和value值都需要是String类型，我们需要对对象的字段值进行处理转为String类型

设置Redis过期时间，我们需要完成在客户没有交互后的30钟内将其过期，而不是用户登入后30就将其强制过期

  1. /**
  2.  * 用户登录
  3.  *
  4.  * @param loginForm
  5.  * @param session
  6.  * @return
  7.  */
  8. @Override
  9. public Result login(LoginFormDTO loginForm, HttpSession session) {
  10.     String phone = loginForm.getPhone();
  11.     String code = loginForm.getCode();
  12.     // 1、判断手机号是否合法
  13.     if (RegexUtils.isPhoneInvalid(phone)) {
  14.         return Result.fail("手机号格式不正确");
  15.     }
  16.     // 2、判断验证码是否正确
  17.     String redisCode = stringRedisTemplate.opsForValue().get(LOGIN_CODE_KEY + phone);
  18.     if (code == null || !code.equals(redisCode)) {
  19.         return Result.fail("验证码不正确");
  20.     }
  21.     // 3、判断手机号是否是已存在的用户
  22.     User user = this.getOne(new LambdaQueryWrapper<User>()
  23.             .eq(User::getPhone, phone));
  24.     if (Objects.isNull(user)) {
  25.         // 用户不存在，需要注册
  26.         user = createUserWithPhone(phone);
  27.     }
  28.     // 4、保存用户信息到Redis中，便于后面逻辑的判断（比如登录判断、随时取用户信息，减少对数据库的查询）
  29.     UserDTO userDTO = BeanUtil.copyProperties(user, UserDTO.class);
  30.     // 将对象中字段全部转成string类型，StringRedisTemplate只能存字符串类型的数据
  31.     Map<String, Object> userMap = BeanUtil.beanToMap(userDTO, new HashMap<>(),
  32.             CopyOptions.create().setIgnoreNullValue(true).
  33.                     setFieldValueEditor((fieldName, fieldValue) -> fieldValue.toString()));
  34.     String token = UUID.randomUUID().toString(true);
  35.  
  36.     String tokenKey = LOGIN_USER_KEY + token;
  37.     stringRedisTemplate.opsForHash().putAll(tokenKey, userMap);
  38.     log.info(tokenKey);
  39.     stringRedisTemplate.expire(tokenKey, LOGIN_USER_TTL, TimeUnit.MINUTES);
  40.  
  41.     // 登录成功，将用户信息保存到ThreadLocal中，方便后续登录校验
  42.     //将用户信息保存到UserDTO，抹去了密码，实现了信息的安全
  43.     ThreadLocalUtls.saveUser(BeanUtil.copyProperties(userMap, UserDTO.class));
  44.     return Result.ok(token);
  45. }
  46.  
  47. /**
  48.  * 根据手机号创建用户并保存
  49.  *
  50.  * @param phone
  51.  * @return
  52.  */
  53. private User createUserWithPhone(String phone) {
  54.     User user = new User();
  55.     user.setPhone(phone);
  56.     user.setNickName(SystemConstants.USER_NICK_NAME_PREFIX + RandomUtil.randomString(10));
  57.     this.save(user);
  58.     return user;
  59. }

拦截器

第一个拦截器，我们获取当前用户放在ThreadLocal，如果没有就放行，并刷新有效期，对所有路径有效。

第二个拦截器，我们只判断当前用户是否登入，ThreadLocal是否有该用户

判断当前用户是否已登入

  1. @Override
  2. public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
  3.     // 判断当前用户是否已登录
  4.     if (ThreadLocalUtls.getUser() == null){
  5.         // 当前用户未登录，直接拦截
  6.         response.setStatus(HttpStatus.HTTP_UNAUTHORIZED);
  7.         return false;
  8.     }
  9.     // 用户存在，直接放行
  10.     return true;
  11. }

刷新Redis中当前用户的有效时间，拦截器没有被注解标识，因此RedisTemplate需要通过构造器的方法进行注入

  1. public class RefreshTokenInterceptor implements HandlerInterceptor {
  2.  
  3.     private StringRedisTemplate stringRedisTemplate;
  4.  
  5.     public RefreshTokenInterceptor(StringRedisTemplate stringRedisTemplate) {
  6.         this.stringRedisTemplate = stringRedisTemplate;
  7.     }
  8.  
  9.     @Override
  10.     public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
  11.         // 1、获取token，并判断token是否存在
  12.         String token = request.getHeader("authorization");
  13.         if (StrUtil.isBlank(token)){
  14.             // token不存在，说明当前用户未登录，不需要刷新直接放行
  15.             return true;
  16.         }
  17.         // 2、判断用户是否存在
  18.         String tokenKey = LOGIN_USER_KEY + token;
  19.         Map<Object, Object> userMap = stringRedisTemplate.opsForHash().entries(tokenKey);
  20.         if (userMap.isEmpty()){
  21.             // 用户不存在，说明当前用户未登录，不需要刷新直接放行
  22.             return true;
  23.         }
  24.         // 3、用户存在，则将用户信息保存到ThreadLocal中，方便后续逻辑处理，比如：方便获取和使用用户信息，Redis获取用户信息是具有侵入性的
  25.         UserDTO userDTO = BeanUtil.fillBeanWithMap(userMap, new UserDTO(), false);
  26.         ThreadLocalUtls.saveUser(BeanUtil.copyProperties(userMap, UserDTO.class));
  27.         // 4、刷新token有效期
  28.         stringRedisTemplate.expire(token, LOGIN_USER_TTL, TimeUnit.MINUTES);
  29.         return true;
  30.     }
  31. }

 将拦截器进行配置，order值越小越先执行

  1. @Configuration
  2. public class WebMvcConfig implements WebMvcConfigurer {
  3.  
  4.     @Resource
  5.     private StringRedisTemplate stringRedisTemplate;
  6.  
  7.     @Override
  8.     public void addInterceptors(InterceptorRegistry registry) {
  9.         // 添加登录拦截器
  10.         registry.addInterceptor(new LoginInterceptor())
  11.                 // 设置放行请求
  12.                 .excludePathPatterns(
  13.                         "/user/code",
  14.                         "/user/login",
  15.                         "/blog/hot",
  16.                         "/shop/**",
  17.                         "/shop-type/**",
  18.                         "/upload/**",
  19.                         "/voucher/**",
  20.                         // knife4j接口文档请求
  21.                         "/doc.html",
  22.                         "/webjars/**",
  23.                         "/swagger-resources",
  24.                         "/v2/api-docs"
  25.                 ).order(1); // 优先级默认都是0，值越大优先级越低
  26.         // 添加刷新token的拦截器
  27.         registry.addInterceptor(new RefreshTokenInterceptor(stringRedisTemplate)).addPathPatterns("/**").order(0);
  28.     }
  29. }

缓存

我们将商品信息直接添加在Redis中，当我们在Redis中找不到的时候再将数据从数据库中更新到Redis中去，提高查询速度，但是会出现数据库和Redis的一致性问题

  Controller接口

  1. /**
  2.  * 根据id查询商铺信息
  3.  *
  4.  * @param id 商铺id
  5.  * @return 商铺详情数据
  6.  */
  7. @GetMapping("/{id}")
  8. public Result queryShopById(@PathVariable("id") Long id) {
  9.     return shopService.queryById(id);
  10. }
  11.  
  12. /**
  13.  * 更新商铺信息
  14.  *
  15.  * @param shop 商铺数据
  16.  * @return 无
  17.  */
  18. @PutMapping
  19. public Result updateShop(@RequestBody Shop shop) {
  20.     return shopService.updateShop(shop);
  21. }

缓存更新策略 

 我们先更新数据后，删除缓存；当查询缓存的时候未命中，我们查询数据库，将数据写入缓存；我们一定是先更新数据库后，删除缓存，而不是删除缓存后更新数据库，降低数据的不一致性。并设置超时时间，保证数据的强一致性

最佳实践方案：

 Service层之更新

  1. @Transactional
  2. @Override
  3. public Result updateShop(Shop shop) {
  4.     // 参数校验, 略
  5.  
  6.     // 1、更新数据库中的店铺数据
  7.     boolean f = this.updateById(shop);
  8.     if (!f) {
  9.         // 缓存更新失败，抛出异常，事务回滚
  10.         throw new RuntimeException("数据库更新失败");
  11.     }
  12.     // 2、删除缓存
  13.     f = stringRedisTemplate.delete(CACHE_SHOP_KEY + shop.getId());
  14.     if (!f) {
  15.         // 缓存删除失败，抛出异常，事务回滚
  16.         throw new RuntimeException("缓存删除失败");
  17.     }
  18.     return Result.ok();
  19. }

缓存穿透

当数据库和Redis中都没有数据的时候，这样缓存就不会生效，请求将直接打到数据库中

 我们修改Service层queryById(Long id)的处理逻辑：

  • 当我们从数据库查询对象为null的时候，我们向redis中放空字符串
  • 当我们从Redis中查询的时候，不为空说明是null或者是空字符串，如果是空字符串我们直接返回“商铺信息为空”，如果是null则从数据库中查询数据
  1. /**
  2.  * 根据id查询数据（处理缓存穿透）
  3.  *
  4.  * @param keyPrefix  key前缀
  5.  * @param id         查询id
  6.  * @param type       查询的数据类型
  7.  * @param dbFallback 根据id查询数据的函数
  8.  * @param timeout    有效期
  9.  * @param unit       有效期的时间单位
  10.  * @param <T>
  11.  * @param <ID>
  12.  * @return
  13.  */
  14. public <T, ID> T handleCachePenetration(String keyPrefix, ID id, Class<T> type,
  15.                                         Function<ID, T> dbFallback, Long timeout, TimeUnit unit) {
  16.     String key = keyPrefix + id;
  17.     // 1、从Redis中查询店铺数据
  18.     String jsonStr = stringRedisTemplate.opsForValue().get(key);
  19.  
  20.     T t = null;
  21.     // 2、判断缓存是否命中
  22.     if (StrUtil.isNotBlank(jsonStr)) {
  23.         // 2.1 缓存命中，直接返回店铺数据
  24.         t = JSONUtil.toBean(jsonStr, type);
  25.         return t;
  26.     }
  27.  
  28.     // 2.2 缓存未命中，判断缓存中查询的数据是否是空字符串(isNotBlank把null和空字符串给排除了)
  29.     if (Objects.nonNull(jsonStr)) {
  30.         // 2.2.1 当前数据是空字符串（说明该数据是之前缓存的空对象），直接返回失败信息
  31.         return null;
  32.     }
  33.     // 2.2.2 当前数据是null，则从数据库中查询店铺数据
  34.     t = dbFallback.apply(id);
  35.  
  36.     // 4、判断数据库是否存在店铺数据
  37.     if (Objects.isNull(t)) {
  38.         // 4.1 数据库中不存在，缓存空对象（解决缓存穿透），返回失败信息
  39.         this.set(key, "", CACHE_NULL_TTL, TimeUnit.SECONDS);
  40.         return null;
  41.     }
  42.     // 4.2 数据库中存在，重建缓存，并返回店铺数据
  43.     this.set(key, t, timeout, unit);
  44.     return t;
  45. }

 缓存雪崩 

 缓存击穿

缓存击穿问题也叫热点Key问题，就是一个被高并发访问并且缓存重建业务较复杂的key突然失效了，无数的请求访问会在瞬间给数据库带来巨大的冲击。

 互斥锁：就是当线程1获得锁后，查询数据库重建缓冲数据，但是再这个时间内，其他线程需要处于一个等待的时间

逻辑过期：线程1获得锁后，委托另一个线程2完成重建缓冲数据和释放锁的工作，同时线程1和其他没有获得锁的线程，返回一个过期的数据。（过期时间作为value值放在Redis）

互斥锁方案更加强调数据的一致性，逻辑过期更加强调数据的可用性

利用互斥锁解决缓存击穿问题

我们使用Redis中的setnx lock 1，作为锁，当数据库中不存在的时候结果返回1，数据库已经存在则会返回0；缓存数据后，我们释放锁del lock

利用逻辑过期解决缓存击穿问题

这里我们先不考虑缓存穿透的问题，即未命中的时候是null还是空字符串

 我们首先需要创建如下的数据结构，记录过期时间

  1. @Data
  2. public class RedisData {
  3.     /**
  4.      * 过期时间
  5.      */
  6.     private LocalDateTime expireTime;
  7.     /**
  8.      * 缓存数据
  9.      */
  10.     private Object data;
  11. }
  1. /**
  2.  * 根据id查询数据（处理缓存击穿）
  3.  *
  4.  * @param keyPrefix  key前缀
  5.  * @param id         查询id
  6.  * @param type       查询的数据类型
  7.  * @param dbFallback 根据id查询数据的函数
  8.  * @param timeout    有效期
  9.  * @param unit       有效期的时间单位
  10.  * @param <T>
  11.  * @param <ID>
  12.  * @return
  13.  */
  14. public <T, ID> T handleCacheBreakdown(String keyPrefix, ID id, Class<T> type,
  15.                                       Function<ID, T> dbFallback, Long timeout, TimeUnit unit) {
  16.     String key = keyPrefix + id;
  17.     // 1、从Redis中查询店铺数据，并判断缓存是否命中
  18.     String jsonStr = stringRedisTemplate.opsForValue().get(key);
  19.     if (StrUtil.isBlank(jsonStr)) {
  20.         // 1.1 缓存未命中，直接返回失败信息
  21.         return null;
  22.     }
  23.     // 1.2 缓存命中，将JSON字符串反序列化未对象，并判断缓存数据是否逻辑过期
  24.     RedisData redisData = JSONUtil.toBean(jsonStr, RedisData.class);
  25.     // 这里需要先转成JSONObject再转成反序列化，否则可能无法正确映射Shop的字段
  26.     JSONObject data = (JSONObject) redisData.getData();
  27.     T t = JSONUtil.toBean(data, type);
  28.     LocalDateTime expireTime = redisData.getExpireTime();
  29.     if (expireTime.isAfter(LocalDateTime.now())) {
  30.         // 当前缓存数据未过期，直接返回
  31.         return t;
  32.     }
  33.  
  34.     // 2、缓存数据已过期，获取互斥锁，并且重建缓存
  35.     String lockKey = LOCK_SHOP_KEY + id;
  36.     boolean isLock = tryLock(lockKey);
  37.     if (isLock) {
  38.         // 获取锁成功，开启一个子线程去重建缓存
  39.         CACHE_REBUILD_EXECUTOR.submit(() -> {
  40.             try {
  41.                 // 查询数据库
  42.                 T t1 = dbFallback.apply(id);
  43.                 // 将查询到的数据保存到Redis
  44.                 this.setWithLogicalExpire(key, t1, timeout, unit);
  45.             } finally {
  46.                 unlock(lockKey);
  47.             }
  48.         });
  49.     }
  50.  
  51.     // 3、获取锁失败，再次查询缓存，判断缓存是否重建（这里双检是有必要的）
  52.     jsonStr = stringRedisTemplate.opsForValue().get(key);
  53.     if (StrUtil.isBlank(jsonStr)) {
  54.         // 3.1 缓存未命中，直接返回失败信息
  55.         return null;
  56.     }
  57.     // 3.2 缓存命中，将JSON字符串反序列化未对象，并判断缓存数据是否逻辑过期
  58.     redisData = JSONUtil.toBean(jsonStr, RedisData.class);
  59.     // 这里需要先转成JSONObject再转成反序列化，否则可能无法正确映射Shop的字段
  60.     data = (JSONObject) redisData.getData();
  61.     t = JSONUtil.toBean(data, type);
  62.     expireTime = redisData.getExpireTime();
  63.     if (expireTime.isAfter(LocalDateTime.now())) {
  64.         // 当前缓存数据未过期，直接返回
  65.         return t;
  66.     }
  67.  
  68.     // 4、返回过期数据
  69.     return t;
  70. }

 获取锁和释放锁

  1. /**
  2.  * 获取锁
  3.  *
  4.  * @param key
  5.  * @return
  6.  */
  7. private boolean tryLock(String key) {
  8.     Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);
  9.     // 拆箱要判空，防止NPE
  10.     return BooleanUtil.isTrue(flag);
  11. }
  12.  
  13. /**
  14.  * 释放锁
  15.  *
  16.  * @param key
  17.  */
  18. private void unlock(String key) {
  19.     stringRedisTemplate.delete(key);
  20. }

优惠卷秒杀

 Redis实现全局唯一ID

实现全局唯一ID我们希望满足以下条件：

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

实现全局唯一ID，我们使用时间戳+序列号，序列号我们取Redis自增长的自增长，但是他的key我们采用业务名称+时间的方式，防止超过范围

  1. /**
  2.  * @author ghp
  3.  * @title 分布式ID生成器
  4.  * @description
  5.  */
  6. @Component
  7. public class RedisIdWorker {
  8.  
  9.     @Resource
  10.     private StringRedisTemplate stringRedisTemplate;
  11.     /**
  12.      * 开始时间戳
  13.      */
  14.     private static final long BEGIN_TIMESTAMP = 1640995200;
  15.     /**
  16.      * 序列化位数
  17.      */
  18.     private static final int COUNT_BITS = 32;
  19.  
  20.     /**
  21.      * 生成分布式ID
  22.      * @param keyPrefix
  23.      * @return
  24.      */
  25.     public long nextId(String keyPrefix){
  26.         // 1、生成时间戳
  27.         LocalDateTime now = LocalDateTime.now();
  28.         long nowSecond = now.toEpochSecond(ZoneOffset.UTC);
  29.         long timestamp = nowSecond - BEGIN_TIMESTAMP;
  30.         // 2、生成序列号
  31.         // 以当天的时间戳为key，防止一直自增下去导致超时，这样每天的极限都是 2^{31}
  32.         String date = now.format(DateTimeFormatter.ofPattern("yyyyMMdd"));
  33.         Long count = stringRedisTemplate.opsForValue().increment(ID_PREFIX + keyPrefix + ":" + date);
  34.         // 3、拼接并返回，时间戳向左移动32位
  35.         return timestamp << COUNT_BITS | count;
  36.     }
  37.  
  38.     public static void main(String[] args) {
  39.         LocalDateTime time = LocalDateTime.of(2022, 1, 1, 0, 0, 0);
  40.         long second = time.toEpochSecond(ZoneOffset.UTC);
  41.         System.out.println("second = " + second);
  42.     }
  43. }

 库存超卖问题

乐观锁

通过判断是否和之前的版本是否一致的乐观锁方法，成功率太低，我们需要对其进行改进，改为库存大于0的情况

set stock = stock - 1 where id = 10 and stock > 1

 一人一单问题

业务上，我们要加上一步，判断是否已经购买过优惠卷

 这样简单的逻辑判断还不行，一个用户可能忽然有多个线程同时进行访问，仍然可能会出现一个用户购买多张优惠卷的问题，这个时候我们要加上锁，保证一个用户只有一个线程在判断是否已经购买。又因为这里是插入操作，我们使用悲观锁（更新操作我们可以使用乐观锁）

悲观锁

  • 我们使用锁synchronized去控制transactional，而不是将synchronized放在transactional中，这是因为锁结束后事务仍未提交，会有线程认为没有订单而造成问题
  • 这里我们使用userId.toString()每次都会new一个新的String对象，因此，我们需要使用userId.toString().intern()返回字符串池中已有的字符串，没有再加入字符串池
  • 非事务的方法调用事务的方法，我们需要使用代理对象，添加代理对象如下

添加依赖 

  1. <dependency>
  2.     <groupId>org.aspectj</groupId>
  3.     <artifactId>aspectjweaver</artifactId>
  4. </dependency>

 在启动类上暴露代理对象

EnableAspectJAutoProxy(exposeProxy = true)

集群下的线程并发安全问题

 在一个JVM中会有一个锁监视器，保证只有一个线程，但是在集群情况下多个JVM就无法做到锁了，这个时候就会出现问题，这个时候我们就需要考虑分布式锁了

分布式锁

什么是分布式锁

 锁误删操作

 线程1在执行任务阻塞的时候，锁被超时误删；如果线程二拿到了线程，线程1执行完任务后就会删除锁，这个时候线程3又进来了，造成了线程2和线程3冲突了

解决：Redis的值设置为UUID+线程id，删除的时候判断是不是自己的线程

锁释放过程发生阻塞

线程在已经判断是自己的锁之后，在释放锁的过程中发生阻塞，这样，又会释放其他线程的锁 

 解决：这就要求我们将判断锁和释放锁封装在一起，使其具有原子性

在Resource目录下添加Lua脚本

编写Lua脚本

  1. ---
  2. --- Generated by EmmyLua(https://github.com/EmmyLua)
  3. --- Created by ghp.
  4. --- DateTime: 2023/7/13 16:19
  5. --- Description 释放锁（简单锁）
  6. ---
  7. -- 比较缓存中的线程标识与当前线程标识是否一致
  8. if (redis.call('get', KEYS[1]) == ARGV[1]) then
  9.     -- 一致，直接删除
  10.     return redis.call('del', KEYS[1])
  11. end
  12. -- 不一致，返回0
  13. return 0

加载Lua脚本

  1. /**
  2.  * 加载Lua脚本
  3.  */
  4. private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;
  5.  
  6. static {
  7.     UNLOCK_SCRIPT = new DefaultRedisScript<>();
  8.     UNLOCK_SCRIPT.setLocation(new ClassPathResource("lua/unlock.lua"));
  9.     UNLOCK_SCRIPT.setResultType(Long.class);
  10. }

完整分布式锁如下（我们通过StringRedisTemplate和name初始化锁，就可以获取锁和释放锁）：

  1. public class SimpleRedisLock implements Lock {
  2.  
  3.     /**
  4.      * RedisTemplate
  5.      */
  6.     private StringRedisTemplate stringRedisTemplate;
  7.  
  8.     /**
  9.      * 锁的名称
  10.      */
  11.     private String name;
  12.     /**
  13.      * key前缀
  14.      */
  15.     private static final String KEY_PREFIX = "lock:";
  16.     /**
  17.      * ID前缀
  18.      */
  19.     private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";
  20.     
  21.     /**
  22.     * 创建锁对象
  23.     */
  24.     public SimpleRedisLock(StringRedisTemplate stringRedisTemplate, String name) {
  25.         this.stringRedisTemplate = stringRedisTemplate;
  26.         this.name = name;
  27.     }
  28.  
  29.  
  30.     /**
  31.      * 获取锁
  32.      *
  33.      * @param timeoutSec 超时时间
  34.      * @return
  35.      */
  36.     @Override
  37.     public boolean tryLock(long timeoutSec) {
  38.         String threadId = ID_PREFIX + Thread.currentThread().getId() + "";
  39.         // SET lock:name id EX timeoutSec NX
  40.         Boolean result = stringRedisTemplate.opsForValue()
  41.                 .setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);
  42.         return Boolean.TRUE.equals(result);
  43.     }
  44.  
  45.     /**
  46.      * 加载Lua脚本
  47.      */
  48.     private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;
  49.  
  50.     static {
  51.         UNLOCK_SCRIPT = new DefaultRedisScript<>();
  52.         UNLOCK_SCRIPT.setLocation(new ClassPathResource("lua/unlock.lua"));
  53.         UNLOCK_SCRIPT.setResultType(Long.class);
  54.     }
  55.  
  56.     /**
  57.      * 释放锁
  58.      */
  59.     @Override
  60.     public void unlock() {
  61.         // 执行lua脚本
  62.         stringRedisTemplate.execute(
  63.                 UNLOCK_SCRIPT,
  64.                 Collections.singletonList(KEY_PREFIX + name),
  65.                 ID_PREFIX + Thread.currentThread().getId()
  66.         );
  67.     }
  68. }

基于setnx所存在的问题

 解决不可重入问题的办法很简单：

在value中增加一个字段count，当同一个线程重新获取锁的时候增加count；同一个线程释放锁的时候减count，count等于0的时候在Redis中释放锁

 基于Redission实现的分布式锁

添加依赖

  1. <dependency>
  2.         <groupId>org.redisson</groupId>
  3.         <artifactId>redisson-spring-boot-starter</artifactId>
  4.         <version>3.11.5</version>
  5. </dependency>

主要步骤：1.创建锁 2.获取锁 3.释放锁

基于分布式锁创建订单如下

  1. /**
  2.  * 创建订单
  3.  *
  4.  * @param voucherOrder
  5.  */
  6. private void handleVoucherOrder(VoucherOrder voucherOrder) {
  7.     Long userId = voucherOrder.getUserId();
  8.     //1.声明一个锁
  9.     RLock lock = redissonClient.getLock(RedisConstants.LOCK_ORDER_KEY + userId);
  10.     //2.获取锁
  11.     //参数1：最大等待时间（解决了不可重试的问题），锁自动释放时间，单位
  12.     boolean isLock = lock.tryLock(1, 10, TimeUnit.SECONDS);
  13.     if (!isLock) {
  14.         // 索取锁失败，重试或者直接抛异常（这个业务是一人一单，所以直接返回失败信息）
  15.         log.error("一人只能下一单");
  16.         return;
  17.     }
  18.     try {
  19.         // 创建订单（使用代理对象调用，是为了确保事务生效）
  20.         proxy.createVoucherOrder(voucherOrder);
  21.     } finally {
  22.         //3.释放锁
  23.         lock.unlock();
  24.     }
  25. }

具体创建订单过程仍然使用事务控制

  1. /**
  2. * 创建订单
  3. *
  4. * @param voucherOrder
  5. * @return
  6. */
  7. @Transactional
  8. @Override
  9. public void createVoucherOrder(VoucherOrder voucherOrder) {
  10. Long userId = voucherOrder.getUserId();
  11. Long voucherId = voucherOrder.getVoucherId();
  12. // 1、判断当前用户是否是第一单
  13. int count = this.count(new LambdaQueryWrapper<VoucherOrder>()
  14.         .eq(VoucherOrder::getUserId, userId));
  15. if (count >= 1) {
  16.     // 当前用户不是第一单
  17.     log.error("当前用户不是第一单");
  18.     return;
  19. }
  20. // 2、用户是第一单，可以下单，秒杀券库存数量减一
  21. boolean flag = seckillVoucherService.update(new LambdaUpdateWrapper<SeckillVoucher>()
  22.         .eq(SeckillVoucher::getVoucherId, voucherId)
  23.         .gt(SeckillVoucher::getStock, 0)
  24.         .setSql("stock = stock -1"));
  25. if (!flag) {
  26.     throw new RuntimeException("秒杀券扣减失败");
  27. }
  28. // 3、将订单保存到数据库
  29. flag = this.save(voucherOrder);
  30. if (!flag) {
  31.     throw new RuntimeException("创建秒杀券订单失败");
  32. }
  33. }

基于Redis的秒杀优化

之前，我们我们资格判断和创建订单操作通过一条线程来完成，多次和数据库发生交换，这就导致我们任务的处理时间过程较长，这里我们主线程从Redis中获取信息判断资格，后面Mysql中数据用操作让另一条线程来完成（相当于服务员在外面帮客户点单，厨师在后面炒菜）。不过这里资格判断仍需要通过Lua脚本完成原子性操作。 

在VoucherServiceImp中实现添加优惠卷信息

  1. /**
  2. * 新增秒杀券
  3. * @param voucher
  4. */
  5. @Override
  6. @Transactional
  7. public void addSeckillVoucher(Voucher voucher) {
  8. // 1、保存优惠券
  9. boolean result = save(voucher);
  10. if (!result){
  11.     throw new RuntimeException("优惠券保存失败");
  12. }
  13. // 2、保存秒杀优惠券信息
  14. SeckillVoucher seckillVoucher = new SeckillVoucher();
  15. seckillVoucher.setVoucherId(voucher.getId());
  16. seckillVoucher.setStock(voucher.getStock());
  17. seckillVoucher.setBeginTime(voucher.getBeginTime());
  18. seckillVoucher.setEndTime(voucher.getEndTime());
  19. result = seckillVoucherService.save(seckillVoucher);
  20. if (!result){
  21.     throw new RuntimeException("秒杀优惠券保存失败");
  22. }
  23. // 3、将秒杀券库存保存到Redis中（这给信息是可以持久化保存到Redis中，不想要的时候就进行手动删除）
  24. stringRedisTemplate.opsForValue().set(SECKILL_STOCK_KEY + voucher.getId(), voucher.getStock().toString());
  25. }

编写Lua脚本stream-seckill.lua

  1. ---
  2. --- Generated by EmmyLua(https://github.com/EmmyLua)
  3. --- Created by ghp.
  4. --- DateTime: 2023/7/16 18:22
  5. --- Description 判断库存是否充足 && 判断用户是否已下单 （以stream作为消息队列）
  6. ---
  7. -- 优惠券id
  8. local voucherId = ARGV[1];
  9. -- 用户id
  10. local userId = ARGV[2];
  11. -- 订单id
  12. local orderId = ARGV[3]
  13.  
  14. -- 库存的key
  15. local stockKey = 'seckill:stock:' .. voucherId;
  16. -- 订单key
  17. local orderKey = 'seckill:order:' .. voucherId;
  18.  
  19. -- 判断库存是否充足 get stockKey > 0 ?
  20. local stock = redis.call('GET', stockKey);
  21. if (tonumber(stock) <= 0) then
  22.     -- 库存不足
  23.     return 1;
  24. end
  25.  
  26. -- 库存充足，判断用户是否已经下过单 SISMEMBER orderKey userId
  27. if (redis.call('SISMEMBER', orderKey, userId) == 1) then
  28.     -- 用户下过单
  29.     return 2;
  30. end
  31.  
  32. -- 库存充足，没有下过单，扣库存、下单
  33. redis.call('INCRBY', stockKey, -1);
  34. redis.call('SADD', orderKey, userId);
  35. -- 发送消息到队列中，XDD stream.orders * key1 value1 key2 value2...
  36. redis.call('XADD', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId);
  37. -- 返回0，表示下单成功
  38. return 0;

加载Lua脚本

  1. /**
  2.  * 加载 判断秒杀券库存是否充足 并且 判断用户是否已下单 的Lua脚本
  3.  */
  4. private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
  5.  
  6. static {
  7.     SECKILL_SCRIPT = new DefaultRedisScript<>();
  8.     SECKILL_SCRIPT.setLocation(new ClassPathResource("lua/stream-seckill.lua"));
  9.     SECKILL_SCRIPT.setResultType(Long.class);
  10. }

异步处理订单问题

基于JVM的异步处理有一个很大的问题，那就是当线程数量变多的时候会出现内存溢出的情况，所以这里，我们使用Redis的消息队列完成异步处理。（基于Redis的单消费者模式最大问题是，当消息数量较多的时候会出现漏读消息的问题）

Stream的消费者组模式有以下的优点

消息可回溯是指在不同消费组都可以收到消息，但是同一个消费者组中都是竞争关系，在同一个消费组里只会被消费一次。总结：独立与JVM以外，消息持久化，消息的确认机制

在Redis中添加队列和消费者组，stream.orders队列，g1消费者组

XGROUP CREATE stream.orders g1 0 MKSTREAM

编写Lua脚本，我们在判断购买资格的时候直接将UserId,VoucherID,OrderId写入消费者组中，并将Lua脚本加载，这一步和上面是一样的，只不过我们增加将消息加入消费组这一步。

秒杀业务完整代码

第一步：seckillVoucher(Long voucherId)在Redis中完成对用户的秒杀，并将信息加入到Redis的队列中去。

第二步：VoucherOrderHandler（）异步去队列接受消息，handlePendingList()处理对异常信息的处理，正常接收消息调用handleVoucherOrder（）完成我们的订单任务，其会获取锁后调用代理对象通过createVoucherOrder（）正真完成我们订单的创建。

基于分布式锁和异步处理订单，最终完成我们的秒杀业务，完整代码如下：

  1. package com.hmdp.service.impl;
  2.  
  3. import cn.hutool.core.bean.BeanUtil;
  4. import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
  5. import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
  6. import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
  7. import com.hmdp.constants.RedisConstants;
  8. import com.hmdp.mapper.VoucherOrderMapper;
  9. import com.hmdp.model.dto.Result;
  10. import com.hmdp.model.entity.SeckillVoucher;
  11. import com.hmdp.model.entity.VoucherOrder;
  12. import com.hmdp.service.ISeckillVoucherService;
  13. import com.hmdp.service.IVoucherOrderService;
  14. import com.hmdp.utils.RedisIdWorker;
  15. import com.hmdp.utils.ThreadLocalUtls;
  16. import org.redisson.api.RLock;
  17. import org.redisson.api.RedissonClient;
  18. import org.springframework.aop.framework.AopContext;
  19. import org.springframework.core.io.ClassPathResource;
  20. import org.springframework.data.redis.connection.RedisConnection;
  21. import org.springframework.data.redis.connection.stream.*;
  22. import org.springframework.data.redis.core.StringRedisTemplate;
  23. import org.springframework.data.redis.core.script.DefaultRedisScript;
  24. import org.springframework.data.redis.core.script.RedisScript;
  25. import org.springframework.stereotype.Service;
  26. import org.springframework.transaction.annotation.Transactional;
  27.  
  28. import javax.annotation.PostConstruct;
  29. import javax.annotation.Resource;
  30. import java.time.Duration;
  31. import java.util.*;
  32. import java.util.concurrent.ExecutorService;
  33. import java.util.concurrent.Executors;
  34. import java.util.concurrent.TimeUnit;
  35.  
  36. import static com.hmdp.constants.RedisConstants.*;
  37.  
  38. /**
  39.  * <p>
  40.  * 服务实现类
  41.  * </p>
  42.  *
  43.  * @author ghp
  44.  * @since 2021-12-22
  45.  */
  46. @Service
  47. public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
  48.  
  49.     @Resource
  50.     private ISeckillVoucherService seckillVoucherService;
  51.  
  52.     @Resource
  53.     private RedisIdWorker redisIdWorker;
  54.  
  55.     @Resource
  56.     private StringRedisTemplate stringRedisTemplate;
  57.  
  58.     @Resource
  59.     private RedissonClient redissonClient;
  60.  
  61.     /**
  62.      * VoucherOrderServiceImpl类的代理对象
  63.      * 将代理对象的作用域进行提升，方面子线程取用
  64.      */
  65.     private IVoucherOrderService proxy;
  66.  
  67.  
  68.  
  69.     /**
  70.      * 加载 判断秒杀券库存是否充足 并且 判断用户是否已下单 的Lua脚本
  71.      */
  72.     private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
  73.  
  74.     static {
  75.         SECKILL_SCRIPT = new DefaultRedisScript<>();
  76.         SECKILL_SCRIPT.setLocation(new ClassPathResource("lua/stream-seckill.lua"));
  77.         SECKILL_SCRIPT.setResultType(Long.class);
  78.     }
  79.  
  80.     //===========================执行秒杀===========================================
  81.     /**
  82.      * 抢购秒杀券，实现对Redis中对资格判断，购买优惠卷的功能，并发送消息，通知去完成对数据库的操作
  83.      *
  84.      * @param voucherId
  85.      * @return
  86.      */
  87.     @Transactional
  88.     @Override
  89.     public Result seckillVoucher(Long voucherId) {
  90.         Long userId = ThreadLocalUtls.getUser().getId();
  91.         long orderId = redisIdWorker.nextId(SECKILL_VOUCHER_ORDER);
  92.  
  93.         // 1、执行Lua脚本，判断用户是否具有秒杀资格
  94.         Long result = null;
  95.         try {
  96.             result = stringRedisTemplate.execute(
  97.                     SECKILL_SCRIPT,
  98.                     Collections.emptyList(),
  99.                     voucherId.toString(),
  100.                     userId.toString(),
  101.                     String.valueOf(orderId)
  102.             );
  103.         } catch (Exception e) {
  104.             log.error("Lua脚本执行失败");
  105.             throw new RuntimeException(e);
  106.         }
  107.         if (result != null && !result.equals(0L)) {
  108.             // result为1表示库存不足，result为2表示用户已下单
  109.             int r = result.intValue();
  110.             return Result.fail(r == 2 ? "不能重复下单" : "库存不足");
  111.         }
  112.  
  113.         // 2、result为0，下单成功，直接返回ok
  114.         // 索取锁成功，创建代理对象，使用代理对象调用第三方事务方法， 防止事务失效
  115.         IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
  116.         this.proxy = proxy;
  117.         return Result.ok();
  118.     }
  119.  
  120.  
  121.     //===================创建线程完成在Mysql中订单的创建=======================
  122.     /**
  123.      * 当前类初始化完毕就立马执行该方法在Redis中创建我们的队列
  124.      */
  125.     @PostConstruct
  126.     private void init() {
  127.         // 创建消息队列
  128.         DefaultRedisScript<Long> mqScript = new DefaultRedisScript<>();
  129.         mqScript.setLocation(new ClassPathResource("lua/stream-mq.lua"));
  130.         mqScript.setResultType(Long.class);
  131.         Long result = null;
  132.         try {
  133.             result = stringRedisTemplate.execute(mqScript,
  134.                     Collections.emptyList(),
  135.                     QUEUE_NAME,
  136.                     GROUP_NAME);
  137.         } catch (Exception e) {
  138.             log.error("队列创建失败", e);
  139.             return;
  140.         }
  141.         int r = result.intValue();
  142.         String info = r == 1 ? "队列创建成功" : "队列已存在";
  143.         log.debug(info);
  144.         // 执行线程任务
  145.         SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
  146.     }
  147.  
  148.     /**
  149.      * 线程池
  150.      */
  151.     private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();
  152.  
  153.     //============================从阻塞队列中获取订单信息进行处理=====================================
  154.     /**
  155.      * 线程任务: 不断从阻塞队列中获取订单信息，使用handleVoucherOrder(voucherOrder)处理订单消息
  156.      *          当出现异常的时候，我们使用handlePendingList()方法从pendinglist中获取消息进行处理
  157.      */
  158.     private class VoucherOrderHandler implements Runnable {
  159.         @Override
  160.         public void run() {
  161.             while (true) {
  162.                 try {
  163.                     // 1、从消息队列中获取订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 1000 STREAMS stream.orders >
  164.                     List<MapRecord<String, Object, Object>> messageList = stringRedisTemplate.opsForStream().read(
  165.                             Consumer.from("g1","c1"),//g1消费组的名称，c1消费者名称
  166.                             //Consumer.from(GROUP_NAME, "c1"),
  167.                             StreamReadOptions.empty().count(1).block(Duration.ofSeconds(1)),//读取一个，阻塞1秒钟
  168.                             StreamOffset.create(QUEUE_NAME, ReadOffset.lastConsumed())//读取队列stream.orders，ReadOffset.lastConsumed()读取最新未读取的队列
  169.                     );
  170.                     // 2、判断消息获取是否成功
  171.                     if (messageList == null || messageList.isEmpty()) {
  172.                         // 2.1 消息获取失败，说明没有消息，进入下一次循环获取消息
  173.                         continue;
  174.                     }
  175.                     // 3、消息获取成功，可以下单
  176.                     // 将消息转成VoucherOrder对象
  177.                     MapRecord<String, Object, Object> record = messageList.get(0);//获取的是消息id,字段名称，字段值
  178.                     Map<Object, Object> messageMap = record.getValue();//获取字段名称和字段值
  179.                     //将结果转为voucher对象，注意：之前我们在Lua脚本中定义的字段名称和voucher字段名称刚好对应
  180.                     VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(messageMap, new VoucherOrder(), true);
  181.  
  182.                     handleVoucherOrder(voucherOrder);
  183.                     // 4、ACK确认 SACK stream.orders g1 id（队列名称，消费组，消息id）
  184.                     stringRedisTemplate.opsForStream().acknowledge(QUEUE_NAME, GROUP_NAME, record.getId());
  185.                 } catch (Exception e) {
  186.                     log.error("处理订单异常", e);
  187.                     //如果处理异常，我们就要尝试去PendingList中将消息取出来进行处理
  188.                     // 处理异常消息
  189.                     handlePendingList();
  190.                 }
  191.             }
  192.         }
  193.     }
  194.  
  195.  
  196.  
  197.  
  198.  
  199.     //==========================使用 handlePendingList对未处理成功的信息进行处理=================================
  200.     private void handlePendingList() {
  201.         while (true) {
  202.             try {
  203.                 // 1、从pendingList中获取订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 1000 STREAMS streams.order 0
  204.                 List<MapRecord<String, Object, Object>> messageList = stringRedisTemplate.opsForStream().read(
  205.                         Consumer.from(GROUP_NAME, "c1"),
  206.                         StreamReadOptions.empty().count(1).block(Duration.ofSeconds(1)),
  207.                         StreamOffset.create(QUEUE_NAME, ReadOffset.from("0"))
  208.                 );
  209.                 // 2、判断pendingList中是否有效性
  210.                 if (messageList == null || messageList.isEmpty()) {
  211.                     // 2.1 pendingList中没有消息，直接结束循环
  212.                     break;
  213.                 }
  214.                 // 3、pendingList中有消息
  215.                 // 将消息转成VoucherOrder对象
  216.                 MapRecord<String, Object, Object> record = messageList.get(0);
  217.                 Map<Object, Object> messageMap = record.getValue();
  218.                 VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(messageMap, new VoucherOrder(), true);
  219.                 handleVoucherOrder(voucherOrder);
  220.                 // 4、ACK确认 SACK stream.orders g1 id
  221.                 stringRedisTemplate.opsForStream().acknowledge(QUEUE_NAME, GROUP_NAME, record.getId());
  222.             } catch (Exception e) {
  223.                 log.error("处理订单异常", e);
  224.                 // 这里不用调自己，直接就进入下一次循环，再从pendingList中取，这里只需要休眠一下，防止获取消息太频繁
  225.                 try {
  226.                     Thread.sleep(20);
  227.                 } catch (InterruptedException ex) {
  228.                     log.error("线程休眠异常", ex);
  229.                 }
  230.             }
  231.         }
  232.     }
  233.  
  234.  
  235.  
  236.     //========================获取锁并调用createVoucherOrder()完成对订单的创建===========================
  237. /**
  238.  * 创建订单
  239.  *
  240.  * @param voucherOrder
  241.  */
  242. private void handleVoucherOrder(VoucherOrder voucherOrder) {
  243.     Long userId = voucherOrder.getUserId();
  244.     //1.声明一个锁
  245.     RLock lock = redissonClient.getLock(RedisConstants.LOCK_ORDER_KEY + userId);
  246.     //2.获取锁
  247.     //参数1：最大等待时间（解决了不可重试的问题），锁自动释放时间，单位
  248.     boolean isLock = lock.tryLock();
  249.     if (!isLock) {
  250.         // 索取锁失败，重试或者直接抛异常（这个业务是一人一单，所以直接返回失败信息）
  251.         log.error("一人只能下一单");
  252.         return;
  253.     }
  254.     try {
  255.         // 创建订单（使用代理对象调用，是为了确保事务生效）
  256.         proxy.createVoucherOrder(voucherOrder);
  257.     } finally {
  258.         //3.释放锁
  259.         lock.unlock();
  260.     }
  261. }
  262.  
  263.     /**
  264.      * 创建订单
  265.      *
  266.      * @param voucherOrder
  267.      * @return
  268.      */
  269.     @Transactional
  270.     @Override
  271.     public void createVoucherOrder(VoucherOrder voucherOrder) {
  272.         Long userId = voucherOrder.getUserId();
  273.         Long voucherId = voucherOrder.getVoucherId();
  274.         // 1、判断当前用户是否是第一单
  275.         int count = this.count(new LambdaQueryWrapper<VoucherOrder>()
  276.                 .eq(VoucherOrder::getUserId, userId));
  277.         if (count >= 1) {
  278.             // 当前用户不是第一单
  279.             log.error("当前用户不是第一单");
  280.             return;
  281.         }
  282.         // 2、用户是第一单，可以下单，秒杀券库存数量减一
  283.         boolean flag = seckillVoucherService.update(new LambdaUpdateWrapper<SeckillVoucher>()
  284.                 .eq(SeckillVoucher::getVoucherId, voucherId)
  285.                 .gt(SeckillVoucher::getStock, 0)
  286.                 .setSql("stock = stock -1"));
  287.         if (!flag) {
  288.             throw new RuntimeException("秒杀券扣减失败");
  289.         }
  290.         // 3、将订单保存到数据库
  291.         flag = this.save(voucherOrder);
  292.         if (!flag) {
  293.             throw new RuntimeException("创建秒杀券订单失败");
  294.         }
  295.     }
  296. }

达人探店

@TableField表示该字段是后面添加进去的

 点赞

为了同时满足判断元素是否在集合内，以及元素的排序，我们在Redis中使用scored_set类型

  1. ZADD z1 1 m1 2 m2 3 m3 //m1 m2 m3 元素的值分别为1 2 3
  2. ZSCORE z1 m1 // 得到结果1
  3. ZRANGE z1 0 4 //查前5名

分析：1.我们使用scored_set类型，按用户的点赞时间作为分数进行排序 2.用户每次点赞我们在数据库设置blog的liked加1 3.在blog实体类中增加字段isLiked表示当前用户是否点赞了博客

判断当前blog是否被点赞，如果被点赞在blog实体类标识被点赞

  1. /**
  2.  * 判断当前用户是否点赞该博客
  3.  */
  4. private void isBlogLiked(Blog blog) {
  5.     UserDTO user = ThreadLocalUtls.getUser();
  6.     if (Objects.isNull(user)) {
  7.         // 当前用户未登录，无需查询点赞
  8.         return;
  9.     }
  10.     Long userId = user.getId();
  11.     String key = BLOG_LIKED_KEY + blog.getId();
  12.     Double score = stringRedisTemplate.opsForZSet().score(key, userId.toString());
  13.     //在blog中标识被当前用户已经点赞
  14.     blog.setIsLike(Objects.nonNull(score));
  15. }

实现点赞功能，用户没点赞过则点赞，用户点赞了则取消点赞

  1. /**
  2.  * 点赞
  3.  *
  4.  * @param id
  5.  * @return
  6.  */
  7. @Override
  8. public Result likeBlog(Long id) {
  9.     // 1、判断用户是否点赞
  10.     Long userId = ThreadLocalUtls.getUser().getId();
  11.     String key = BLOG_LIKED_KEY + id;
  12.     // 在score_set中查用户的分数，如果查到了说明存在
  13.     Double score = stringRedisTemplate.opsForZSet().score(key, userId.toString());
  14.     boolean result;
  15.     if (score == null) {
  16.         // 1.1 用户未点赞，点赞数+1
  17.         result = this.update(new LambdaUpdateWrapper<Blog>()
  18.                 .eq(Blog::getId, id)
  19.                 .setSql("liked = liked + 1"));
  20.         if (result) {
  21.             // 数据库更新成功，更新缓存 zadd key value score，分数我们按时间戳去存
  22.             stringRedisTemplate.opsForZSet().add(key, userId.toString(), System.currentTimeMillis());
  23.         }
  24.     } else {
  25.         // 1.2 用户已点赞，点赞数-1
  26.         result = this.update(new LambdaUpdateWrapper<Blog>()
  27.                 .eq(Blog::getId, id)
  28.                 .setSql("liked = liked - 1"));
  29.         if (result) {
  30.             // 数据更新成功，更新缓存 zrem key value
  31.             stringRedisTemplate.opsForZSet().remove(key, userId.toString());
  32.         }
  33.     }
  34.     return Result.ok();
  35. }

查询点赞前5的用户，我们先从Redis中查询前5的用户id，再用id从Mysql中查询，注意id的顺序必须是降序

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

好友关注

关注功能实现

用户和博主属于多对多的关系，因此我们建立一张多对多的关系表

关注和取消关注，我们同时在Redis和关系表中取消增加用户信息

  1. /**
  2. * 关注用户
  3. *
  4. * @param followUserId 关注用户的id
  5. * @param isFollow     是否已关注
  6. * @return
  7. */
  8. @Override
  9. public Result follow(Long followUserId, Boolean isFollow) {
  10. Long userId = ThreadLocalUtls.getUser().getId();
  11. String key = FOLLOW_KEY + userId;
  12. if (isFollow) {
  13.     // 用户未关注，则关注
  14.     Follow follow = new Follow();
  15.     follow.setUserId(userId);
  16.     follow.setFollowUserId(followUserId);
  17.     boolean isSuccess = this.save(follow);
  18.     if (isSuccess) {
  19.         // 用户关注信息保存成功，把关注的用户id放入Redis的Set集合中，
  20.         stringRedisTemplate.opsForSet().add(key, followUserId.toString());
  21.     }
  22. } else {
  23.     // 用户已关注，删除关注信息
  24.     boolean isSuccess = this.remove(new LambdaQueryWrapper<Follow>()
  25.             .eq(Follow::getUserId, userId)
  26.             .eq(Follow::getFollowUserId, followUserId));
  27.     if (isSuccess) {
  28.         stringRedisTemplate.opsForSet().remove(key, followUserId.toString());
  29.     }
  30. }
  31. return Result.ok();
  32. }

直接从数据中判断用户是否关注博主

  1. /**
  2.  * 是否关注用户
  3.  *
  4.  * @param followUserId 关注用户的id
  5.  * @return
  6.  */
  7. @Override
  8. public Result isFollow(Long followUserId) {
  9.     Long userId = ThreadLocalUtls.getUser().getId();
  10.     int count = this.count(new LambdaQueryWrapper<Follow>()
  11.             .eq(Follow::getUserId, userId)
  12.             .eq(Follow::getFollowUserId, followUserId));
  13.     return Result.ok(count > 0);
  14. }

共同关注

利用Redis中SINTER实现共同关注

  1. SADD s1 m1 m2
  2. SADD s2 m2 m3
  3. SINTER s1 s2 //返回m2

  1. /**
  2.  * 查询共同关注
  3.  *
  4.  * @param id
  5.  * @return
  6.  */
  7. @Override
  8. public Result followCommons(Long id) {
  9.     Long userId = ThreadLocalUtls.getUser().getId();
  10.     String key1 = FOLLOW_KEY + userId;
  11.     String key2 = FOLLOW_KEY + id;
  12.     // 查询当前用户与目标用户的共同关注对象
  13.     Set<String> intersect = stringRedisTemplate.opsForSet().intersect(key1, key2);
  14.     if (Objects.isNull(intersect) || intersect.isEmpty()) {
  15.         return Result.ok(Collections.emptyList());
  16.     }
  17.     List<Long> ids = intersect.stream().map(Long::valueOf).collect(Collectors.toList());
  18.     // 查询共同关注的用户信息
  19.     List<UserDTO> userDTOList = userService.listByIds(ids).stream()
  20.             .map(user -> BeanUtil.copyProperties(user, UserDTO.class))
  21.             .collect(Collectors.toList());
  22.     return Result.ok(userDTOList);
  23. }

Feed流实现方案

大V对普通粉丝使用拉的方式先发到发件箱里，他们需要的时候再将消息推送给普通用户，对于活跃用户使用推的方式，直接将消息推送到他们的收件箱；对于普通人，直接使用推的方式，直接将消息发送到他们的收件箱 

 使用sorted_set和list都可以实现上面的需求但是，但是这里我们需要按滚动分页查询，即使有用户发送了新的消息，我们都从上一次分页的最后一条开始查询，所以这里我们直接采用sorted_set实现该功能

  1. /**
  2.  * 保存探店笔记
  3.  *
  4.  * @param blog
  5.  * @return
  6.  */
  7. @Override
  8. public Result saveBlog(Blog blog) {
  9.     Long userId = ThreadLocalUtls.getUser().getId();
  10.     blog.setUserId(userId);
  11.     // 保存探店笔记
  12.     boolean isSuccess = this.save(blog);
  13.     if (!isSuccess) {
  14.         return Result.fail("笔记保存失败");
  15.     }
  16.     // 查询笔记作者的所有粉丝，找到FollowId与等于作者UserId的所有用户就可以了
  17.     List<Follow> follows = followService.list(new LambdaQueryWrapper<Follow>()
  18.             .eq(Follow::getFollowUserId, userId));
  19.     // 将笔记推送给所有的粉丝
  20.     for (Follow follow : follows) {
  21.         // 获取粉丝的id
  22.         Long id = follow.getUserId();
  23.         // 推送笔记
  24.         String key = FEED_KEY + id;
  25.         //将blogId推送到每一个粉丝的收件箱，粉丝收件箱的key=FEED_KEY + id
  26.         stringRedisTemplate.opsForZSet().add(key, blog.getId().toString(), System.currentTimeMillis());
  27.     }
  28.     return Result.ok(blog.getId());
  29. }

滚动分页查询的实现

参数：5~0表示范围，1表示小于等于5的偏移量，3表示输出3个

这样输出有一个问题如果sorted_set末尾有两个相同的数，偏移量仍是1的话就会出现问题

可以看到offset仍是1的话，6会重复出现，所以offset应该是上一次查询最后一个数的个数

  1. /**
  2.  * 关注推送页面的分页查询
  3.  */
  4. @GetMapping("/of/follow")
  5. public Result queryBlogOfFollow(@RequestParam("lastId") Long max,
  6.                                 @RequestParam(value = "offset", defaultValue = "0") Integer offset) {
  7.     return blogService.queryBlogOfFollow(max, offset);
  8. }

 注意：这里使用blogId查询blog，我们需要使用自己的顺序

  1. /**
  2.  * 关注推送页面的笔记分页
  3.  *
  4.  * @param max
  5.  * @param offset
  6.  * @return
  7.  */
  8. @Override
  9. public Result queryBlogOfFollow(Long max, Integer offset) {
  10.     // 1、查询收件箱
  11.     Long userId = ThreadLocalUtls.getUser().getId();
  12.     String key = FEED_KEY + userId;
  13.     // ZREVRANGEBYSCORE key Max Min LIMIT offset count
  14.     Set<ZSetOperations.TypedTuple<String>> typedTuples = stringRedisTemplate.opsForZSet()
  15.             .reverseRangeByScoreWithScores(key, 0, max, offset, 2);
  16.     // 2、判断收件箱中是否有数据
  17.     if (typedTuples == null || typedTuples.isEmpty()) {
  18.         return Result.ok();
  19.     }
  20.  
  21.     // 3、收件箱中有数据，则解析数据: blogId、minTime（时间戳）、offset
  22.     List<Long> ids = new ArrayList<>(typedTuples.size());
  23.     long minTime = 0; // 记录当前最小值
  24.     int os = 1; // 偏移量offset，用来计数
  25.     for (ZSetOperations.TypedTuple<String> tuple : typedTuples) { // 5 4 4 2 2
  26.         // 获取id
  27.         ids.add(Long.valueOf(tuple.getValue()));
  28.         // 获取分数（时间戳）
  29.         long time = tuple.getScore().longValue();
  30.         if (time == minTime) {
  31.             // 当前时间等于最小时间，偏移量+1
  32.             os++;
  33.         } else {
  34.             // 当前时间不等于最小时间，重置
  35.             minTime = time;
  36.             os = 1;
  37.         }
  38.     }
  39.  
  40.     // 4、根据id查询blog（使用in查询的数据是默认按照id升序排序的，这里需要使用我们自己指定的顺序排序）
  41.     String idStr = StrUtil.join(",", ids);
  42.     List<Blog> blogs = this.list(new LambdaQueryWrapper<Blog>().in(Blog::getId, ids)
  43.             .last("ORDER BY FIELD(id," + idStr + ")"));
  44.  
  45.     // 设置blog相关的用户数据，是否被点赞等属性值
  46.     for (Blog blog : blogs) {
  47.         // 查询blog有关的用户
  48.         queryUserByBlog(blog);
  49.         // 查询blog是否被点赞
  50.         isBlogLiked(blog);
  51.     }
  52.  
  53.     // 5、封装并返回
  54.     ScrollResult scrollResult = new ScrollResult();
  55.     scrollResult.setList(blogs);
  56.     scrollResult.setOffset(os);
  57.     scrollResult.setMinTime(minTime);
  58.  
  59.     return Result.ok(scrollResult);
  60. }

附件商铺

1.添加数据，并计算距离

 2.搜索附件10KM火车站

 导入店铺数据到GEO

我们按店铺类型将店铺的坐标信息导入Redis中，我们将SHOP_GEO_KEY+店铺类型id作为我们的key，店铺的坐标+id作为我们的value

  1. /**
  2.  * 预热店铺数据，按照typeId进行分组，用于实现附近商户搜索功能
  3.  */
  4. @Test
  5. public void loadShopListToCache() {
  6.     // 1、获取店铺数据
  7.     List<Shop> shopList = shopService.list();
  8.     // 2、根据 typeId 进行分类
  9. //        Map<Long, List<Shop>> shopMap = new HashMap<>();
  10. //        for (Shop shop : shopList) {
  11. //            Long shopId = shop.getId();
  12. //            if (shopMap.containsKey(shopId)){
  13. //                // 已存在，添加到已有的集合中
  14. //                shopMap.get(shopId).add(shop);
  15. //            }else{
  16. //                // 不存在，直接添加
  17. //                shopMap.put(shopId, Arrays.asList(shop));
  18. //            }
  19. //        }
  20.     // 使用 Lambda 表达式，更加优雅（优雅永不过时）
  21.     Map<Long, List<Shop>> shopMap = shopList.stream()
  22.             .collect(Collectors.groupingBy(Shop::getTypeId));
  23.  
  24.     // 3、将分好类的店铺数据写入redis
  25.     for (Map.Entry<Long, List<Shop>> shopMapEntry : shopMap.entrySet()) {
  26.         // 3.1 获取 typeId
  27.         Long typeId = shopMapEntry.getKey();
  28.         List<Shop> values = shopMapEntry.getValue();
  29.         // 3.2 将同类型的店铺的写入同一个GEO ( GEOADD key 经度 维度 member )
  30.         String key = SHOP_GEO_KEY + typeId;
  31.         // 方式一：单个写入(这种方式，一个请求一个请求的发送，十分耗费资源，我们可以进行批量操作)
  32. //            for (Shop shop : values) {
  33. //                stringRedisTemplate.opsForGeo().add(key, new Point(shop.getX(), shop.getY()),
  34. //                shop.getId().toString());
  35. //            }
  36.         // 方式二：批量写入
  37.         List<RedisGeoCommands.GeoLocation<String>> locations = new ArrayList<>();
  38.         for (Shop shop : values) {
  39.            locations.add(new RedisGeoCommands.GeoLocation<>(shop.getId().toString(),
  40.                    new Point(shop.getX(), shop.getY())));
  41.         }
  42.         stringRedisTemplate.opsForGeo().add(key, locations);
  43.     }
  44. }

实现附件商铺功能

 接口

  1. /**
  2.  * 根据商铺类型分页查询商铺信息
  3.  *
  4.  * @param typeId  商铺类型
  5.  * @param current 页码
  6.  * @return 商铺列表
  7.  */
  8. @GetMapping("/of/type")
  9. public Result queryShopByType(
  10.         @RequestParam("typeId") Integer typeId,
  11.         @RequestParam(value = "current", defaultValue = "1") Integer current,
  12.         @RequestParam(value = "x", required = false) Double x,
  13.         @RequestParam(value = "y", required = false) Double y) {
  14.     return shopService.queryShopByType(typeId, current, x, y);
  15. }

实现附件商铺

当没有传入x y的时候，我们直接从数据库中查询数据，当有x y传入的时候我们再从Redis中查询数据，但是Redis只会返回0~end的数据，我们需要手动截取from~end的数据

  1. /**
  2.  * 分页查询店铺数据
  3.  *
  4.  * @param typeId 店铺类型ID
  5.  * @param current 页码
  6.  * @param x 坐标x轴
  7.  * @param y 坐标y轴
  8.  * @return
  9.  */
  10. @Override
  11. public Result queryShopByType(Integer typeId, Integer current, Double x, Double y) {
  12.     // 1、判断是否需要根据坐标查询
  13.     if (x == null || y == null) {
  14.         // 不需要坐标查询，按数据库查询
  15.         Page<Shop> page = this.page(new Page<>(current, SystemConstants.DEFAULT_PAGE_SIZE),
  16.                 new LambdaQueryWrapper<Shop>().eq(Shop::getTypeId, typeId));
  17.         // 返回数据
  18.         return Result.ok(page.getRecords());
  19.     }
  20.  
  21.     // 2、需要查询坐标，则需要到Redis中进行查询
  22.     // 2.1 计算分页参数
  23.     int from = (current - 1) * SystemConstants.DEFAULT_PAGE_SIZE;
  24.     int end = current * SystemConstants.DEFAULT_PAGE_SIZE;
  25.     // 2.2 根据经纬度从redis中查询店铺数据，并按照距离排序、分页
  26.     String key = SHOP_GEO_KEY + typeId;
  27.     // GEOSEARCH key BYLONLAT x y BYRADIUS 10 WITHDISTANCE 结果: shopId、distance
  28.     GeoResults<RedisGeoCommands.GeoLocation<String>> results = stringRedisTemplate
  29.             .opsForGeo().search(
  30.                     key,
  31.                     GeoReference.fromCoordinate(x, y),
  32.                     // 默认搜索范围是5km，默认单位m
  33.                     new Distance(5000),
  34.                     // 查询从0到end，所以后面还需要截取from到end之间的数据
  35.                     RedisGeoCommands.GeoSearchCommandArgs.newGeoSearchArgs().includeDistance().limit(end));
  36.  
  37.     // 4、解析出店铺id
  38.     // 4.1 健壮性判断，防止skip出现NPE
  39.     if (results == null) {
  40.         // 缓存中不存在店铺数据
  41.         return Result.ok(Collections.emptyList());
  42.     }
  43.  
  44.  
  45.  
  46.  
  47.     // 4.2 缓存中存在店铺数据，则需要截取 from ~ end的部分，需要判断from到end之间的数据是否存在
  48.     List<GeoResult<RedisGeoCommands.GeoLocation<String>>> list = results.getContent();
  49.     if (list.size() <= from) {
  50.         // 当前数据比起始索引还要小，说明没有我们要查询页的数据
  51.         return Result.ok(Collections.emptyList());
  52.     }
  53.     // 4.3 from到end之间的数据存在，则解析出店铺id
  54.     List<Long> ids = new ArrayList<>(list.size());
  55.     Map<String, Distance> distanceMap = new HashMap<>(list.size());
  56.     // skip表示直接从第from个数据开始遍历
  57.     list.stream().skip(from).forEach(result -> {
  58.         // 获取店铺id
  59.         String shopIdStr = result.getContent().getName();
  60.         ids.add(Long.valueOf(shopIdStr));
  61.         // 获取店铺距离
  62.         Distance distance = result.getDistance();
  63.         distanceMap.put(shopIdStr, distance);
  64.     });
  65.  
  66.  
  67.  
  68.  
  69.     // 5、根据店铺ids查询出店铺数据
  70.     String idStr = StrUtil.join(",", ids);
  71.     // 5.1 查寻出所有符合条件的店铺数据（这里需要利用ORDER BY FIELD确保id的有序性）
  72.     List<Shop> shopList = query().in("id", ids).last("ORDER BY FIELD(id," + idStr + ")").list();
  73. //        List<Shop> shopList = this.list(new LambdaQueryWrapper<Shop>()
  74. //                .in(Shop::getId, ids)
  75. //                .last("ORDER BY FIELD(id," + idStr + ")"));
  76.     // 5.2 为店铺的距离属性进行赋值
  77.     for (Shop shop : shopList) {
  78.         shop.setDistance(distanceMap.get(shop.getId().toString()).getValue());
  79.     }
  80.  
  81.     // 6、返回
  82.     return Result.ok(shopList);
  83. }

用户签到

使用Redis中的BitMap类型统计签到

 

 BITFIELD bm1 GET u2 0  u表示按无符号 2表示读取2个数 0表示从0位开始读 返回10进制数对应的二进制就是11

BITPOS bm1 0 表示第一次出现0的位置 BITPOS bm1 1 表示第一次出现1的位置

实现签到功能

  1. /**
  2.  * 用户签到
  3.  *
  4.  * @return
  5.  */
  6. @Override
  7. public Result sign() {
  8.     // 获取当前登录用户
  9.     Long userId = ThreadLocalUtls.getUser().getId();
  10.     // 获取日期
  11.     LocalDateTime now = LocalDateTime.now();
  12.     // 拼接key
  13.     String keySuffix = now.format(DateTimeFormatter.ofPattern(":yyyyMM"));
  14.     String key = USER_SIGN_KEY + userId + keySuffix;
  15.     // 获取今天是本月的第几天
  16.     int dayOfMonth = now.getDayOfMonth();
  17.     // 写入Redis SETBIT key offset 1
  18.     stringRedisTemplate.opsForValue().setBit(key, dayOfMonth - 1, true);
  19.     return Result.ok();
  20. }

实现统计连续签到功能

  1. /**
  2.  * 记录连续签到的天数
  3.  *
  4.  * @return
  5.  */
  6. @Override
  7. public Result signCount() {
  8.     // 1、获取签到记录
  9.     // 获取当前登录用户
  10.     Long userId = ThreadLocalUtls.getUser().getId();
  11.     // 获取日期
  12.     LocalDateTime now = LocalDateTime.now();
  13.     // 拼接key
  14.     String keySuffix = now.format(DateTimeFormatter.ofPattern(":yyyyMM"));
  15.     String key = USER_SIGN_KEY + userId + keySuffix;
  16.     // 获取今天是本月的第几天
  17.     int dayOfMonth = now.getDayOfMonth();
  18.     // 获取本月截止今天为止的所有的签到记录，返回的是一个十进制的数字 BITFIELD sign:5:202203 GET u14 0
  19.     List<Long> result = stringRedisTemplate.opsForValue().bitField(
  20.             key,
  21.             BitFieldSubCommands.create()
  22.                     .get(BitFieldSubCommands.BitFieldType.unsigned(dayOfMonth)).valueAt(0)
  23.     );
  24.     // 2、判断签到记录是否存在
  25.     if (result == null || result.isEmpty()) {
  26.         // 没有任何签到结果
  27.         return Result.ok(0);
  28.     }
  29.     // 3、获取本月的签到数（List<Long>是因为BitFieldSubCommands是一个子命令，可能存在多个返回结果，这里我们知识使用了Get，
  30.     // 可以明确只有一个返回结果，即为本月的签到数，所以这里就可以直接通过get(0)来获取）
  31.     Long num = result.get(0);
  32.     if (num == null || num == 0) {
  33.         // 二次判断签到结果是否存在，让代码更加健壮
  34.         return Result.ok(0);
  35.     }
  36.     // 4、循环遍历，获取连续签到的天数（从当前天起始）
  37.     int count = 0;
  38.     while (true) {
  39.         // 让这个数字与1做与运算，得到数字的最后一个bit位，并且判断这个bit位是否为0
  40.         if ((num & 1) == 0) {
  41.             // 如果为0，说明未签到，结束
  42.             break;
  43.         } else {
  44.             // 如果不为0，说明已签到，计数器+1
  45.             count++;
  46.         }
  47.         // 把数字右移一位，抛弃最后一个bit位，继续下一个bit位
  48.         num >>>= 1;
  49.     }
  50.     return Result.ok(count);
  51. }

UV统计

 

使用HLL不可以统计重复的数据，相同的数据只会被当作一次统计

 最后返回结果997593，误差为百分之一

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号