【Redis实战】黑马点评项目

一、短信登录模块

1.发送短信验证码

1.controller请求:

@Slf4j
@RestController
@RequestMapping("/user")
public class UserController {

    @Resource
    private IUserService userService;

    @Resource
    private IUserInfoService userInfoService;

    /**
     * 发送手机验证码
     */
    @PostMapping("/code")
    public Result sendCode(@RequestParam("phone") String phone, HttpSession session) {
        //发送短信验证码并保存验证码
        return userService.sendCode(phone,session);;
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

2.登录业务方法

@Service
public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements IUserService {
   @Override
   public Result sendCode(String phone, HttpSession session) {
       //1.校验手机号
       //2.如果不符合，则返回错误dto
       if (RegexUtils.isPhoneInvalid(phone)) {
           return Result.fail("手机号格式错误！");
       }
       //3.不满足第2点，说明符合，生成验证码
       String code = RandomUtil.randomNumbers(6);
       //4.保存验证码到session
       session.setAttribute("code",code);
       //5.发送验证码:用日志模拟发送验证码
       log.debug("验证码发送成功,验证码:"+code);
       return Result.ok();
   }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

2.登录验证

1.controller登录请求

/**
     * 登录功能
     * @param loginForm 登录参数，包含手机号、验证码；或者手机号、密码
     */
    @PostMapping("/login")
    public Result login(@RequestBody LoginFormDTO loginForm, HttpSession session){
        //实现登录功能
        return userService.login(loginForm,session);
    }

1
2
3
4
5
6
7
8
9
10

2.登录业务逻辑

 @Override
    public Result login(LoginFormDTO loginForm, HttpSession session) {
        //1.校验手机号
        String phone = loginForm.getPhone();
        if (RegexUtils.isPhoneInvalid(phone)) {
            return Result.fail("手机号格式错误！");
        }
        //2.校验验证码
        String code = loginForm.getCode();
        Object cacheCode = session.getAttribute("code");
        //3.如果验证码不一致，返回错误信息
        if (cacheCode == null || !cacheCode.toString().equals(code)) {
            return Result.fail("验证码错误！");
        }
        //4.如果一致，查询数据库有没有这样的用户:select * from user where phone = #{phone} //list or one
        User user = query().eq("phone", phone).one();
        //5.查不到，创建新用户，保存到数据库
        if (user == null) {
            user = createAndSaveUser(phone);//新建的用户
        }
        //6.查得到直接把用户保存到session中，查不到就把新建的用户保存到session中
        session.setAttribute("user", user);
        return Result.ok();
    }

    private User createAndSaveUser(String phone) {
        User user = new User();
        user.setPhone(phone);//用户输入的手机号
        user.setNickName(USER_NICK_NAME_PREFIX + RandomUtil.randomString(8));
        save(user);
        return user;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

3.登录校验和拦截

1.登录拦截

public class LoginInterceptor implements HandlerInterceptor {
    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        //TODO 获取session中的用户信息 如果有就存到ThreadLocal去 没有就拦截
        HttpSession session = request.getSession();
        User user = (User) session.getAttribute("user");
        if (user == null) {
            response.setStatus(401);//拦截，返回401：用户不存在于session
            return false;
        }
        UserHolder.saveUser(user);//存到ThreadLocal去
        return true;
    }
    //视图渲染之后，返回用户之前的时刻需要做移除用户信息操作以防信息泄露。
    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
        UserHolder.removeUser();    
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

怎么存到ThreadLocal呢，答案就在UserHolder里,这个工具类已经提供了对ThreadLocal进行读写操作。

package com.hmdp.utils;

import com.hmdp.entity.User;

public class UserHolder {
    private static final ThreadLocal<User> tl = new ThreadLocal<>();

    public static void saveUser(User user){
        tl.set(user);
    }

    public static User getUser(){
        return tl.get();
    }

    public static void removeUser(){
        tl.remove();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

返回信息给个人中心页面：

1

2.在MVC配置类里配置拦截器

放行发送验证和登录验证的请求：

@Configuration
public class WebMVCConfig implements WebMvcConfigurer {
    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(new LoginInterceptor())
                .excludePathPatterns(
                        "/user/code",
                        "/user/login"
                );
    }
}
1
2
3
4
5
6
7
8
9
10
11

3.获取用户并返回

    @GetMapping("/me")
    public Result me(){
        //从ThreadLocal获取用户信息返回前端
        User user = UserHolder.getUser();
        return Result.ok(user);
    }
1
2
3
4
5
6

4.脱敏处理

UserDTO只封装了常用的且不暴露用户信息的属性：

@Data
public class UserDTO {
    private Long id;
    private String nickName;
    private String icon;
}

1
2
3
4
5
6
7

因此我们在把user保存到session中的时候，不需要保存所有的字段：

 //6.查得到直接把用户保存到session中，查不到就把新建的用户保存到session中
session.setAttribute("user", BeanUtil.copyProperties(user, UserDTO.class));
1
2

但是要注意把UserHolder工具类里的User类全部修改为UserDTO确保后续的登录校验拦截也是保存的是UserDTO而非全部信息

public class UserHolder {
    private static final ThreadLocal<UserDTO> tl = new ThreadLocal<>();

    public static void saveUser(UserDTO user){
        tl.set(user);
    }

    public static UserDTO getUser(){
        return tl.get();
    }

    public static void removeUser(){
        tl.remove();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

返回给me请求的时候，只需要返回UserDTO对象了。

    @GetMapping("/me")
    public Result me(){
        // 获取当前登录的用户并返回
        UserDTO user = UserHolder.getUser();
        return Result.ok(user);
    }
1
2
3
4
5
6

4.集群的session共享的问题

由于客户端访问nginx之后，如果有多台tomcat做并发集群，当nginx作负载均衡，在多个tomcat之间做轮循，每一个tomcat都会有自己的session，如果同一个用户负载均衡进入的tomcat不一样，那么他们的session保存的数据就不一致，所以为了解决session数据同步性，共享性，存储性，我们需要用redis来代替session。

5.Redis代替session实现短信登录模块（必学）

校验手机号，生成验证码，把校验成功的手机号作为key，验证码作为value存入到Redis中去。那么当用户点击登录/注册按钮的时候，进行校验验证码的时候，就可以从redis中去获取value，那么当我们验证码校验成功后，用户存在我们需要把用户存入到redis中，则需要手动的生成一个随机的token作为key，用户的信息作为value存入到redis中。

校验登录状态时，请求并携带token，通过随机token作为key获取到用户数据。用户存在就保存到ThreadLocal中并放行当前请求。拦截用户不存在的请求。

1.短信验证码发送

	@Resource
    StringRedisTemplate stringRedisTemplate;

    @Override
    public Result sendCode(String phone, HttpSession session) {
        //1.校验手机号
        //2.如果不符合，则返回错误信息
        if (RegexUtils.isPhoneInvalid(phone)) {
            return Result.fail("手机号格式错误！");
        }
        //3.生成验证码
        String code = RandomUtil.randomNumbers(6);
        //4.以手机号作为key保存验证码到redis
        stringRedisTemplate.opsForValue().set(LOGIN_CODE_KEY + phone, code, LOGIN_CODE_TTL, TimeUnit.MINUTES);
        //5.发送验证码:用日志模拟发送验证码
        log.debug("验证码发送成功,验证码:" + code);
        return Result.ok();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

1. 在UserServceImpl类里，第一步先注入StringRedisTemplate ，用来对数据基于redis进行操作。
2. 拿到前端传来的手机号，进行校验。
3. 校验通过后，生成验证码。
4. 以手机号作为key，验证码作为value存入到redis中
5. 然后发送验证码。

2.登录验证

    @Override
    public Result login(LoginFormDTO loginForm, HttpSession session) {
        //1.校验手机号
        String phone = loginForm.getPhone();
        if (RegexUtils.isPhoneInvalid(phone)) {
            return Result.fail("手机号格式错误！");
        }
        //2.以手机号作为key读取验证码，再进行校验验证码
        String code = loginForm.getCode();
        String cacheCode = stringRedisTemplate.opsForValue().get(LOGIN_CODE_KEY + phone);
        //3.如果验证码不一致，返回错误信息
        if (cacheCode == null || !cacheCode.equals(code)) {
            return Result.fail("验证码错误！");
        }
        //4.如果一致，查询数据库有没有这样的用户
        User user = query().eq("phone", phone).one();
        //5.判断用户是否存在
        if (user == null) {
            //6.不存在，新建用户
            user = createAndSaveUser(phone);
        }
        //7.把用户信息保存到redis中
        //7.1 随机生成token 作为登录令牌
        String token = UUID.randomUUID().toString(true);//不带下划线的UUID
        String tokenKey = LOGIN_USER_KEY + token;
        //7.2 将user对象转成Hash存储
        UserDTO userDTO = BeanUtil.copyProperties(user, UserDTO.class);
        Map<String, Object> userMap = BeanUtil.beanToMap(userDTO, new HashMap<>(),
                CopyOptions.create().setIgnoreNullValue(true).
                        setFieldValueEditor((fieldName, fieldValue) -> fieldValue.toString()));
        //7.3 存储
        stringRedisTemplate.opsForHash().putAll(tokenKey, userMap);
        //7.4设置有效期
        stringRedisTemplate.expire(tokenKey, LOGIN_USER_TTL, TimeUnit.MINUTES);
        //8.返回token到前端
        return Result.ok(token);
    }

    private User createAndSaveUser(String phone) {
        User user = new User();
        user.setPhone(phone);//用户输入的手机号
        user.setNickName(USER_NICK_NAME_PREFIX + RandomUtil.randomString(8));
        save(user);
        return user;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

1. 校验手机号
2. 通过手机号作为key从redis中取出验证码
3. 把用户输入的验证码和从redis中取出的验证码进行比对
4. 比对成功后，通过手机号查询数据库是否存在这样的用户，若不存在就新建一个用户
5. 将脱敏后的UserDTO对象，通过工具类BeanUtil转成HashMap类型，存入redis中的hash结构（随机token作为key，用户信息作为value），并设置有效期。
6. 最后返回token给前端，以便拦截器获取它。

3.登录校验和拦截

public class LoginInterceptor implements HandlerInterceptor {
    private StringRedisTemplate stringRedisTemplate;

    public LoginInterceptor(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        //1.获取请求头中的token
        String token = request.getHeader("authorization");
        //2.基于token获取redis中的用户
        String tokenKey = RedisConstants.LOGIN_USER_KEY + token;
        Map<Object, Object> userMap = stringRedisTemplate.opsForHash().entries(tokenKey);
        //3.判断用户是否存在
        if (userMap.isEmpty()){
            //4.用户不存在，拦截
            response.setStatus(401);
            return false;
        }
        //5.将查询到的Hash数据转为UserDTO对象
        UserDTO userDTO = BeanUtil.fillBeanWithMap(userMap, new UserDTO(), false);
        //6.如果存在，保存到ThreadLocal
        UserHolder.saveUser(userDTO);
        //7.更新token有效期
        stringRedisTemplate.expire(tokenKey,RedisConstants.LOGIN_USER_TTL, TimeUnit.MINUTES);
        //8.放行
        return true;
    }

    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
        //注销用户
        UserHolder.removeUser();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

前置拦截方法内：

1. 获取请求头中的token
2. 通过token作为key，获取redis中的用户信息（HashMap类型）
3. 判断用户信息是否存在，不存在则拦截
4. 将Map转成UserDTO类型，并将userDTO保存到ThreadLocal中去。
5. 更新redis中token的有效期。
6. 最后放行。

这里说一下为什么要在拦截器里刷新token的有效期：
如果不刷新token的有效期，用户访问登录接口，过了这个有效期，token就过期失效了。而我们希望的是，只要用户不断的去访问，就应该不断的更新redis中的token。问题是我怎么知道用户什么时候访问，而用户每访问一次登录请求之后，都会经过一次拦截器，所以我们就可以在拦截此时进行token的更新，从而实现用户不断的访问，token不断的更新。

4.拦截器的优化

为了实现用户访问所有的请求都可以刷新token而不只是登录请求，我们需要在登录拦截器基础上再加一层拦截器RefreshTokenInterceptor：
RefreshTokenInterceptor：用来刷新token,并且让他拦截一切的请求。
LoginInterceptor ：只判断ThreadLocal中是否存在用户信息实现拦截和放行。

public class RefreshTokenInterceptor implements HandlerInterceptor {
    private StringRedisTemplate stringRedisTemplate;

    public RefreshTokenInterceptor(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        //1.获取请求头中的token
        String token = request.getHeader("authorization");
        //2.基于token获取redis中的用户
        String tokenKey = RedisConstants.LOGIN_USER_KEY + token;
        Map<Object, Object> userMap = stringRedisTemplate.opsForHash().entries(tokenKey);
        //3.判断用户是否存在
        if (userMap.isEmpty()){
            //4.用户不存在，先放行，交给登录拦截器拦截
            return true;
        }
        //5.将查询到的Hash数据转为UserDTO对象
        UserDTO userDTO = BeanUtil.fillBeanWithMap(userMap, new UserDTO(), false);
        //6.如果存在，保存到ThreadLocal
        UserHolder.saveUser(userDTO);
        //7.更新token有效期
        stringRedisTemplate.expire(tokenKey,RedisConstants.LOGIN_USER_TTL, TimeUnit.MINUTES);
        //8.放行
        return true;
    }

    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
        //注销用户
        UserHolder.removeUser();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

public class LoginInterceptor implements HandlerInterceptor {

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        //1.查看ThreadLocal中是否存在用户信息
        if (UserHolder.getUser()==null){
            response.setStatus(401);
            //2.没有则拦截
            return false;
        }
        //3.有则放行
        return true;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

由于我们加了一个RefreshTokenInterceptor,用于刷新token，且该拦截器应该最先执行。故我们需要在MVC配置类里添加它并设置优先级最高。

@Configuration
public class MvcConfig implements WebMvcConfigurer {
    @Resource
    StringRedisTemplate stringRedisTemplate;
    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(new LoginInterceptor())
                .excludePathPatterns(//排除拦截（放行）的页面视图路径
                        "/shop/**",
                        "/voucher/**",
                        "/shop-type/**",
                        "/upload/**",
                        "/blog/hot",
                        "/user/code",
                        "/user/login"
                ).order(1);
        // token刷新的拦截器 order代表优先级
        registry.addInterceptor(new RefreshTokenInterceptor(stringRedisTemplate)).addPathPatterns("/**").order(0);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

二、商户查询缓存模块

1.查询商品缓存

@Override
    public Result queryShopById(Long id) {
        //1.从redis查询缓存
        String shopKey = RedisConstants.CACHE_SHOP_KEY + id;
        String shopJson = stringRedisTemplate.opsForValue().get(shopKey);
        //2.判断缓存中是否存在
        if (StrUtil.isNotBlank(shopJson)) {
            //3.存在，则返回
            Shop shop = JSONUtil.toBean(shopJson, Shop.class);
            return Result.ok(shop);
        }
        //4.不存在，查询数据库
        Shop shop = getById(id);
        //5.数据库查不到，返回错误信息
        if (shop == null) {
            Result.fail("商铺信息不存在！");
        }
        //6.数据库查的到，存入redis缓存，返回信息
        stringRedisTemplate.opsForValue().set(shopKey, JSONUtil.toJsonStr(shop));
        return Result.ok(shop);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

2.商户类型缓存

	@Override
    public Result queryTypeList() {
        //1.从redis查询缓存的list
        String shopTypeKey = RedisConstants.CACHE_SHOPTYPE_KEY;
        List<String> shopTypeJsonList = stringRedisTemplate.opsForList().range(shopTypeKey, 0, -1);
        //2.判断缓存中list是否存在
        if (!shopTypeJsonList.isEmpty()) {
            //3.存在，则返回
            ArrayList<ShopType> shopTypeList = new ArrayList<>();
            for (String shopTypeJson : shopTypeJsonList) {
                ShopType shopType = JSONUtil.toBean(shopTypeJson, ShopType.class);
                shopTypeList.add(shopType);
            }
            return Result.ok(shopTypeList);
        }
        //4.判断数据库中是否存在
        List<ShopType> typeList = query().orderByAsc("sort").list();
        //5.不存在，则返回错误
        if (typeList.isEmpty()) {
            return Result.fail("分类错误！");
        }
        //6.存在,缓存到redis中去
        //6.1 把泛型为shopType的list转成泛型为json String的list
        ArrayList<String> jsonList = new ArrayList<>();
        for (ShopType shopType : typeList) {
            String json = JSONUtil.toJsonStr(shopType);
            jsonList.add(json);
        }
        //6.2 存入redis中去
        stringRedisTemplate.opsForList().rightPushAll(shopTypeKey,jsonList);
        //7.返回数据
        return Result.ok(typeList);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

3.缓存更新策略

业务场景:
低一致性需求:使用内存淘汰机制。例如店铺类型的查询缓存
高一致性需求:主动更新，并以超时剔除作为兜底方案。例如店铺详情查询的缓存

1.主动更新策略：

1. Cache Aside Pattern：
   在更新数据库的同时更新缓存。
2. Read/Write Through Pattern：
   缓存和数据库整合成为一个服务，由服务来维护一致性。
3. Write Behind Caching Pattern：
   只操作缓存，由其他线程异步的将缓存的数据持久化到数据库中，从而保证一致性。

其中企业中最常用的策略正是Cache Aside Pattern：

2.Cache Aside Pattern需要考虑以下问题：

1. 删除缓存优于更新缓存：
   更新缓存：如果数据库要更新n次，那么缓存也要更新n次，在这更新的n次里如果没有用户访问，并且从缓存取数据的时候只有最后一次更新的数据才是有效的，因此无效的更新操作太多了。
   删除缓存：如果数据库要更新n次，我只用删一次缓存，在这更新的n次里如果没有用户访问，也不用去更新缓存，什么时候有用户访问，什么时候更新缓存，写的操作频率低，有效更新更多。
2. 必须保证缓存与数据库两个线程操作同时成功或失败：
   2.1. 如果是单体系统，缓存和数据库可以放在一个事务里，基于事务的ACID原则（尤其是原子性），可以保证他们同时成功或失败。
   2.2. 如果是分布式系统，缓存操作，数据库操作可能是两个不同的服务，那么此时可以利用TCC等分布式事务方案。
3. 删除缓存和更新数据库讲究顺序问题：

   1. 先删除缓存，再更新数据库：如果有一个线程1做删除缓存，随后更新数据库的操作，在线程1执行过程中，突然穿插了线程2做查询缓存，由于缓存已经被删了，所以未命中，就去查询数据库，最后写入缓存。最后线程1的数据库的更新操作才执行，但是刚刚线程2写入的缓存是数据库更新前的旧数据，这样数据库和缓存的数据就不一致。而且线程2的执行时间远小于线程1的，这种情况发生的概率比较高。

   2. 先更新数据库，再删除缓存：如果有一个线程1做查询缓存，假如缓存过期了，未命中，就去查数据库，查到的数据是旧数据。在线程1执行写入缓存中操作之前，假如此时有线程2更新数据库，数据库的值被更新了，然后删除了缓存，注意缓存本身就没有数据，所以删了相当于没删。然后线程1 的写入缓存的操作才开始执行，此时写的数据是之前查的旧数据，因此缓存里的数据是旧数据，而数据库里的数据是新数据，就不一致了。但是线程2的执行时间是远大于线程1的，所以这种情况发生的概率比较小。

   3. 因此我们先更新数据库，再删除缓存的执行顺序更能保护我们的线程安全。

那么有了缓存更新策略的理论支持：我们来实现商铺缓存与数据库的双写一致：

4.商铺缓存与数据库的双写一致

修改ShopController中的业务逻辑，满足下面的需求:

1. 根据id查询店铺时，如果缓存未命中，则查询数据库，将数据库结果写入缓存并设置超时时间
2. 根据id修改店铺时，先更新数据库，再删除缓存

修改queryShopById方法：增加expire参数，实现缓存超时剔除。

	//6.数据库查的到，存入redis缓存，返回信息
	stringRedisTemplate.opsForValue().set(shopKey, JSONUtil.toJsonStr(shop),
	RedisConstants.CACHE_SHOP_TTL, TimeUnit.MINUTES);
	return Result.ok(shop);
1
2
3
4

更新商户信息：缓存与数据库的双写一致

	@Override
    @Transactional
    public Result update(Shop shop) {
        //1.更新数据库
        Long id = shop.getId();
        if (id == null) {
            return Result.fail("商户id不能为空！");
        }
        updateById(shop);
        //2.删除缓存
        stringRedisTemplate.delete(RedisConstants.CACHE_SHOP_KEY + id);
        return Result.ok();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13

5.缓存穿透

缓存穿透是指用户请求的数据在缓存和数据库中都不存在的情况，缓存永远不会生效，这些请求会打到数据库中查到一个null的数据。
解决方案：

1. 缓存空对象：
   
   当缓存和数据库都未命中时，缓存一个null的空对象，并设置TTL有效期，用来清除多次缓存穿透造成的额外内存消耗。优点是实现简单，便于维护。缺点是缓存了太多不必要的无效数据，额外内存消耗。以及可能造成短期数据不一致：在发生缓存穿透之后，刚好数据库更新了，就会造成数据不一致。
2. 布隆过滤器
   
   在用户请求后，在redis缓存之前加一层布隆过滤器，用来过滤可能会发生缓存穿透的请求，通过数据库中的Hash值转换成二进制存放在布隆过滤器中，从而判断是否请求是否能命中缓存和数据库。优点是内存占用少，没有多余的key。但是缺点是可能会发生误判，并且布隆过滤器实现复杂。

1.解决缓存穿透

需要在原来基础上修改两个地方：
其一是：在查询数据库时如果未命中，就把空值写入redis中去。
其二是：为了防止请求命中这个空值的缓存，我们还需要在缓存命中后判断命中的是否为空值。

    @Override
    public Result queryShopById(Long id) {
        //1.从redis查询缓存
        String shopKey = CACHE_SHOP_KEY + id;
        String shopJson = stringRedisTemplate.opsForValue().get(shopKey);
        //2.判断缓存中是否存在
        if (StrUtil.isNotBlank(shopJson)) {
            //3.存在，则返回
            Shop shop = JSONUtil.toBean(shopJson, Shop.class);
            return Result.ok(shop);
        }
        //缓存命中后，需要判断命中的是否为空值
        if (shopJson!=null){//缓存里没有值，又不为空，只能是""
            return Result.fail("店铺不存在！");
        }
        //4.不存在，查询数据库
        Shop shop = getById(id);
        //5.数据库查不到，返回错误信息
        if (shop == null) {
            //5.1为了解决缓存穿透，把未命中的空值写入到redis中去
            stringRedisTemplate.opsForValue().set(shopKey, "",CACHE_NULL_TTL, TimeUnit.MINUTES);
            return Result.fail("店铺不存在！");
        }
        //6.数据库查的到，存入redis缓存，返回信息
        stringRedisTemplate.opsForValue().set(shopKey, JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.MINUTES);
        return Result.ok(shop);
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

2.总结

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

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

缓存穿透的解决方案有哪些?

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

6.缓存雪崩

缓存雪崩是指在同一时段大量的缓存key同时失效或者Redis服务宕机，导致大量请求到达数据库，带来巨大压力。

解决方案:

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

7.缓存击穿

1.缓存击穿应用场景

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

2.解决方案：

1. 互斥锁
2. 逻辑过期
   
   
   
   总结：追求可用性则用逻辑过期，追求一致性则用互斥锁。

3.用互斥锁解决缓存击穿

利用redis里的setNX命令类似于互斥锁的机制，定义获取锁和释放锁两个方法：

	private boolean tryLock(String key) {
        //如果不存在这样的key才set value，并设置ttl作为兜底，防止锁没释放的意外
        Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);
        return BooleanUtil.isTrue(flag);
    }

    private void unlock(String key) {
        stringRedisTemplate.delete(key);
    }
1
2
3
4
5
6
7
8
9

解决缓存击穿的方法queryWithPassMutex：

       public Shop queryWithPassMutex(Long id) {
        //1.从redis查询缓存
        String shopKey = CACHE_SHOP_KEY + id;
        String shopJson = stringRedisTemplate.opsForValue().get(shopKey);
        //2.判断缓存中是否命中
        if (isCacheExist(shopKey)) {
            //3.缓存命中，则返回shop
            return JSONUtil.toBean(shopJson, Shop.class);
        }
        //3.缓存命中后，需要判断命中的是否为空值
        if (shopJson != null) {//缓存里没有值，又不为空，只能是""
            return null;
        }
        //4.缓存未命中，实现缓存重建，解决缓存击穿
        // 4.1 获取互斥锁
        Shop shop = null;
        String lockKey = LOCK_SHOP_KEY + id;
        try {
            boolean isLock = tryLock(lockKey);
            // 4.2 判断是否获取成功
            if (!isLock) {
                // 4.3 获取锁失败，休眠，重试
                Thread.sleep(50);
                queryWithPassMutex(id);
            }
            //4.4 获取锁成功，再次检查redis缓存中是否命中，如果命中，则直接返回
            if (isCacheExist(shopKey)) {
                return JSONUtil.toBean(stringRedisTemplate.opsForValue().get(shopKey), Shop.class);
            }
            //4.5 未命中，则查询数据库
            shop = getById(id);
            //模拟重建的延时
            Thread.sleep(200);
            //5.数据库查不到，返回错误信息
            if (shop == null) {
                //5.1为了解决缓存穿透，把未命中的空值写入到redis中去
                stringRedisTemplate.opsForValue().set(shopKey, "", CACHE_NULL_TTL, TimeUnit.MINUTES);
                return null;
            }
            //6.数据库查的到，存入redis缓存，返回信息
            stringRedisTemplate.opsForValue().set(shopKey, JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            //7.释放互斥锁
            unlock(lockKey);
        }
        return shop;
    }

    /**
     * 判断缓存中是否命中
     * @param key
     * @return boolean
     */
    public Boolean isCacheExist(String key){
        String Json = stringRedisTemplate.opsForValue().get(key);
        //2.判断缓存中是否存在
        if (StrUtil.isNotBlank(Json)) {
            //3.存在，则返回
            return true;
        }
        return false;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

主调用方法：

	@Override
    public Result queryShopById(Long id) {
        Shop shop = queryWithPassMutex(id);
        if (shop == null) {
            return Result.fail("店铺不存在！");
        }
        return Result.ok(shop);
    }
1
2
3
4
5
6
7
8

接下来我们用jmeter压力测试工具进行高并发测试（首先要确保redis中缓存中没有这个key）：

每秒查询效率（QPS=200）每秒查询200个线程。

运行完发现，数据库只访问了一次，却完成了1000个线程的并发。
其原理就是：
第1个线程，缓存未命中，就去获取锁，获取锁成功后，再次二次检查缓存是否未命中，未命中的情况下，就访问数据库，就把数据存入redis缓存中了，最后释放锁，返回数据。而第2个线程，是和第一个线程并行，但是获取锁失败，于是就休眠，直到等待第一个锁释放，此时缓存中已经有数据了，因此就直接返回了，就不会访问数据库。此后的所有线程与第2给线程一样。都直接从缓存中取，因此，数据库只走了一次。

4.用逻辑过期解决缓存击穿

逻辑过期要求我们的热点key的有效期是永久的。因此我们第一次要做缓存重建。在第一个线程判断逻辑过期时间时，若不提前做缓存重建，就会报空指针异常：

@SpringBootTest
class HmDianPingApplicationTests {
    @Resource
    ShopServiceImpl shopService;
    @Test
    void testSaveShop() throws InterruptedException {
        shopService.saveShopToRedis(1L,10L);
    }
}
1
2
3
4
5
6
7
8
9

  public Shop queryWithLogicExpire(Long id) {

        //1.从redis查询缓存
        String shopKey = CACHE_SHOP_KEY + id;
        String shopJson = stringRedisTemplate.opsForValue().get(shopKey);
        //2.判断缓存中是否命中
        if (!isCacheExist(shopKey)) {
            //3.1 缓存未命中，则返回空
            return null;
        }
        //4.将json反序列化为RedisData对象
        RedisData redisData = JSONUtil.toBean(shopJson, RedisData.class);
        Shop shop = JSONUtil.toBean((JSONObject) redisData.getData(), Shop.class);
        LocalDateTime expireTime = redisData.getExpireTime();
        //3.2 缓存命中后，需要判断缓存是否过期
        if (expireTime.isAfter(LocalDateTime.now())) {
            //4.1 未过期，返回旧数据
            return shop;
        }
        //4.2 过期,缓存重建
        //5.尝试获取互斥锁
        String lockKey = LOCK_SHOP_KEY + id;
        //6.判断是否获取到锁
        boolean isLock = tryLock(lockKey);
        if (isLock) {
            //6.1 获取成功，则开启一个独立线程做缓存重建
            CACHE_REBUILD_EXECUTOR.submit(() -> {
                try {
                    saveShopToRedis(id, 20L);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                } finally {
                    unlock(lockKey);
                }
            });
        }
        //6.2 获取失败
        //7.统一返回
        return shop;
    }
    /**
     * 判断缓存中是否命中
     *
     * @param key
     * @return boolean
     */
    public Boolean isCacheExist(String key) {
        String Json = stringRedisTemplate.opsForValue().get(key);
        //2.判断缓存中是否存在
        if (StrUtil.isNotBlank(Json)) {
            //3.存在，则返回
            return true;
        }
        return false;
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

   private void saveShopToRedis(Long id, Long expireSeconds) throws InterruptedException {
        //1.查询数据库数据
        Shop shop = getById(id);
        //模拟重建延时
        Thread.sleep(200);
        //2.封装逻辑过期时间
        RedisData redisData = new RedisData(LocalDateTime.now().plusSeconds(expireSeconds), shop);
        System.out.println(LocalDateTime.now().plusSeconds(expireSeconds));
        //3.写入Redis
        stringRedisTemplate.opsForValue().set(CACHE_SHOP_KEY + id, JSONUtil.toJsonStr(redisData));
    }
1
2
3
4
5
6
7
8
9
10
11

8.缓存封装工具类和总结

@Slf4j
@Configuration
public class CacheClient {
    @Resource
    private StringRedisTemplate stringRedisTemplate;

    private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);

    //写入redis中
    public void set(String key, Object value, Long time, TimeUnit unit) {
        stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(value), time, unit);
    }
    //设置逻辑过期 & 写入Redis
    public void setWithLogicalExpire(String key, Object value, Long time, TimeUnit unit) {
        // 设置逻辑过期
        RedisData redisData = new RedisData();
        redisData.setData(value);
        redisData.setExpireTime(LocalDateTime.now().plusSeconds(unit.toSeconds(time)));
        // 写入Redis
        stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(redisData));
    }

    /**
     * 缓存穿透
     *
     * @param keyPrefix
     * @param id
     * @param type
     * @param dbFallback
     * @param time
     * @param timeUnit
     * @param 
     * @param 
     * @return
     */
    public <R, ID> R queryWithPassThrough(String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit timeUnit) {
        //1.从redis查询缓存
        String key = keyPrefix + id;
        String json = stringRedisTemplate.opsForValue().get(key);
        //2.判断缓存中是否存在
        if (isCacheExist(key)) {
            //3.存在，则返回
            return JSONUtil.toBean(json, type);
        }
        //缓存命中后，需要判断命中的是否为空值
        if (json != null) {//缓存里没有值，又不为空，只能是""
            return null;
        }
        //4.不存在，查询数据库
        R r = dbFallback.apply(id);
        //5.数据库查不到，返回错误信息
        if (r == null) {
            //5.1为了解决缓存穿透，把未命中的空值写入到redis中去
            set(key,"",CACHE_NULL_TTL,TimeUnit.MINUTES);
            return null;
        }
        //6.数据库查的到，存入redis缓存，返回信息
        set(key, r, time, timeUnit);
        return r;
    }



    /**
     * 互斥锁解决缓存击穿
     * @param keyPrefix
     * @param id
     * @param type
     * @param dbFallback
     * @param time
     * @param timeUnit
     * @param 
     * @param 
     * @return
     */
    public <R, ID> R queryWithMutex(String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit timeUnit) {
        //1.从redis查询缓存
        String key = keyPrefix + id;
        String json = stringRedisTemplate.opsForValue().get(key);
        //2.判断缓存中是否命中
        if (isCacheExist(key)) {
            //3.缓存命中，则返回shop
            return JSONUtil.toBean(json, type);
        }
        //3.缓存命中后，需要判断命中的是否为空值
        if (json != null) {//缓存里没有值，又不为空，只能是""
            return null;
        }
        //4.缓存未命中，实现缓存重建，解决缓存击穿
        // 4.1 获取互斥锁
        R r = null;
        String lockKey = LOCK_SHOP_KEY + id;
        try {
            boolean isLock = tryLock(lockKey);
            // 4.2 判断是否获取成功
            if (!isLock) {
                // 4.3 获取锁失败，休眠，重试
                Thread.sleep(50);
                return queryWithMutex(keyPrefix, id, type, dbFallback, time, timeUnit);
            }
            //4.4 获取锁成功，再次检查redis缓存中是否命中，如果命中，则直接返回
            if (isCacheExist(key)) {
                return JSONUtil.toBean(stringRedisTemplate.opsForValue().get(key), type);
            }
            //4.5 未命中，则查询数据库
            r = dbFallback.apply(id);
            //模拟重建的延时
            Thread.sleep(200);
            //5.数据库查不到，返回错误信息
            if (r == null) {
                //5.1为了解决缓存穿透，把未命中的空值写入到redis中去
                set(key,"",CACHE_NULL_TTL,TimeUnit.MINUTES);
                return null;
            }
            //6.数据库查的到，存入redis缓存，返回信息
            set(key,r,time,timeUnit);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            //7.释放互斥锁
            unlock(lockKey);
        }
        return r;
    }

    /**
     * 逻辑过期解决缓存击穿
     * @param keyPrefix
     * @param id
     * @param type
     * @param dbFallback
     * @param time
     * @param timeUnit
     * @param 
     * @param 
     * @return
     */
    public <R, ID> R queryWithLogicExpire(String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit timeUnit) {

        //1.从redis查询缓存
        String key = keyPrefix + id;
        String json = stringRedisTemplate.opsForValue().get(key);
        //2.判断缓存中是否命中
        if (!isCacheExist(key)) {
            //3.1 缓存未命中，则返回空
            return null;
        }
        //4.将json反序列化为RedisData对象
        RedisData redisData = JSONUtil.toBean(json, RedisData.class);
        R r = JSONUtil.toBean((JSONObject) redisData.getData(), type);
        LocalDateTime expireTime = redisData.getExpireTime();
        //3.2 缓存命中后，需要判断缓存是否过期
        if (expireTime.isAfter(LocalDateTime.now())) {
            //4.1 未过期，返回旧数据
            return r;
        }
        //4.2 过期,缓存重建
        //5.尝试获取互斥锁
        String lockKey = LOCK_SHOP_KEY + id;
        //6.判断是否获取到锁
        boolean isLock = tryLock(lockKey);
        if (isLock) {
            //6.1 获取成功，则开启一个独立线程做缓存重建
            CACHE_REBUILD_EXECUTOR.submit(() -> {
                try {
                    R newR = dbFallback.apply(id);
                    //缓存重建
                    setWithLogicalExpire(key, newR, time, timeUnit);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                } finally {
                    unlock(lockKey);
                }
            });
        }
        //6.2 获取失败
        //7.统一返回
        return r;
    }
    /**
     * 判断缓存中是否存在
     *
     * @param key
     * @return
     */
    public Boolean isCacheExist(String key) {
        String Json = stringRedisTemplate.opsForValue().get(key);
        //2.判断缓存中是否存在
        if (StrUtil.isNotBlank(Json)) {
            //3.存在，则返回
            return true;
        }
        return false;
    }
    
    private boolean tryLock(String key) {
        //如果不存在这样的key才set value，并设置ttl作为兜底，防止锁没释放的意外
        Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", LOCK_SHOP_TTL, TimeUnit.SECONDS);
        return BooleanUtil.isTrue(flag);
    }

    private void unlock(String key) {
        stringRedisTemplate.delete(key);
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208

@Service
public class ShopServiceImpl extends ServiceImpl<ShopMapper, Shop> implements IShopService {
    @Resource
    StringRedisTemplate stringRedisTemplate;

    @Resource
    CacheClient cacheClient;

    @Override
    public Result queryShopById(Long id) {
//        Shop shop = cacheClient.queryWithPassThrough(CACHE_SHOP_KEY, id, Shop.class, this::getById, CACHE_SHOP_TTL, TimeUnit.MINUTES);
        //互斥锁解决缓存击穿
        Shop shop = cacheClient.queryWithMutex(CACHE_SHOP_KEY, id, Shop.class, this::getById, CACHE_SHOP_TTL, TimeUnit.SECONDS);
        //逻辑过期解决缓存击穿
//        Shop shop = cacheClient.queryWithLogicExpire(CACHE_SHOP_KEY, id, Shop.class, this::getById, CACHE_SHOP_TTL, TimeUnit.SECONDS);
        if (shop == null) {
            return Result.fail("店铺不存在！");
        }
        return Result.ok(shop);
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

三、优惠券秒杀模块

1.全局自增ID

全局唯一ID生成策略:

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

Redis自增ID策略:

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

获取某个时刻的时间戳：

	 public static void main(String[] args) {
        LocalDateTime time = LocalDateTime.of(2022, 10, 10, 0, 0, 0);
        long second = time.toEpochSecond(ZoneOffset.UTC);
        System.out.println(second);
    }
1
2
3
4
5

把获取到的时间戳定义成常量

    private static final long BEGIN_TIMESTAMP =1665360000L;
1

自增ID的实现：

@Component
public class RedisIdWorker {
    @Resource
    private StringRedisTemplate stringRedisTemplate;

    private static final long BEGIN_TIMESTAMP =1665360000L;

    private static final int COUNT_BITS = 32;
    public long nextId(String keyPrefix){
        //1.生成时间戳
        LocalDateTime now = LocalDateTime.now();
        long nowSecond = now.toEpochSecond(ZoneOffset.UTC);
        long timeStamp = nowSecond - BEGIN_TIMESTAMP;
        //2.生成序列号（默认32bit）
        //2.1 获取当前日期
        String date = now.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
        //2.2 自增长
        Long count = stringRedisTemplate.opsForValue().increment("incr" + keyPrefix + ":" + date);
        //3.拼接时间戳和序列号并返回:将时间戳的最高位向左移动32（序列号占32）位,并把自增长结果补给余下位
        return timeStamp << COUNT_BITS | count;
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

测试：

@Resource
    private RedisIdWorker redisIdWorker;
    // 线程池
    private ExecutorService executorService = Executors.newFixedThreadPool(500);

    @Test
    void testIdWorker() throws InterruptedException {
        //为了让begin和end标志位和所有线程一起执行，使用CountDownLatch
        CountDownLatch latch = new CountDownLatch(300);

        Runnable task = () -> {
            for (int i = 0; i < 100; i++) {
                long id = redisIdWorker.nextId("order");
                System.out.println("id=" + id);
            }
            //任务执行完之前countDown，记录begin
            latch.countDown();
        };
        long begin = System.currentTimeMillis();
        for (int i = 0; i < 300; i++) {
            executorService.submit(task);
        }
        //任务提交后等待，记录end
        latch.await();
        long end = System.currentTimeMillis();
        System.out.println("executeTime=" + (end - begin));
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

2.添加优惠券

拦截器开放voucher请求

	 registry.addInterceptor(new LoginInterceptor())
                .excludePathPatterns(
                        "/user/code",
                        "/user/login",
                        "/shop-type/**",
                        "/shop/**",
                        "/voucher/**"
                ).order(1);
1
2
3
4
5
6
7
8

使用postman发送请求

随后查看数据库以及前端即可。

3.实现下单秒杀

1. 创建业务方法seckillVoucher

@RestController
@RequestMapping("/voucher-order")
public class VoucherOrderController {
    @Resource
    private IVoucherOrderService voucherOrderService;
    @PostMapping("seckill/{id}")
    public Result seckillVoucher(@PathVariable("id") Long voucherId) {
        return voucherOrderService.seckillVoucher(voucherId);
    }
}
1
2
3
4
5
6
7
8
9
10

2. 编写业务逻辑，由于操作了两个不同的表，为了原子性，加一个事务注解。

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Resource
    ISeckillVoucherService seckillVoucherService;
    @Resource
    RedisIdWorker redisIdWorker;

    @Override
    @Transactional
    public Result seckillVoucher(Long voucherId) {
        //1. 查询优惠券信息
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        //2. 判断秒杀是否开始
        if (voucher.getCreateTime().isAfter(LocalDateTime.now())) {
            //3. 尚未开始 返回异常
            return Result.fail("秒杀未开始!");
        }
        if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
            //3.1 结束，返回异常
            return Result.fail("秒杀已结束！");
        }
        //4. 开始，先判断库存是否充足
        if (voucher.getStock() < 1) {
            //5. 不充足，返回异常
            return Result.fail("库存不足！");
        }
        //6. 充足，扣减库存
        boolean success = seckillVoucherService.update()
                .setSql("stock= stock -1")
                .eq("voucher_id", voucherId).update();
        if (!success) {
            return Result.fail("库存不足！");
        }
        //7. 创建订单:订单id，用户id，代金券id
        VoucherOrder voucherOrder = new VoucherOrder();
        long orderId = redisIdWorker.nextId("order");
        voucherOrder.setId(orderId);

        Long userId = UserHolder.getUser().getId();
        voucherOrder.setUserId(userId);

        voucherOrder.setVoucherId(voucherId);

        save(voucherOrder);
        //8. 返回
        return Result.ok(orderId);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

4.库存超卖问题

乐观锁：版本号法和CAS法。
版本号法（增加一个版本号用于判断是否被修改过）：

CAS法（根据数据本身是否被修改作为条件）：

对于当前线程：判断库存值是否修改过，如果库存值与原来查询的库存值一致，说明没有被修改过，则放心大胆的去扣减库存。如果不一致，说明已经有别的线程修改过了，就不进行扣减库存。

但是这样做的弊端就是，对于库存只剩最后一件的情况这么做，才能实现库存不被超卖，但是对于库存很充足的情况下，如果用乐观锁，则会导致其他线程以为上一个线程修改过了而不去扣减库存的情况，因此这里的where条件不应该是stock = 1而是stock > 0;这样就只针对只剩最后一件库存的去情况去保证库存超卖的问题:

于是seckillVoucher方法里进行修改这一行代码：

boolean success = seckillVoucherService.update()
                .setSql("stock= stock -1")
                .eq("voucher_id", voucherId)
                .gt("stock",0)
                .update();
1
2
3
4
5

总结：

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

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

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

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

5.解决一人一单功能

1.业务逻辑

在原来基础上，判断库存充足之后，如果充足，则根据优惠券id和用户id查询是否有唯一的订单存在，如果不存在，说明该用户之前没有下过单，此时就可以扣减库存和创建订单。如果存在，说明用户之前下过单，则返回异常信息即可。

    @Override
    @Transactional
    public Result seckillVoucher(Long voucherId) {
        //1. 查询优惠券信息
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        //2. 判断秒杀是否开始
        if (voucher.getCreateTime().isAfter(LocalDateTime.now())) {
            //3. 尚未开始 返回异常
            return Result.fail("秒杀未开始!");
        }
        if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
            //3.1 结束，返回异常
            return Result.fail("秒杀已结束！");
        }
        //4. 开始，先判断库存是否充足
        if (voucher.getStock() < 1) {
            //5. 不充足，返回异常
            return Result.fail("库存不足！");
        }
        //6.充足，一人一单：根据优惠券id和用户id查询唯一订单
        Long userId = UserHolder.getUser().getId();
        Long count = query().eq("voucher_id", voucherId).eq("user_id", userId).count();
        //6.1判断订单是否存在
        if (count >0) {
            return Result.fail("不可以重复下单！");
        }
        //7. 充足，扣减库存
        boolean success = seckillVoucherService.update()
                .setSql("stock= stock -1")
                .eq("voucher_id", voucherId)
                .gt("stock",0)
                .update();
        if (!success) {
            return Result.fail("库存不足！");
        }
        //8. 创建订单:订单id，用户id，代金券id
        VoucherOrder voucherOrder = new VoucherOrder();
        long orderId = redisIdWorker.nextId("order");
        voucherOrder.setId(orderId);

        voucherOrder.setUserId(userId);

        voucherOrder.setVoucherId(voucherId);

        save(voucherOrder);
        //9. 返回
        return Result.ok(orderId);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

2.解决线程安全问题

在高并发的情况下，根据优惠券id和用户id查询的count值可能都为0.就会出现一个用户重复下单的情况。但是我们此时的业务场景是查询，无法根据是否被修改来加乐观锁，因此我们只能加悲观锁，这里用户是唯一的，因此以userId作为关键字加锁是最理想的：

    @Transactional
    public Result createVoucherOrder(Long voucherId) {
        //6.充足，一人一单：根据优惠券id和用户id查询唯一订单
        Long userId = UserHolder.getUser().getId();
        synchronized (userId.toString().intern()) {
            Long count = query().eq("voucher_id", voucherId).eq("user_id", userId).count();
            //6.1判断订单是否存在
            if (count > 0) {
                return Result.fail("不可以重复下单！");
            }
            //7. 充足，扣减库存
            boolean success = seckillVoucherService.update()
                    .setSql("stock= stock -1")
                    .eq("voucher_id", voucherId)
                    .gt("stock", 0)
                    .update();
            if (!success) {
                return Result.fail("库存不足！");
            }
            //8. 创建订单:订单id，用户id，代金券id
            VoucherOrder voucherOrder = new VoucherOrder();
            long orderId = redisIdWorker.nextId("order");
            voucherOrder.setId(orderId);

            voucherOrder.setUserId(userId);

            voucherOrder.setVoucherId(voucherId);

            save(voucherOrder);
            //9. 返回
            return Result.ok(orderId);
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

但是@Transactional注解，事务是在这个方法执行完之后（也就是最后一个花括号）才提交的，而synchronized锁是在倒数第二个花括号执行完后释放的。此时其他线程就可以进来了，而事务尚未提交，就会造成线程不安全问题。所以我们需要让锁的范围扩大至整个方法，以保证事务提交之后再释放锁：

    @Override
    public Result seckillVoucher(Long voucherId) {
        //1. 查询优惠券信息
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        //2. 判断秒杀是否开始
        if (voucher.getCreateTime().isAfter(LocalDateTime.now())) {
            //3. 尚未开始 返回异常
            return Result.fail("秒杀未开始!");
        }
        if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
            //3.1 结束，返回异常
            return Result.fail("秒杀已结束！");
        }
        //4. 开始，先判断库存是否充足
        if (voucher.getStock() < 1) {
            //5. 不充足，返回异常
            return Result.fail("库存不足！");
        }
        Long userId = UserHolder.getUser().getId();
        synchronized (userId.toString().intern()) {
            return createVoucherOrder(voucherId);//让锁的范围扩大至整个方法，
            //以保证事务提交之后再释放锁：
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

但是由于我们@Transactional事务加在了createVoucherOrder方法上，seckillVoucher方法却没有（因为这一块不需要事务）。那么我们调用createVoucherOrder方法实质上是通过this调用，this就是实现类VoucherOrderServiceImpl。而不是代理对象，而事务的本质是动态代理，this是目标对象，而非代理对象，所以这里的事务就不能生效。
解决方案：

   synchronized (userId.toString().intern()) {
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        }
1
2
3
4

拿到代理对象，并用接口去生成代理对象。这个代理对象是接口的对象，所以接口里要重写createVoucherOrder方法。

并且导入织入依赖aspectjweaver。

以及暴露代理

@EnableAspectJAutoProxy(exposeProxy = true)
@MapperScan("com.hmdp.mapper")
@SpringBootApplication
public class HmDianPingApplication {
1
2
3
4

6.集群下一人一单功能的并发线程安全问题

当nginx负载均衡多台服务器做集群的情况下，每一个JVM都会有锁监视器，每一个jvm只能确保锁自己线程池中的线程，这就导致其他的JVM也会并发的执行，而不会受到别的JVM锁的影响，从而导致并发线程安全问题。
解决方案：分布式锁。

四、分布式锁

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

1.分布式锁的实现

public class SimpleRedisLock implements ILock {
    private String name;
    private StringRedisTemplate stringRedisTemplate;

    public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {
        this.name = name;
        this.stringRedisTemplate = stringRedisTemplate;
    }

    private static final String KEY_PREFIX = "lock:";

    @Override
    public boolean tryLock(long timeoutSec) {
        //1.获取当前线程id
        String threadName = Thread.currentThread().getName();
        //2.获取key
        String key = KEY_PREFIX + name;
        //3.设置锁
        Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, threadName, timeoutSec, TimeUnit.SECONDS);
        return BooleanUtil.isTrue(success);
    }

    @Override
    public void unlock() {
        stringRedisTemplate.delete(KEY_PREFIX + name);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

    @Override
    public Result seckillVoucher(Long voucherId) {
        //1. 查询优惠券信息
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        //2. 判断秒杀是否开始
        if (voucher.getCreateTime().isAfter(LocalDateTime.now())) {
            //3. 尚未开始 返回异常
            return Result.fail("秒杀未开始!");
        }
        if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
            //3.1 结束，返回异常
            return Result.fail("秒杀已结束！");
        }
        //4. 开始，先判断库存是否充足
        if (voucher.getStock() < 1) {
            //5. 不充足，返回异常
            return Result.fail("库存不足！");
        }
        Long userId = UserHolder.getUser().getId();
        //创建锁对象
        SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
        boolean isLock = lock.tryLock(1200);
        //判断锁是否获取成功
        if (!isLock) {
            return Result.fail("不允许重复下单！");
        }
        try {
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        } finally {
            //释放锁
            lock.unlock();
        }

    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

2.分布式锁极端情况：超时释放锁导致误删其他线程锁

解决方案：在每次释放锁之前判断一下锁的标识是否与当前锁一致，如果是则释放，否则什么都不做。另外要在每次获取锁的时候存入线程标识。

private static final String KEY_PREFIX = "lock:";
    private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";

    @Override
    public boolean tryLock(long timeoutSec) {
        //1.获取当前线程标识
        String theadId = ID_PREFIX + Thread.currentThread().getId();
        //2.获取key
        String key = KEY_PREFIX + name;
        //3.设置锁
        Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, theadId, timeoutSec, TimeUnit.SECONDS);
        return BooleanUtil.isTrue(success);
    }

    @Override
    public void unlock() {
        //获取线程标识
        String theadId = ID_PREFIX + Thread.currentThread().getId();
        //获取锁的标识
        String lockId = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
        //判断是否一致
        if (theadId.equals(lockId)) {
            //释放锁
            stringRedisTemplate.delete(KEY_PREFIX + name);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

3.分布式极端情况：释放锁阻塞

基于上一种解决方案下，如果gc回收的时候导致判断完后释放锁阻塞了：

解决方案：让判断标识是否一致和释放锁的动作具有原子性。
那么怎么保证这两个动作原子性呢？

4.lua脚本实现多条命令原子性问题

--获取锁的标识
local lockId = redis.call("get", KEYS[1]);
--获取线程的标识
local threadId = ARGS[1];
--判断是否一致
if threadId==lockId then
    return redis.call("del",KEYS[1])
end
return 0;
1
2
3
4
5
6
7
8
9

简化后

if redis.call("get", KEYS[1])==ARGS[1] then
    return redis.call("del",KEYS[1])
end
return 0;
1
2
3
4

释放锁的方法里就一行代码用于调用lua脚本，但是在此之前DefaultRedisScript类需要在类加载之前初始化：

 private static final String KEY_PREFIX = "lock:";
    private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";
    private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;
    static {
        UNLOCK_SCRIPT = new DefaultRedisScript<>();
        UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
        UNLOCK_SCRIPT.setResultType(Long.class);
    }
    @Override
    public boolean tryLock(long timeoutSec) {
        //1.获取当前线程标识
        String theadId = ID_PREFIX + Thread.currentThread().getId();
        //2.获取key
        String key = KEY_PREFIX + name;
        //3.设置锁
        Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(key, theadId, timeoutSec, TimeUnit.SECONDS);
        return BooleanUtil.isTrue(success);
    }
    @Override
    public void unlock() {
       stringRedisTemplate.execute(
               UNLOCK_SCRIPT,
               Collections.singletonList(KEY_PREFIX + name),
               ID_PREFIX + Thread.currentThread().getId());
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

五、Redisson

1.入门

配置redisson：

		
        <dependency>
            <groupId>org.redissongroupId>
            <artifactId>redissonartifactId>
            <version>3.18.0version>
        dependency>
1
2
3
4
5
6

@Configuration
public class RedissonConfig {
    @Bean
    public RedissonClient redissonClient(){
        //配置
        Config config = new Config();
        config.useSingleServer().setAddress("redis://192.168.239.130:6379").setPassword("123321");
        //创建客户端
        return Redisson.create(config);
    }
}
1
2
3
4
5
6
7
8
9
10
11

注入redissonClient到实现类，通过getLock方法获取锁：

	   //创建锁对象
	   //SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
        RLock lock = redissonClient.getLock("order:" + userId);
        boolean isLock = lock.tryLock();
        //判断锁是否获取成功
        if (!isLock) {
            return Result.fail("不允许重复下单！");
        }
        try {
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        } finally {
            //释放锁
            lock.unlock();
        }

    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

2.Redisson的可重入锁原理

获取锁时，除了保存线程标识，还要保存一个锁的计数器，用来表示锁被重入的次数。所以采用hash结构进行存储。
第一次tryLock，初始化锁的计数器为1，当同一个线程里，再次被重入时，计数+1。当unlock时，锁的计数-1。当所有的锁被释放后，同一个线程的锁的计数一定为0。

具体流程：
tryLock：判断锁是否存在，（这里的线程标识用于下一次被重入时判断是否为同一把锁）
如果锁不存在，就获取锁并添加线程标识，设置锁的有效期
如果锁存在，就通过线程标识判断是否是同一个线程下的锁，
        如果不是，则获取失败。
        如果是，说明可重入，则锁计数+1，然后设置有效期。

unLock：通过线程标识来判断锁是否是自己的锁，
如果不是，则说明锁已经被释放了，就不用再释放。
如果是，则锁计数-1，然后在执行锁释放前，需要先判断锁计数是不是0，
        如果不是0，则需要重置锁的有效期，再回到之前判断步骤。
        如果是0，就可以放心释放锁。

为了确保原子性，这些都会被编写成lua脚本在Redisson的tryLock方法和unLock方法的源码里。

3.Redisson的可重试和超时释放

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

可重试源码解析：
tryLock：
参数有等待时间，释放时间，单位元
进来之后首先，把等待时间转换成毫秒单位参与后面的运算，然后获取当前时间，以及线程Id
调用tryAcquire方法，返回一个ttl有效期（剩余时间）
进入tryAcquire方法里：如果没有传释放时间，也就是小于0，则释放时间会给一个默认值，看门狗超时释放值为30秒，
然后调用执行获取锁的lua脚本，这段脚本的包含了可重入的功能：
首先判断锁是否存在，如果不存在，就可重试计数加1，设置有效期；如果存在，就判断锁是不是自己的，如果是就重试计数加1，设置有效期。
获取锁成功都返回nil，失败则返回一个ttl，也就是剩余有效期。
如果有效期等于null，则返回true，代表锁获取成功
否则就是失败的情况：当前时间减去尝试获取锁之前的时间，得到尝试获取锁消耗的时间。然后再用等待时间减去消耗时间，得到剩余等待时间。
如果剩余等待时间小于等于0，说明消耗时间太长了以至于把等待时间都消耗完了，因此返回一个false，获取失败
如果剩余等待时间大于0，记录当前时间，然后订阅当前线程上一次释放锁的信号，返回一个future结果。
然后通过future来判断剩余等待时间内有没有得到释放，如果剩余等待时间内还没有收到释放的通知，也就是超时了，就取消这个订阅，然后返回false
如果没有超时，就再次根据当前时间计算剩余等待时间，便开始重试。再重试的时候，不是立马就重试，依然要通过futrue结果通过信号量的方式去、
类似的，如果有效期ttl小于剩余等待时间，说明等待的时候就已经释放了，那就没有必要再等了，所以执行tryAcquire的等待时间参数就是有效期ttl，等ttl时间释放即可。
如果有效期ttl大于剩余等待时间，说明剩余等待时间到期了还没释放，执行tryAcquire的等待时间参数就是等待时间，这里面获取锁失败交给它来做。
最后再计算一下剩余等待时间，如果没有了，就返回false，否则继续重试。逻辑同上。

总结：

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

4. Redisson的主从一致性解决方案

在Redis里，尝试获取锁的时候，假如java客户端有执行set lock thead1 Nx命令，向redis主节点发起，就在主节点存入了这个锁，为了保证安全性，redis通常还有一个从节点去同步主节点。但是在还没同步之前，一旦主节点宕机了，我们的redis就会把从节点当成主节点，但是此时的主节点，是没有锁的，这样下一个线程就能获取到了，就会造成线程不安全。
那么Redisson是怎么解决的呢？

Redisson没有主节点也没有从节点，如果一个线程要想成功获取到锁，必须拿到所有的节点的锁，一旦有一个不成功，就会获取失败。那么如果有一个节点宕机了，那么这个节点是没有锁的，这个时候如果有其他线程来的时候，它能获取到这个节点的锁，但是不能获取到其他节点的锁，因此线程是安全的。这样就解决了主从一致性问题。

5.总结

1. 不可重入Redis分布式锁:
   原理:利用setnx的互斥性;利用ex避免死锁;释放锁时判断线程标示

   缺陷:不可重入、无法重试、锁超时失效

2. 可重入的Redis分布式锁:
   原理:利用hash结构，记录线程标示和重入次数;利用watchDog延续锁时间;利用信号量控制锁重试等待

   缺陷:redis宕机引起锁失效问题

3. Redisson的multiLock:
   原理:多个独立的Redis节点，必须在所有节点都获取重入锁，才算获取锁成功

   缺陷:运维成本高、实现复杂

六、秒杀业务优化–异步秒杀

1.思路

在原来的秒杀业务里，基于redis的读写操作和数据库的读写操作都是串行执行的。而对数据库读写操作本身比较耗时，在并发量较大的情况下，单位时间内执行的线程数会少，这会降低并发能力。
因此我们把是否有秒杀资格的业务交给Redis去做：
判断库存是否充足，不充足返回1，充足的情况下，根据set集合里是否有这样的userId，如果有说明重复，则返回1，不允许重复下单。如果没有则把userId添加到set集合里，并且返回0。
为了确保原子性，我们把这一块业务封装到lua脚本，根据该脚本的返回值，来决定是否拥有下单的资格。

2.改进的需求

1.添加优惠券

添加优惠券到数据库的同时保存到redis中去。

    @Override
    @Transactional
    public void addSeckillVoucher(Voucher voucher) {
        // 保存优惠券
        save(voucher);
        // 保存秒杀信息
        SeckillVoucher seckillVoucher = new SeckillVoucher();
        seckillVoucher.setVoucherId(voucher.getId());
        seckillVoucher.setStock(voucher.getStock());
        seckillVoucher.setBeginTime(voucher.getBeginTime());
        seckillVoucher.setEndTime(voucher.getEndTime());
        seckillVoucherService.save(seckillVoucher);
        // 保存到redis中
        stringRedisTemplate.opsForValue().set(SECKILL_STOCK_KEY + voucher.getId(), voucher.getStock().toString());
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

2.编写业务：秒杀库存和一人一单决定下单资格的lua脚本

local voucherId = ARGV[1]
local userId = ARGV[2]

local stockKey = 'seckill:stock' .. voucherId --库存key
local orderKey = 'seckill:order' .. voucherId --订单key：存放userId的set

if (tonumber(redis.call('get', stockKey)) <= 0) then
    return 1
end
-- orderKey的订单集合中是否有userId的成员：下单重复
if (redis.call('sismember',orderKey,userId)==1) then
    return 2
end
-- 以上情况不满足说明，可以下单了
-- 扣减库存incrby key -1
redis.call('incrby',stockKey,-1)
-- 将userId存入到set集合中 sadd orderKey userId
redis.call('sadd',orderKey,userId)
return 0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

3.执行脚本

    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }

    @Override
    public Result seckillVoucher(Long voucherId) {
        Long userId = UserHolder.getUser().getId();
        //1.执行lua脚本
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(), voucherId.toString(), userId.toString()
        );
        int r = result.intValue();
        //2.判断lua脚本的返回值是否为0，为0则代表有下单资格
        if (r != 0) {//没有下单资格
              return r == 1 ? Result.fail("库存不足！") :  Result.fail("不允许重复下单");
        }

        //TODO 3.把用户id，优惠券id，订单id放入阻塞队列
        long orderId = redisIdWorker.nextId("order:");

        //4.返回订单id
        return Result.ok(orderId);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

4.完成异步下单需求

4.1创建阻塞队列和异步线程

    private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

    @PostConstruct//当前类初始化完毕后执行
    private void init() {
        //提交线程任务
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandle());
    }

    private class VoucherOrderHandle implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    //获取阻塞队列中的订单信息
                    VoucherOrder voucherOrder = orderTasks.take();
                    //创建订单
                    handleVoucherOrder(voucherOrder);
                } catch (InterruptedException e) {
                    log.error("处理订单异常", e);
                }
            }
        }
    }
    private void handleVoucherOrder(VoucherOrder voucherOrder) {
        //1.获取用户
        Long userId = voucherOrder.getUserId();
        //2.创建锁对象
        RLock lock = redissonClient.getLock("lock:order:" + userId);
        //3.获取锁
        boolean isLock = lock.tryLock();
        //4.判断锁是否获取成功
        if (!isLock) {
            log.error("不允许重复下单！");
            return;
        }
        try {
            proxy.createVoucherOrder(voucherOrder);
        } finally {
            //释放锁
            lock.unlock();
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

4.2将订单信息添加到阻塞队列

    private IVoucherOrderService proxy;
    @Override
    public Result seckillVoucher(Long voucherId) {
        Long userId = UserHolder.getUser().getId();
        //1.执行lua脚本
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(), voucherId.toString(), userId.toString()
        );
        int r = result.intValue();
        //2.判断lua脚本的返回值是否为0，为0则代表有下单资格
        if (r != 0) {//没有下单资格
            return r == 1 ? Result.fail("库存不足！") : Result.fail("不允许重复下单");
        }

        //TODO 3.把用户id，优惠券id，订单id放入阻塞队列
        VoucherOrder voucherOrder = new VoucherOrder();
        long orderId = redisIdWorker.nextId("order");
        voucherOrder.setId(orderId);

        voucherOrder.setUserId(userId);

        voucherOrder.setVoucherId(voucherId);
        //3.1放入阻塞队列
        orderTasks.add(voucherOrder);
        //3.2获取代理对象
        proxy = (IVoucherOrderService) AopContext.currentProxy();
        //4.返回订单id
        return Result.ok(orderId);
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

4.3修改创建订单方法

    @Transactional
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        //6.充足，一人一单：根据优惠券id和用户id查询唯一订单
        Long userId = voucherOrder.getUserId();
        Long count = query().eq("voucher_id", voucherOrder.getVoucherId()).eq("user_id", userId).count();
        //6.1判断订单是否存在
        if (count > 0) {
            log.error("不可以重复下单！");
            return;
        }
        //7. 充足，扣减库存
        boolean success = seckillVoucherService.update()
                .setSql("stock= stock -1")
                .eq("voucher_id", voucherOrder.getVoucherId())
                .gt("stock", 0)
                .update();
        if (!success) {
            log.error("库存不足！");
            return;
        }
        //8. 创建订单:订单id，用户id，代金券id
        save(voucherOrder);

    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

3.思考

1. 秒杀业务的优化思路是什么？
   先利用redis完成库存余量、一人一单判断，完成抢单业务
   再将下单业务放入阻塞队列，利用独立线程异步下单

2. 基于阻塞队列的异步秒杀存在哪些问题？
   内存限制问题
   数据安全问题

4.消息队列优化

前面说了，使用JDK提供的阻塞队列存在两大问题：一是会占用JVM内存，二是不能保证数据安全问题：因为没有持久化，一旦JVM宕机了，数据就丢失了。
因此我们需要使用消息队列，这里我们介绍基于Redis模拟消息队列、消息中间件：RabbitMQ、kfk等

1.基于Redis模拟消息队列的三种方式

基于List的消息队列有哪些优缺点?

1. 优点:

• 利用Redis存储，不受限于JVM内存上限
• 基于Redis的持久化机制，数据安全性有保证.可以满足消息有序性

2. 缺点:

• 无法避免消息丢失(使用pop命令，是直接remove and get，但是没处理，其他消费者就拿不到)
• 只支持单消费者（发送的消息有一个消费者拿走了，其他的消费者就拿不到了）

基于PubSub的消息队列有哪些优缺点?

1. 优点:

• 采用发布订阅模型，支持多生产、多消费

2. 缺点:

• 不支持数据持久化（因为pubsub本身是用来做传递消息的媒介，不是一个数据结构）
• 无法避免消息丢失（发布完了一个消息，如果在一段时间内没有消费者订阅消息就会丢失）
• 消息堆积有上限，超出时数据丢失（如果消费者消费能力远低于生产者的生产能力时，就会堆积在消费者的缓存区里，而这个缓存区是有限的，超出了就丢失了）

STREAM类型消息队列的XREAD命令特点:

• 消息可回溯
• 一个消息可以被多个消费者读取
• 可以阻塞读取
• 有消息漏读的风险（当我们指定起始ID为$时，代表读取最新的消息，如果我们处理一条消息的过程中，又有超过1条以上的消息到达队列，则下次获取时也只能获取到最新的一条，会出现漏读消息的问题）

2.基于Stream的消息队列-消费者组

消费者组（(Consumer Group):将多个消费者划分到一个组中，监听同一个队列。具备下列特点:

1. 消息分流
   队列中的消息会分流给组内的不同消费者，而不是重复消费，从而加快消息处理的速度

2. 消息标示
   消费者组会维护一个标示，记录最后一个被处理的消息，哪怕消费者宕机重启，还会从标示之后读取消息。确保每一个消息都会被消费

3. 消息确认
   消费者获取消息后，消息处于pending状态，并存入一个pending-list。当处理完成后需要通过XACK来确认消息，标记消息为已处理，才会从pending-list移除。

创建消费者组:
XGROUP CREATE key groupName ID [MKSTREAM]

• key:队列名称
• groupName:消费者组名称
• ID:起始ID标示，$代表队列中最后一个消息，0则代表队列中第一个消息
• MKSTREAM:队列不存在时自动创建队列
  其它常见命令:

#删除指定的消费者组
XGROUP DESTORY key groupName
#给指定的消费者组添加消费者
XGROUP CREATECONSUMER key groupname consumername
#删除消费者组中的指定消费者
XGROUP DELCONSUMER key groupname consumername

1
2
3
4
5
6
7

从消费者组读取消息:
XREADGROUP GROUP group consumer[COUNT count][BLOCK milliseconds][NOACK] STREAMNSkey [key ...]ID [ID ...]

• group:消费组名称
• consumer:消费者名称，如果消费者不存在，会自动创建一个消费者
• count:本次查询的最大数量
• BLOCK milliseconds: 当没有消息时最长等待时间
• NOACK:无需手动ACK，获取到消息后自动确认
• STREAMS key:指定队列名称
• ID:获取消息的起始ID:
  • “>”:从下一个未消费的消息开始
  • 其它:根据指定id从pending-list中获取已消费但未确认的消息，例如0，是从 pending-list中的第一个消息开始

STREAM类型消息队列的XREADGROUP命令特点:

• 消息可回溯
• 可以多消费者争抢消息，加快消费速度
• 可以阻塞读取
• 没有消息漏读的风险
• 有消息确认机制，保证消息至少被消费一次

七、达人探店模块

1.查看博客功能

@Service
public class BlogServiceImpl extends ServiceImpl<BlogMapper, Blog> implements IBlogService {
    @Resource
    private IUserService userService;

    @Override
    public Result queryHotBlog(Integer current) {
        // 根据用户查询
        Page<Blog> page = query()
                .orderByDesc("liked")
                .page(new Page<>(current, SystemConstants.MAX_PAGE_SIZE));
        // 获取当前页数据
        List<Blog> records = page.getRecords();
        // 查询用户
        records.forEach(this::queryBlog);
        return Result.ok(records);
    }

    @Override
    public Result queryBlogById(Integer id) {
        //1.获取笔记
        Blog blog = getById(id);
        //2.判断笔记是否为空
        if (blog == null){
            return Result.fail("笔记为空！");
        }
        //3.查询返回Blog
        queryBlog(blog);
        return Result.ok(blog);
    }

    public void queryBlog(Blog blog){
        Long userId = blog.getUserId();
        User user = userService.getById(userId);
        blog.setName(user.getNickName());
        blog.setIcon(user.getIcon());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

2. 点赞功能

需求:

• 同一个用户只能点赞一次，再次点击则取消点赞
• 如果当前用户已经点赞，则点赞按钮高亮显示（前端已实现，判断字段Blog类的isLike属性)

实现步骤:

• 给Blog类中添加一个isLike字段，标示是否被当前用户点赞
• 修改点赞功能，利用Redis的set集合判断是否点赞过，未点赞过则点赞数+1，已点赞过则点赞数-1
• 修改根据id查询Blog的业务，判断当前登录用户是否点赞过，赋值给isLike字段
• 修改分页查询Blog业务，判断当前登录用户是否点赞过，赋值给isLike字段

	@Override
    public Result likeBlog(Long id) {
        //1.获取用户id
        Long userId = UserHolder.getUser().getId();
        //2.获取redis中的set集合中是否有重复的元素：是否已经被点赞过（isLike）
        Boolean isMember = stringRedisTemplate.opsForSet().isMember(BLOG_LIKED_KEY + id, userId.toString());
        if(BooleanUtil.isTrue(isMember)){
            //2.1如果已经被点赞了
            //2.2数据库的liked字段-1:update blog set liked = liked-1  where id = #{id}
            boolean isSuccess = update().setSql("liked = liked - 1").eq("id", id).update();
            //2.3将redis中的set集合中的用户id移除
            if(isSuccess){
                stringRedisTemplate.opsForSet().remove(BLOG_LIKED_KEY + id,userId.toString());
            }
        }else {
            //3.1如果没有点赞 数据库的liked字段+1
            boolean isSuccess = update().setSql("liked = liked + 1").eq("id", id).update();
            //3.3将点赞的用户添加到redis中的set集合中
            if(isSuccess){
                stringRedisTemplate.opsForSet().add(BLOG_LIKED_KEY + id,userId.toString());
            }
        }
        return Result.ok();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

写一个setBlogLiked方法用于设置是否点赞属性到实体类Blog里

    private void setBlogLiked(Blog blog) {
   		UserDTO user = UserHolder.getUser();
        //用户未登录状态，不获取userId，防止空指针异常
        if (user == null){
            return;
        }
        //1.获取用户id
        Long userId = blog.getUserId();
        //2.获取redis中的set集合中是否有重复的元素：是否已经被点赞过（isLike）
        Boolean isMember = stringRedisTemplate.opsForSet().isMember(BLOG_LIKED_KEY +blog.getId(), userId.toString());
        blog.setIsLike(BooleanUtil.isTrue(isMember));
    }

1
2
3
4
5
6
7
8
9
10
11
12
13

由于相比较之前博客，对于当前用户而已，多了是否已经点赞的信息，因此除了博主的用户信息要展示在博客上，还要把是否已经点赞信息展示。

    @Override
    public Result queryHotBlog(Integer current) {
        // 根据用户查询
        Page<Blog> page = query()
                .orderByDesc("liked")
                .page(new Page<>(current, SystemConstants.MAX_PAGE_SIZE));
        // 获取当前页数据
        List<Blog> records = page.getRecords();
        // 查询用户、是否被点赞
        records.forEach(blog -> {
            this.setBlogUser(blog);
            this.setBlogLiked(blog);
        });
        return Result.ok(records);
    }

    @Override
    public Result queryBlogById(Long id) {
        //1.获取笔记
        Blog blog = getById(id);
        //2.判断笔记是否为空
        if (blog == null) {
            return Result.fail("笔记为空！");
        }
        //3.Blog、以及是否被点赞
        setBlogUser(blog);
        setBlogLiked(blog);
        return Result.ok(blog);
    }
    public void setBlogUser(Blog blog) {
        Long userId = blog.getUserId();
        User user = userService.getById(userId);
        blog.setName(user.getNickName());
        blog.setIcon(user.getIcon());
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

3.排行榜功能

由于需要排序，原来的set是不可重复但无序的，因此这里我们需要把点赞业务中进行修改：没有点赞过的，点赞了就把用户id作为value，和当前时间戳作为score一起存入redis中去。后续我们通过score的值来确定排行榜的顺序，score值越前，用户排行越前。

    @Override
    public Result likeBlog(Long id) {
        //1.获取用户id
        Long userId = UserHolder.getUser().getId();
        //2.获取redis中的set集合中是否有重复的元素：是否已经被点赞过（isLike）
        Double score = stringRedisTemplate.opsForZSet().score(BLOG_LIKED_KEY + id, userId.toString());
        if(score != null){
            //2.1如果已经被点赞了
            //2.2数据库的liked字段-1:update blog set liked = liked-1  where id = #{id}
            boolean isSuccess = update().setSql("liked = liked - 1").eq("id", id).update();
            //2.3将redis中的sortedset集合中的键值对移除
            if(isSuccess){
                stringRedisTemplate.opsForZSet().remove(BLOG_LIKED_KEY + id,userId.toString());
            }
        }else {
            //3.1如果没有点赞 数据库的liked字段+1
            boolean isSuccess = update().setSql("liked = liked + 1").eq("id", id).update();
            //3.3将点赞的用户id以及当前时间戳添加到redis中的sortedset集合中
            if(isSuccess){
                stringRedisTemplate.opsForZSet().add(BLOG_LIKED_KEY + id,userId.toString(), System.currentTimeMillis());
            }
        }
        return Result.ok();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

同样的用于判断是否点赞的方法也要相应的修改：

    private void setBlogLiked(Blog blog) {
        UserDTO user = UserHolder.getUser();
        //用户未登录状态，不获取userId，防止空指针异常
        if (user == null){
            return;
        }
        //1.获取用户id
        Long userId = blog.getUserId();
        //2.获取redis中的set集合中是否有重复的元素：是否已经被点赞过（isLike）
        Double score = stringRedisTemplate.opsForZSet().score(BLOG_LIKED_KEY + blog.getId(), userId.toString());
        blog.setIsLike(BooleanUtil.isTrue(score != null));
    }


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

接下来就是排行榜的业务实现：

    @Override
    public Result likesBlog(Long id) {
        //1.从sortedset查询top5的点赞用户
        Set<String> top5 = stringRedisTemplate.opsForZSet().range(BLOG_LIKED_KEY + id, 0, 4);
        if (top5 == null||top5.isEmpty()){
            return Result.ok(Collections.emptyList());
        }
        //2.解析出其中的用户id
        List<Long> ids = top5.stream().map(Long::valueOf).collect(Collectors.toList());
        //3.根据用户id查询用户 select * from table_user where id1 = #{id1} or id2 = #{id2}
        String idStr = StrUtil.join(",",ids);
        //ORDER BY FIELD做顺序处理
        List<User> users = userService.query().in("id", ids).last("ORDER BY FIELD(id," + idStr + ")").list();
        //将user转成UserDTO
        List<UserDTO> userDTOS = BeanUtil.copyToList(users, UserDTO.class);
        //4.返回
        return Result.ok(userDTOS);
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

八、好友关注模块

页面加载进来时先判断是否被关注的布尔值，通过这个布尔值，来决定关注业务里面是否关注的标识。页面一加载就会走判断是否被关注的接口，而点击关注按钮，才会走关注接口。

1.好友关注和取关功能

    @Override
    public Result follow(Long followUserId,Boolean isFollow) {
        //1.获取当前用户id
        Long userId = UserHolder.getUser().getId();
        //2.1 判断是否被关注:isFollow为true
        if (isFollow) {
            //3.如果isFollow = true说明，就关注，保存到数据库
            Follow follow = new Follow();
            follow.setUserId(userId);
            follow.setFollowUserId(followUserId);
            save(follow);
        }else{
            //2.如果isFollow = false，从关注表中移除用户
            remove(new QueryWrapper<Follow>().eq("user_id",userId).eq("follow_user_id",followUserId));
        }
        return Result.ok();
    }

    @Override
    public Result isFollow(Long followUserId) {
        //1.获取用户id
        Long userId = UserHolder.getUser().getId();
        //2.查询数据库有没有关注的用户
        Follow follow = query().eq("user_id", userId).eq("follow_user_id", followUserId).one();
        return Result.ok(follow!=null);//不为空就返回true 的结果
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

2.点击头像显示个人主页

首先去数据库查询用户,返回一个DTO给前端即可

@GetMapping("/{id}")
public Result queryUserById(@PathVariable("id") Long userId){
	// 查询详情
	User user = userService.getById(userId);
	if (user == null) {
		return Result.ok();
	}
	UserDTO userDTO = BeanUtil.copyProperties(user, UserDTO.class);
	// 返回
	return Result.ok(userDTO);
}
1
2
3
4
5
6
7
8
9
10
11

显示笔记，分页查询：

@GetMapping("/of/user")
public Result queryBlogByUserId(
		@RequestParam(value = "current", defaultValue = "1") Integer current,
		@RequestParam("id") Long id) {
	// 根据用户查询
	Page<Blog> page = blogService.query()
			.eq("user_id", id).page(new Page<>(current, SystemConstants.MAX_PAGE_SIZE));
	// 获取当前页数据
	List<Blog> records = page.getRecords();
	return Result.ok(records);
}
1
2
3
4
5
6
7
8
9
10
11

3.共同关注功能

思路：通过redis中的set集合的SINTER key1 key2 命令求出两个集合的交集，然后通过求出的这些满足的用户id，去查询数据库，返回一个个完整的用户，并用List集合封装起来，最后转成List的形式传递给前端做渲染。

1. 修改关注业务的代码，在关注和取关后分别添加对redis进行操作的步骤：

    @Override
    public Result follow(Long followUserId,Boolean isFollow) {
        //1.获取当前用户id
        Long userId = UserHolder.getUser().getId();
        //2.1 判断是否被关注:isFollow为true
        if (isFollow) {
            //3.如果isFollow = true说明，就关注，保存到数据库
            Follow follow = new Follow();
            follow.setUserId(userId);
            follow.setFollowUserId(followUserId);
            boolean isSuccess = save(follow);
            if (isSuccess){
                //4.将被关注的用户保存到redis中的set集合中去：为求交集实现共同关注功能
                stringRedisTemplate.opsForSet().add(FOLLOW_COMMONS_KEY+userId,followUserId.toString());
            }

        }else{
            //2.2如果isFollow = false，从关注表中移除用户
            boolean isSuccess = remove(new QueryWrapper<Follow>().eq("user_id", userId).eq("follow_user_id", followUserId));
            if (isSuccess){
                //2.3将set中被关注的用户移除
                stringRedisTemplate.opsForSet().remove(FOLLOW_COMMONS_KEY+userId,followUserId.toString());
            }

        }
        return Result.ok();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

2. 共同关注的业务：

    @Override
    public Result commons(Long id) {//此id是点进主页的id
        //1.获取自己的id
        Long userId = UserHolder.getUser().getId();
        //2.根据用户id和被关注者的id，通过redis中set集合求交集
        Set<String> commonSet = stringRedisTemplate.opsForSet().intersect(FOLLOW_COMMONS_KEY + userId, FOLLOW_COMMONS_KEY + id);
        if (commonSet == null||commonSet.isEmpty()){
            return Result.ok();
        }
        //3.解析集合
        List<Long> ids = commonSet.stream().map(Long::valueOf).collect(Collectors.toList());
        //这个ids集合里只存放共同用户的id，我们最终需要展示的是整个用户，所以我们需要通过这个id查询数据库，然后返回
        //4，从数据库查询共同关注用户，并且封装成list集合
        List<User> userList = userService.listByIds(ids);
        //5.最后要返回一个UserDTO的List集合封装共同关注用户
        List<UserDTO> userDTOS = BeanUtil.copyToList(userList, UserDTO.class);
        return Result.ok(userDTOS);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

4.关注推送功能

首先博主每发一篇笔记，保存到数据库的同时，我们还要保存到粉丝的收件箱里：

    @Override
    public Result saveBlog(Blog blog) {
        // 1.获取登录用户
        UserDTO user = UserHolder.getUser();
        blog.setUserId(user.getId());
        // 2.保存探店博文
        boolean isSuccess = blogService.save(blog);
        if (!isSuccess){
            return Result.fail("笔记保存失败！");
        }
        //3.查询博主所有粉丝
        List<Follow> follows = followService.query().eq("follow_user_id", user.getId()).list();
        //4.推模式：将博客id保存到粉丝收件箱中
        for (Follow follow : follows) {
            //4.1获取每一个粉丝id
            Long userId = follow.getUserId();
            //4.2推送:将博客id和当前时间戳作为score保存到粉丝的收件箱：
            String key = "feed:"+userId;
            stringRedisTemplate.opsForZSet().add(key,blog.getId().toString(),System.currentTimeMillis());
        }
        //5. 返回id
        return Result.ok(blog.getId());
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

推送业务：

    @Override
    public Result followPush(Long max, Integer offset) {
        //1.获取当前用户
        Long userId = UserHolder.getUser().getId();
        //2.通过userId查询收件箱:zReverangeByScore
        String key =FEED_KEY+userId;
        //key:收件箱，min：每页查询的最小分数写死成0、max：查询的最大分数，offset：偏移量，第一次为0，往后如果有相同的score值的value，偏移量为相同的score值的value的个数，count：2，每页查询2条，写死。
        Set<ZSetOperations.TypedTuple<String>> typedTuples = stringRedisTemplate.opsForZSet().reverseRangeByScoreWithScores(key, 0, max, offset, 2);
        if (typedTuples==null||typedTuples.isEmpty()){
            return Result.ok();
        }
        //3.解析数据：blogId、minTime、offset
        long minTime = 0;
        int os = 1;
        ArrayList<Long> ids = new ArrayList<>(typedTuples.size());
        for (ZSetOperations.TypedTuple<String> typedTuple : typedTuples) {
            //3.1获取blogId：每层循环就是一个页:value就是每一个blogId，一页2个
            String idStr = typedTuple.getValue();
            ids.add(Long.valueOf(idStr));
            //3.2获取分数score
            long time = typedTuple.getScore().longValue();
            if (time == minTime){
                os++;
            }else {
                minTime=time;
                os = 1;
            }

        }
        //4.根据blogId查询blog
        String idStr = StrUtil.join(",", ids);
        List<Blog> blogs = blogService.query().in("id", ids).last("ORDER BY FIELD(id," + idStr + ")").list();

        for (Blog blog : blogs) {
            setBlogUser(blog);
            setBlogLiked(blog);
        }

        ResultScroll r = new ResultScroll();
        r.setBlogs(blogs);
        r.setOffset(os);//偏移量
        r.setMinTime(minTime);//最后一个元素的时间戳

        //5.封装成ResultScroll返回
        return Result.ok(r);
    }


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

但是这里会有bug，也就是取关博主后，还能在关注列表受到已经取关的博主的推送，所以我们在取关业务里，除了移除follows的键值对，还要移除feed的键值对。也就是移除粉丝的收件箱。对应的，当我们关注一个博主的时候，添加粉丝的收件箱。

    @Override
   public Result follow(Long followUserId, Boolean isFollow) {
       //1.获取当前用户id
       Long userId = UserHolder.getUser().getId();
       //2.1 判断是否被关注:isFollow为true
       if (isFollow) {
           //3.如果isFollow = true说明，就关注，保存到数据库
           Follow follow = new Follow();
           follow.setUserId(userId);
           follow.setFollowUserId(followUserId);
           boolean isSuccess = save(follow);
           if (isSuccess) {
               //4.将被关注的用户保存到redis中的set集合中去：为求交集实现共同关注功能
               stringRedisTemplate.opsForSet().add(FOLLOW_COMMONS_KEY + userId, followUserId.toString());
               //4.1关注了后，将博主笔记添加到粉丝的收件箱
               List<Blog> blogs = blogService.query().eq("user_id", followUserId).list();
               for (Blog blog : blogs) {
                   stringRedisTemplate.opsForZSet().add(FEED_KEY + userId,blog.getId().toString(),System.currentTimeMillis());
               }
           }

       } else {
           //2.2如果isFollow = false，从关注表中移除用户
           boolean isSuccess = remove(new QueryWrapper<Follow>().eq("user_id", userId).eq("follow_user_id", followUserId));
           if (isSuccess) {
               //2.3将set中被关注的用户移除
               stringRedisTemplate.opsForSet().remove(FOLLOW_COMMONS_KEY + userId, followUserId.toString());
               //2.4移除粉丝的收件箱
               //2.4.1获取被关注用户的blogs
               List<Blog> blogs = blogService.query().eq("user_id", followUserId).list();
               for (Blog blog : blogs) {
                   //2.4.2将每一个当前的blogId移除
                   stringRedisTemplate.opsForZSet().remove(FEED_KEY + userId,blog.getId().toString());
               }
           }

       }
       return Result.ok();
   }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

九、附近商铺

1.GEO

GEO就是Geolocation的简写形式，代表地理坐标。Redis在3.2版本中加入了对GEO的支持，允许存储地理坐标信息，帮助我们根据经纬度来检索数据。常见的命令有:

1. GEOADD:添加一个地理空间信息，包含:经度（longitude)、纬度(latitude)、值( member)
2. GEODIST:计算指定的两个点之间的距离并返回
3. GEOHASH:将指定member的坐标转为hash字符串形式并返回
4. GEOPOS:返回指定member的坐标
5. GEORADIUS:指定圆心、半径，找到该圆内包含的所有member，并按照与圆心之间的距离排序后返回。6.2以后已废弃
6. GEOSEARCH:在指定范围内搜索member，并按照与指定点之间的距离排序后返回。范围可以是圆形或矩形。6.2.新功能
7. GEOSEARCHSTORE:与GEOSEARCH功能一致，不过可以把结果存储到一个指定的key。6.2.新功能

2.导入商铺数据到GEO中

    @Test
    void loadShopData(){
        //1.查询店铺信息 select * from shop
        List<Shop> list = shopService.list();
        //2.店铺信息按typeId分组，typeId一致的放到一个集合里
        Map<Long,List<Shop>> map = list.stream().collect(Collectors.groupingBy(Shop::getTypeId));
        //3.分批写入Redis
        for (Map.Entry<Long, List<Shop>> entry : map.entrySet()) {
            //3.1每一个entry就是一个根据shopId分好的组
            Long typeId = entry.getKey();
            //3.2获取同类型的店铺集合
            List<Shop> shopList = entry.getValue();
            String key ="shop:geo:"+typeId;
            List<RedisGeoCommands.GeoLocation<String>> locations =new ArrayList<>(shopList.size());
            //3.3写入redis GEOADD key x y member , shopId作为member
            for (Shop shop : shopList) {
                //这样写，重复写入redis操作开销太大，最好使用批量写入：GeoLocation的可迭代对象
                //stringRedisTemplate.opsForGeo().add(key,new Point(shop.getX(), shop.getY()),shop.getId().toString())
                locations.add(new RedisGeoCommands.GeoLocation<>(shop.getId().toString(),new Point(shop.getX(), shop.getY())));
            }
            stringRedisTemplate.opsForGeo().add(key,locations);
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

3.附近商铺显示距离并排序业务：

这两个依赖必须使用稳定版的，否则会报错

       <dependency>
            <groupId>org.springframework.datagroupId>
            <artifactId>spring-data-redisartifactId>
            <version>2.6.2version>
        dependency>
        <dependency>
            <groupId>io.lettucegroupId>
            <artifactId>lettuce-coreartifactId>
            <version>6.1.6.RELEASEversion>
        dependency>
1
2
3
4
5
6
7
8
9
10

    public Result queryShopByType(Integer typeId, Integer current, Double x, Double y) {
        // 1.判断是否需要根据坐标查询
        if (x == null || y == null) {
            // 不需要坐标查询，按数据库查询
            Page<Shop> page = query()
                    .eq("type_id", typeId)
                    .page(new Page<>(current, SystemConstants.DEFAULT_PAGE_SIZE));
            // 返回数据
            return Result.ok(page.getRecords());
        }

        // 2.计算分页参数
        int from = (current - 1) * SystemConstants.DEFAULT_PAGE_SIZE;
        int end = current * SystemConstants.DEFAULT_PAGE_SIZE;

        // 3.查询redis、按照距离排序、分页。结果：shopId、distance
        String key = SHOP_GEO_KEY + typeId;
        GeoResults<RedisGeoCommands.GeoLocation<String>> results = stringRedisTemplate.opsForGeo() // GEOSEARCH key BYLONLAT x y BYRADIUS 10 WITHDISTANCE
                .search(
                        key,
                        GeoReference.fromCoordinate(x, y),
                        new Distance(5000),
                        RedisGeoCommands.GeoSearchCommandArgs.newGeoSearchArgs().includeDistance().limit(end)
                );
        // 4.解析出id
        if (results == null) {
            return Result.ok(Collections.emptyList());
        }
        List<GeoResult<RedisGeoCommands.GeoLocation<String>>> list = results.getContent();
        if (list.size() <= from) {
            // 没有下一页了，结束
            return Result.ok(Collections.emptyList());
        }
        // 4.1.截取 from ~ end的部分
        List<Long> ids = new ArrayList<>(list.size());
        Map<String, Distance> distanceMap = new HashMap<>(list.size());
        list.stream().skip(from).forEach(result -> {
            // 4.2.获取店铺id
            String shopIdStr = result.getContent().getName();
            ids.add(Long.valueOf(shopIdStr));
            // 4.3.获取距离
            Distance distance = result.getDistance();
            distanceMap.put(shopIdStr, distance);
        });
        // 5.根据id查询Shop
        String idStr = StrUtil.join(",", ids);
        List<Shop> shops = query().in("id", ids).last("ORDER BY FIELD(id," + idStr + ")").list();
        for (Shop shop : shops) {
            shop.setDistance(distanceMap.get(shop.getId().toString()).getValue());
        }
        // 6.返回
        return Result.ok(shops);
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

十、用户签到

1.BitMap

Redis中是利用string类型数据结构实现BitMap，因此最大上限是512M，转换为bit则是2^32个bit位。BitMap的操作命令有:

1. SETBIT:向指定位置（ offset）存入一个0或1
2. GETBIT:获取指定位置（ offset)的bit值
3. BITCOUNT:统计BitMap中值为1的bit位的数量
4. BITFIELD:操作（查询、修改、自增）BitMap中bit数组中的指定位置（offset)的值
5. BITFIELD_RO:获取BitMap中bit数组，并以十进制形式返回
6. BITOP:将多个BitMap的结果做位运算（与、或、异或)
7. BITPOS:查找bit数组中指定范围内第一个0或1出现的位置

2.签到业务

    @Override
    public Result sign() {
        //1.获取用户信息
        Long userId = UserHolder.getUser().getId();
        //2.获取当天日期
        LocalDateTime now = LocalDateTime.now();
        //3.拼接key
        String prefix = now.format(DateTimeFormatter.ofPattern(":yyyyMM"));
        String key = USER_SIGN_KEY + userId + prefix;
        //4.判断当天是本月的第几天
        int dayOfMonth = now.getDayOfMonth();
        //5.存入redis中 setBit key offset 1
        stringRedisTemplate.opsForValue().setBit(key, dayOfMonth - 1, true);
        return Result.ok();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

3.统计当天之前连续签到次数

    @Override
    public Result signCount() {
        //1.获取用户信息
        Long userId = UserHolder.getUser().getId();
        //2.获取当天日期
        LocalDateTime now = LocalDateTime.now();
        //3.拼接key
        String prefix = now.format(DateTimeFormatter.ofPattern(":yyyyMM"));
        String key = USER_SIGN_KEY + userId + prefix;
        //4.判断当天是本月的第几天
        int dayOfMonth = now.getDayOfMonth();
        //5.
        List<Long> result = stringRedisTemplate.opsForValue()
                .bitField(key, BitFieldSubCommands.create()
                        .get(BitFieldSubCommands.BitFieldType.unsigned(dayOfMonth)).valueAt(0));
        if (result == null || result.isEmpty()) {
            return Result.ok(0);
        }
        Long num = result.get(0);
        if (num == 0) {
            return Result.ok(0);
        }
        int count = 0;
        while (true) {
            //判断 最后一位与1进行与运算的值 就是当前这个位，也就是第几天 是否为0
            if ((num & 1) == 0) {
                //如果为0，说明未签到
                break;
            } else {
                //如果为1，说明签到，计数器加1
                count++;
            }
            //将位，向右移一位
            num=num>>1;
        }
        return Result.ok(count);
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38