Redis项目仿大众点评

项目初始化

创建数据库以及数据库SQL：

tb_user：用户表
tb_user_info：用户详情表
tb_shop：商户信息表
tb_shop_type：商户类型表
tb_blog：用户日记表（达人探店日记）
tb_follow：用户关注表
tb_voucher：优惠券表
tb_voucher_order：优惠券的订单表

项目的架构：

项目的 pom 文件：

    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <!-- redis的依赖以及对应的连接池 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-pool2</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <scope>runtime</scope>
            <version>5.1.47</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
            <version>3.4.3</version>
        </dependency>
        <!-- hutool，包含各种各样的工具类，比如字符串处理、json工具类，日期工具类等 -->
        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.7.17</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>

</project>

项目的 appliaction.yml 配置信息：

server:
  port: 8081
spring:
  application:
    name: hmdp
  datasource:
    driver-class-name: com.mysql.jdbc.Driver
    url: jdbc:mysql://127.0.0.1:3306/hmdp?useSSL=false&serverTimezone=UTC
    username: root
    password: 123456
  redis:
    host: 192.168.88.129
    port: 6379
    password: leadnews
    lettuce:
      pool:
        max-active: 10
        max-idle: 10
        min-idle: 1
        time-between-eviction-runs: 10s
  jackson:
    default-property-inclusion: non_null # JSON处理时忽略非空字段
mybatis-plus:
  type-aliases-package: com.hmdp.entity # 别名扫描包
logging:
  level:
    com.hmdp: debug

按住 alt + 8 找到服务，启动项目

启动项目后，在浏览器访问：http://localhost:8081/shop-type/list ，如果可以看到数据则证明运行没有问题

再启动nginx，完成前后端联调

短信登录

基于 session 实现短信登录

基于session实现登录

发送验证码：

/**
 * 发送手机验证码
 *
 * @param phone
 * @param session
 * @return
 */
@Override
public Result sendCode(String phone, HttpSession session) {
    // 1. 校验手机号
    boolean phoneInvalid = RegexUtils.isPhoneInvalid(phone);
    if (phoneInvalid) {
        
        // 2. 如果不符合，返回错误信息
        return Result.fail("手机号格式错误");
    }
    
    // 3. 如果符合，生成验证码
    String code = RandomUtil.randomNumbers(6);
    
    // 4. 保存验证码到session
    session.setAttribute("code", code);
    
    // 5. 发送验证码
    // 使用阿里云第三方平台实现
    log.debug("发送短信验证码成功, 验证码：{}", code);
    
    // 返回ok
    return Result.ok();
}

/**
 * 登录功能
 *
 * @param loginForm
 * @param session
 * @return
 */
@Autowired
private UserMapper userMapper;
@Override
public Result login(LoginFormDTO loginForm, HttpSession session) {
    
    // 1. 校验手机号
    boolean phoneInvalid = RegexUtils.isPhoneInvalid(loginForm.getPhone());
    if (phoneInvalid) {
        
        // 1.2 如果不符合，返回错误信息
        return Result.fail("手机号格式错误");
    }
    
    // 2. 校验验证码
    String cacheCode = (String) session.getAttribute("code");
    String code = loginForm.getCode();
    if (cacheCode == null || !code.equals(cacheCode)) {
        
        // 3. 结果不一致，报错
        return Result.fail("验证码错误");
    }
    
    // 4. 结果一致，根据手机号查询用户
    User user = userMapper.selectOne(Wrappers
            .<User>lambdaQuery()
            .eq(User::getPhone, loginForm.getPhone()));
    
    // 5. 判断用户是否存在
    if (user == null) {
        
        // 6. 不存在，创建新用户并保存
        user = createUserWithPhone(loginForm.getPhone());
    }
    
    // 7. 保存用户信息到session中
    session.setAttribute("user", user);
    return Result.ok();
}

/**
 * 创建新用户
 *
 * @param phone
 */
private User createUserWithPhone(String phone) {
    
    // 1. 创建用户
    User user = new User();
    user.setPhone(phone);
    user.setNickName(SystemConstants.USER_NICK_NAME_PREFIX + RandomUtil.randomString(10));
    
    // 2. 保存用户
    save(user);
    
    // 3. 返回
    return user;
}

实现登录拦截功能：

在 threadLocal 中，无论是它的 put 方法和它的 get 方法，都是先从获得当前用户的线程，然后从线程中取出线程的成员变量 map，只要线程不一样，map 就不一样，所以可以通过这种方式来做到线程隔离

创建拦截器

public class LoginInterceptor implements HandlerInterceptor {
    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {

        // 1. 获取session
        HttpSession session = request.getSession();

        // 2. 获取session中的用户
        User user = (User) session.getAttribute("user");

        // 3. 判断用户是否存在
        if (user == null) {

            // 4. 不存在直接拦截
            response.setStatus(401);
            return false;
        }

        UserDTO userDTO = new UserDTO();
        BeanUtils.copyProperties(user, userDTO);

        // 5. 存在，则把用户信息保存到ThreadLocal
        UserHolder.saveUser(userDTO);

        // 6. 放行
        return true;
    }

    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
        // 移除用户，避免内存泄露
        UserHolder.removeUser();
    }
}

添加拦截器到mvc中，并排除不需要拦截的路径

@Configuration
public class MvcConfig implements WebMvcConfigurer {

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(new LoginInterceptor())
                .excludePathPatterns(
                        "/user/code",
                        "/user/login",
                        "/blog/hot",
                        "/shop/**",
                        "/shop-type/**",
                        "/voucher/**",
                        "/upload/**",
                        ".user/me");
    }
}

集群的 session 共享问题

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

session 的替代方案 应该满足：

数据共享
内存存储
key、value 结构

集群的 session 共享问题

基于 Redis 实现共享 session 登录

在 Redis 中保存单个对象时，有两种结构：

String 类型：保存登录的用户信息，可以使用 String 结构，以 JSON 字符串来保存，比较直观
Hash 类型：Hash 结构可以将对象中的每个字段独立存储，可以针对 单个字段做 CRUD，并且 内存占用更少

整体访问流程：

访问流程

当注册完成后，用户去登录会去校验用户提交的手机号和验证码，是否一致，如果一致，则根据手机号查询用户信息，不存在则新建，最后将用户数据保存到 redis，并且生成token 作为 redis 的 key，当校验用户是否登录时，会去携带着token 进行访问，从 redis 中取出 token 对应的 value，判断是否存在这个数据，如果没有则拦截，如果存在则将其保存到 threadLocal 中，并且放行

UserServiceImpl 代码：

@Service
@Slf4j
public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements IUserService {

    /**
     * 发送手机验证码
     *
     * @param phone
     * @param session
     * @return
     */

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public Result sendCode(String phone, HttpSession session) {


        // 1. 校验手机号
        boolean phoneInvalid = RegexUtils.isPhoneInvalid(phone);
        if (!phoneInvalid) {

            // 2. 如果不符合，返回错误信息
            return Result.fail("手机号格式错误");
        }

        // 3. 如果符合，生成验证码
        String code = RandomUtil.randomNumbers(6);

        // 4. 保存验证码到redis中
        // session.setAttribute("code", code);
        // 执行redis中的hash操作，key为手机号，value为验证码
        // set key value ex 120
        stringRedisTemplate.opsForValue().set("login:code:" + phone, code, 2, TimeUnit.MINUTES);

        // 5. 发送验证码
        // 使用阿里云第三方平台实现
        log.debug("发送短信验证码成功, 验证码：{}", code);

        // 返回ok
        return Result.ok();
    }

    /**
     * 登录功能
     *
     * @param loginForm
     * @param session
     * @return
     */

    @Autowired
    private UserMapper userMapper;

    @Override
    public Result login(LoginFormDTO loginForm, HttpSession session) {

        // 1. 校验手机号
        boolean phoneInvalid = RegexUtils.isPhoneInvalid(loginForm.getPhone());
        if (!phoneInvalid) {

            // 2. 如果不符合，返回错误信息
            return Result.fail("手机号格式错误");
        }

        // 2. 从Redis中获取校验验证码
        // String cacheCode = (String) session.getAttribute("code");
        String cacheCode = stringRedisTemplate.opsForValue().get("login:code:" + loginForm.getPhone());
        String code = loginForm.getCode();

        if (cacheCode == null || !code.equals(cacheCode)) {
            // 3. 结果不一致，报错
            return Result.fail("验证码错误");
        }

        // 4. 结果一致，根据手机号查询用户
        User user = userMapper.selectOne(Wrappers
                .<User>lambdaQuery()
                .eq(User::getPhone, loginForm.getPhone()));

        // 5. 判断用户是否存在
        if (user == null) {
            // 6. 不存在，创建新用户并保存
            user = createUserWithPhone(loginForm.getPhone());
        }

        // 7. 保存用户信息到redis中
        // 7.1 随机生成token，作为登录令牌
        String token = UUID.randomUUID().toString(true); // 这种 toString 不带 _

        // 7.2 将User对象转为HashMap存储
        UserDTO userDTO = BeanUtil.copyProperties(user, UserDTO.class);

        // 7.3 存储
        String tokenKey = "login:token:" + token;
        Map<String, Object> map = BeanUtil.beanToMap(userDTO, new HashMap<>()
                , CopyOptions.create()
                        .setIgnoreNullValue(true)
                        .setFieldValueEditor((filedName, filedValue) -> filedValue.toString()));
        stringRedisTemplate.opsForHash().putAll(tokenKey, map);

        // 7.4 设置token的有效期
        stringRedisTemplate.expire(tokenKey, 30, TimeUnit.MINUTES);

        // 8. 返回token
        return Result.ok(token);
    }

    /**
     * 创建新用户
     *
     * @param phone
     */
    private User createUserWithPhone(String phone) {
        // 1. 创建用户
        User user = new User();
        user.setPhone(phone);
        user.setNickName(SystemConstants.USER_NICK_NAME_PREFIX + RandomUtil.randomString(10));

        // 2. 保存用户
        save(user);

        // 3. 返回
        return user;
    }
}

拦截器部分代码：

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
        // HttpSession session = request.getSession();
        String token = request.getHeader("authorization");
        if (StrUtil.isBlank(token)) {
            // 不存在，直接拦截
            response.setStatus(401);
            return false;
        }

        // 2. 基于token获取redis中的用户
        // User user = (User) session.getAttribute("user");
        Map<Object, Object> map = stringRedisTemplate.opsForHash()
                .entries(RedisConstants.LOGIN_USER_KEY + "token");

        // 3. 判断用户是否存在
        if (map.isEmpty()) {

            // 4. 不存在直接拦截
            response.setStatus(401);
            return false;
        }

        // 5. 将查询到的Hash数据转换为UserDTO对象
        UserDTO userDTO = BeanUtil.fillBeanWithMap(map, new UserDTO(), false);

        // 6. 存在，则把用户信息保存到ThreadLocal
        UserHolder.saveUser(userDTO);

        // 7. 刷新token有效期
        stringRedisTemplate.expire(RedisConstants.LOGIN_USER_KEY + "token"
                , 30, TimeUnit.MINUTES);

        // 8. 放行
        return true;
    }

    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
        // 移除用户，避免内存泄露
        UserHolder.removeUser();
    }
}

@Configuration
public class MvcConfig implements WebMvcConfigurer {

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(new LoginInterceptor(stringRedisTemplate))
                .excludePathPatterns(
                        "/user/code",
                        "/user/login",
                        "/blog/hot",
                        "/shop/**",
                        "/shop-type/**",
                        "/voucher/**",
                        "/upload/**",
                        ".user/me");
    }
}

总结：

Redis 代替 session 需要考虑的问题：

选择合适的数据结构
选择合适的 key
选择合适的存储粒度

登录拦截器的优化

原始的拦截器

在这个方案中，确实可以使用对应路径的拦截，同时刷新登录token令牌的存活时间，但是现在这个拦截器只是拦截需要被拦截的路径，假设当前用户访问了一些不需要拦截的路径，那么这个拦截器就不会生效，所以此时令牌刷新的动作实际上就不会执行，所以这个方案是存在问题的

优化后的拦截器

第一个拦截器有了 threadLocal 的数据，第二个拦截器只需要判断拦截器中的user对象是否存在即可，完成整体刷新功能

流程：

创建一个新的拦截器 RefreshTokenInterceptor

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");
        if (StrUtil.isBlank(token)) {
            return true;
        }

        // 2. 基于token获取redis中的用户
        // User user = (User) session.getAttribute("user");
        Map<Object, Object> map = stringRedisTemplate.opsForHash()
                .entries(RedisConstants.LOGIN_USER_KEY + "token");

        // 3. 判断用户是否存在
        if (map.isEmpty()) {

            return true;
        }

        // 4. 将查询到的Hash数据转换为UserDTO对象
        UserDTO userDTO = BeanUtil.fillBeanWithMap(map, new UserDTO(), false);

        // 5. 存在，则把用户信息保存到ThreadLocal
        UserHolder.saveUser(userDTO);

        // 6. 刷新token有效期
        stringRedisTemplate.expire(RedisConstants.LOGIN_USER_KEY + "token"
                , 30, TimeUnit.MINUTES);

        // 7. 放行
        return true;
    }

    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
        // 移除用户，避免内存泄露
        UserHolder.removeUser();
    }
}

修改原来的拦截器 LoginInterceptor

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);
            // 拦截
            return false;
        }
        return true;
    }

    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
        // 移除用户，避免内存泄露
        UserHolder.removeUser();
    }
}

注册拦截器

@Configuration
public class MvcConfig implements WebMvcConfigurer {

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        
        // 登录拦截器
        registry.addInterceptor(new LoginInterceptor())
                .excludePathPatterns(
                        "/user/code",
                        "/user/login",
                        "/blog/hot",
                        "/shop/**",
                        "/shop-type/**",
                        "/voucher/**",
                        "/upload/**",
                        ".user/me"
                ).order(1); // order值越大优先级越低，越小优先级越高

        // token刷新拦截器
        registry.addInterceptor(new RefreshTokenInterceptor(stringRedisTemplate))
                .addPathPatterns("/**").order(0);
    }
}

商户查询缓存

缓存的概述

浏览器缓存：主要是存在于浏览器端的缓存

应用层缓存：可以分为 tomcat本地缓存，比如之前提到的 map，或者是使用 redis 作为缓存

数据库缓存：在数据库中有一片空间是 buffer pool，增改查数据都会先加载到 mysql 的缓存中

CPU缓存：当代计算机最大的问题是 cpu 性能提升了，但内存读写速度没有跟上，所以为了适应当下的情况，增加了cpu的L1、L2、L3级的缓存

缓存(Cache)，就是数据交换的缓冲区，俗称的缓存就是缓冲区内的数据，一般从数据库中获取，存储于本地代码，例如：

// 例1：本地用于高并发
Static final ConcurrentHashMap<K,V> map = new ConcurrentHashMap<>(); 

// 例2：用于redis等缓存
static final Cache<K,V> USER_CACHE = CacheBuilder.newBuilder().build(); 

// 例3：本地缓存
Static final Map<K,V> map =  new HashMap();

对于例3：由于其被 Static 修饰，所以随着类的加载而被加载到 内存之中，作为本地缓存，由于其又被 final 修饰，所以其引用（例3:map）和对象（例3:new HashMap()）之间的关系是固定的，不能改变，因此不用担心赋值 (=) 导致缓存失效

缓存的作用及成本

添加 redis 缓存

未添加缓存作用的模型

添加缓存作用的模型

对应的实现代码：

@Service
public class ShopServiceImpl extends ServiceImpl<ShopMapper, Shop> implements IShopService {

    /**
     * 根据id查询商铺信息
     *
     * @param id
     * @return
     */

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public Result queryById(Long id) {

        // 1. 从redis里查询商铺缓存
        // String key = "cache:shop:" + id;
        String key = RedisConstants.CACHE_SHOP_KEY + id;
        String shopJson = stringRedisTemplate.opsForValue().get(key);

        // 2. 判断缓存是否存在
        if (StrUtil.isNotBlank(shopJson)) {

            // 3. 存在，直接返回
            Shop shop = JSONUtil.toBean(shopJson, Shop.class);
            return Result.ok(shop);
        }

        // 4. 不存在，根据id查询数据库
        Shop shop = getById(id);

        // 5. 不存在，直接返回错误
        if (shop == null) {
            return Result.fail("店铺不存在");
        }

        // 6. 存在，将商铺数据写入redis
        stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop));

        // 7. 返回
        return Result.ok(shop);
    }
}

缓存更新策略

缓存更新是 redis 为了节约内存而设计出来的一个东西，主要是因为内存数据宝贵，当向 redis 插入太多数据，此时就可能会导致缓存中的数据过多，所以 redis 会对部分数据进行更新

缓存更新策略

内存淘汰：redis 自动进行，当 redis 内存达到自定义设定的 max-memery 的时候，会自动触发淘汰机制，淘汰掉一些不重要的数据（可以自定义设置策略方式）
超时剔除：当自定义给 redis 设置了过期时间TTL 之后，redis 会将超时的数据进行删除，方便之后继续使用缓存
主动更新：可以手动调用方法把缓存删掉，通常用于解决缓存和数据库不一致问题

数据库缓存不一致解决方案：

Cache Aside Pattern 人工编码方式：缓存调用者在更新完数据库后再去更新缓存，也称之为双写方案（常用方案）
Read/Write Through Pattern : 由系统本身完成，数据库与缓存的问题交由系统本身去处理
Write Behind Caching Pattern ：调用者只操作缓存，其他线程去异步处理数据库，实现最终一致

采用 Cache Aside Pattern 人工编码方式，需要考虑的问题：

删除缓存还是更新缓存？
- 更新缓存：每次更新数据库都更新缓存，无效写操作较多（×）
- 删除缓存：更新数据库时让缓存失效，查询时再更新缓存（√）
如何保证缓存与数据库的操作的同时成功或失败？
- 单体系统，将缓存与数据库操作放在一个事务
- 分布式系统，利用 TCC 等分布式事务方案
先操作缓存还是先操作数据库？
- 先删除缓存，再操作数据库（出现概率高，因为数据库的写入操作时间长，而 redis 缓存写入操作时间短）
- 先操作数据库，再删除缓存（出现概率低）

Cache Aside Pattern

总结：

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

低一致性需求：使用 Redis 自带的内存淘汰机制
高一致性需求：主动更新，并以超时剔除作为兜底方案
- 读操作：
  - 缓存命中则直接返回
  - 缓存未命中则查询数据库，并写入缓存，设定超时时间
- 写操作：
  - 先写数据库，然后再删除缓存
  - 要确保数据库与缓存操作的原子性

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

根据 id 查询店铺时，如果缓存未命中，则查询数据库，将数据库结果写入缓存，并设置超时时间

@Resource
private StringRedisTemplate stringRedisTemplate;
@Override
public Result queryById(Long id) {
    // 1. 从redis里查询商铺缓存
    // String key = "cache:shop:" + id;
    String key = RedisConstants.CACHE_SHOP_KEY + id;
    String shopJson = stringRedisTemplate.opsForValue().get(key);
    
    // 2. 判断缓存是否存在
    if (StrUtil.isNotBlank(shopJson)) {
        
        // 3. 存在，直接返回
        Shop shop = JSONUtil.toBean(shopJson, Shop.class);
        return Result.ok(shop);
    }
    
    // 4. 不存在，根据id查询数据库
    Shop shop = getById(id);
    
    // 5. 不存在，直接返回错误
    if (shop == null) {
        return Result.fail("店铺不存在");
    }
    
    // 6. 存在，将商铺数据写入redis
    // 新增：设置过期时间
    stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop), 30L, TimeUnit.MINUTES);
    
    // 7. 返回
    return Result.ok(shop);
}

根据 id 修改店铺时，先修改数据库，再删除缓存

@Override
@Transactional
public Result update(Shop shop) {
    Long id = shop.getId();
    if (id == null) {
        return Result.fail("店铺id不能为空");
    }
    // 1. 更新数据库
    updateById(shop);
    
    // 2. 删除缓存
    stringRedisTemplate.delete(RedisConstants.CACHE_SHOP_KEY + id);
    return Result.ok();
}

缓存穿透

缓存穿透：缓存穿透是指客户端请求的数据在缓存中和数据库中都不存在，这样缓存永远不会生效，这些请求都会达到数据库

常见的解决方案有两种：

缓存空对象
- 优点：实现简单，维护方便
- 缺点：
  - 额外的内存消耗
  - 可能造成短期的不一致（这种缺点只能手动添加数据）
布隆过滤
- 优点：内存占用较少，没有多余key
- 缺点：
  - 实现复杂
  - 存在误判可能，有一定的缓存穿透风险

左边：缓存空对象，右边：布隆过滤

修改shopService的代码：

采用缓存空对象的解决流程

在原来的逻辑中，如果发现这个数据在 mysql 中不存在，直接就返回 404 了，这样是会存在缓存穿透问题的
现在的逻辑中：如果这个数据不存在，不会返回404 ，还是会把这个数据写入到 Redis 中，并且将 value 设置为空，当再次发起查询时，如果发现命中之后，判断这个 value 是否是 null，如果是 null，则是之前写入的数据，证明是缓存穿透数据，如果不是，则直接返回数据

@Resource
private StringRedisTemplate stringRedisTemplate;
@Override
public Result queryById(Long id) {
    
    // 1. 从redis里查询商铺缓存
    // String key = "cache:shop:" + id;
    String key = RedisConstants.CACHE_SHOP_KEY + id;
    String shopJson = stringRedisTemplate.opsForValue().get(key);
    
    // 2. 判断缓存是否存在
    if (StrUtil.isNotBlank(shopJson)) {
        
        // 3. 存在，直接返回
        Shop shop = JSONUtil.toBean(shopJson, Shop.class);
        return Result.ok(shop);
    }
    
    // 4. 不存在，根据id查询数据库
    Shop shop = getById(id);
    
    // 5. 不存在，直接返回错误
    if (shop == null) {
        
        // TODO：新增
        // 将空值写入redis
        stringRedisTemplate.opsForValue().set(key, "", 2L, TimeUnit.MINUTES);
        // 返回错误信息
        return Result.fail("店铺不存在");
    }
    
    // TODO：新增
    // 判断命中的是否为空值
    if (shopJson != null) {
        // 返回错误信息
        return Result.fail("店铺信息不存在！");
    }
    
    // 6. 存在，将商铺数据写入redis
    // 新增：设置过期时间
    stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop), 30L, TimeUnit.MINUTES);
    
    // 7. 返回
    return Result.ok(shop);
}

总结：

缓存穿透产生的原因是：

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

缓存穿透的解决方案有：

缓存 null 值
布隆过滤
增强 id 的复杂度，避免被猜测 id 规律
- 比如说 id 为 0，这种id不可能存在
做好数据的基础格式校验
加强用户权限校验
做好热点参数的限流
- 比如说对空值或者热点数据做限流

缓存雪崩

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

解决方案：

给不同的 Key 的 TTL 添加随机值
利用 Redis集群提高服务的可用性
给缓存业务添加降级限流策略
给业务添加多级缓存
- 比如：浏览器添加缓存，Nginx添加缓存，JVM内部本地缓存等等

缓存击穿（热点 key）

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

缓存击穿图示

常见的解决方案有两种：

互斥锁
- 假设现在线程1 过来访问，它去查询缓存没有命中，但是此时它获得到了锁的资源，那么线程1 就会自己去执行逻辑，假设现在线程2 过来，线程2 在执行过程中，并没有获得到锁，那么线程2 就可以进行到休眠，直到线程1 把锁释放后，线程2 获得到锁，然后再来执行逻辑，此时就能够从缓存中拿到数据了
逻辑过期
- 之所以会出现这个缓存击穿问题，主要原因是在于对 key 设置了过期时间，不设置过期时间，其实就不会有缓存击穿的问题，但是不设置过期时间，这样数据不就一直占用内存了，但是可以采用逻辑过期方案，就是：把过期时间设置在 redis 的 value 中，注意：这个过期时间并不会直接作用于redis，而是后续通过逻辑去处理
- 假设线程1 去查询缓存，然后从 value 中判断出来当前的数据已经过期了，此时线程1 去获得互斥锁，那么其他线程会进行阻塞，获得了锁的线程他会开启一个线程去进行以前的重构数据的逻辑，直到新开的线程完成这个逻辑后，才释放锁，而线程1 直接进行返回，假设现在线程3 过来访问，由于线程2 持有着锁，所以线程3 无法获得锁，线程3 也直接返回数据，只有等到新开的线程2 把重建数据构建完后，其他线程才能走返回正确的数据

缓存击穿解决方案对比

基于互斥锁方式解决缓存击穿问题

需求：修改根据 id 查询商铺的业务，基于互斥锁方式来解决缓存击穿问题

互斥锁解决流程

操作锁：核心思路就是 利用 redis 的 setnx 方法来表示获取锁。只有当 setnx 中的 key 不存在时，才会写入新 key，当 setnx 中的 key 存在时，是不能写入新 key 的，并且会一直返回之前存在的 key 值

@Service
public class ShopServiceImpl extends ServiceImpl<ShopMapper, Shop> implements IShopService {

    /**
     * 根据id查询商铺信息
     *
     * @param id
     * @return
     */

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public Result queryById(Long id) {

        // 缓存穿透
        // Shop shop = queryWithPassThrough(id);

        // 用互斥锁来解决缓存击穿
        Shop shop = queryWithMutex(id);
        if (shop == null) {
            return Result.fail("店铺不存在！");
        }

        // 7. 返回
        return Result.ok(shop);
    }

    /**
     * 获取锁
     *
     * @param key
     * @return
     */
    private boolean tryLock(String key) {
        Boolean flag = stringRedisTemplate.opsForValue()
                .setIfAbsent(key, "1", 10L, TimeUnit.SECONDS);

        return BooleanUtil.isTrue(flag);
    }

    /**
     * 释放锁
     *
     * @param key
     */
    private void unLock(String key) {
        stringRedisTemplate.delete(key);
    }

    /**
     * 用互斥锁来解决缓存击穿
     * @param id
     * @return
     */
    public Shop queryWithMutex(Long id) {

        // 1. 从redis里查询商铺缓存
        // String key = "cache:shop:" + id;
        String key = RedisConstants.CACHE_SHOP_KEY + id;
        String shopJson = stringRedisTemplate.opsForValue().get(key);

        // 2. 判断缓存是否存在
        if (StrUtil.isNotBlank(shopJson)) {

            // 3. 存在，直接返回
            Shop shop = JSONUtil.toBean(shopJson, Shop.class);
            return shop;
        }

        // TODO：新增
        // 判断命中的是否为空值
        if (shopJson != null) {
            // 返回一个错误信息
            return null;
        }
        String lockKey = null;
        Shop shop = null;

        try {
            // 4. 实现缓存重建
            // 4.1 获取互斥锁
            lockKey = "lock:shop:" + id;
            boolean isLock = tryLock(lockKey);

            // 4.2 判断是否获取成功
            if (!isLock) {
                // 4.3 如果失败，则休眠并重试
                Thread.sleep(50);
                 return queryWithMutex(id);
            }

            // 4.4 如果成功，根据id查询数据库
            shop = getById(id);

            // 5. 不存在，直接返回错误
            if (shop == null) {
    
                // TODO：新增
                // 将空值写入redis
                stringRedisTemplate.opsForValue().set(key, "", 2L, TimeUnit.MINUTES);
    
                // 返回错误信息
                return null;
            }


            // 6. 存在，将商铺数据写入redis
            // 新增：设置过期时间
            stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop), 30L, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            // 7. 释放互斥锁
            unLock(lockKey);
        }

        
        // 8. 返回
        return shop;
    }
}

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

需求：修改根据 id 查询商铺的业务，基于互斥锁方式来解决缓存击穿问题

逻辑过期解决流程

思路分析：当用户开始查询 redis 时，判断是否命中，如果没有命中则直接返回空数据，不查询数据库，而一旦命中后，将 value 取出，判断 value 中的过期时间是否满足，如果没有过期，则直接返回 redis 中的数据，如果过期，则在开启独立线程后直接返回之前的数据，独立线程去重构数据，重构完成后释放互斥锁

流程：

新建一个实体类，存储过期时间，这个方案，对原来代码没有侵入性

@Data
public class RedisData {

    // 过期时间
    private LocalDateTime expireTime;

    // 存进redis的数据
    private Object data;
}

用逻辑时间解决缓存击穿问题

@Service
public class ShopServiceImpl extends ServiceImpl<ShopMapper, Shop> implements IShopService {

    /**
     * 根据id查询商铺信息
     *
     * @param id
     * @return
     */

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public Result queryById(Long id) {

        // 缓存穿透
        // Shop shop = queryWithPassThrough(id);


        // 用互斥锁来解决缓存击穿
        // Shop shop = queryWithMutex(id);

        // 用逻辑过期解决缓存击穿
        Shop shop = queryWithLogicalExpire(id);

        if (shop == null) {
            return Result.fail("店铺不存在！");
        }

        // 7. 返回
        return Result.ok(shop);
    }

    /**
     * 获取锁
     *
     * @param key
     * @return
     */
    private boolean tryLock(String key) {
        Boolean flag = stringRedisTemplate.opsForValue()
                .setIfAbsent(key, "1", 10L, TimeUnit.SECONDS);

        return BooleanUtil.isTrue(flag);
    }

    /**
     * 释放锁
     *
     * @param key
     */
    private void unLock(String key) {
        stringRedisTemplate.delete(key);
    }


    /**
     * 将shop的数据及逻辑过期时间添加到redis中
     *
     * @param id
     * @param expireSeconds
     */
    private void saveShop2Redis(Long id, Long expireSeconds) {
        // 1. 查询店铺数据
        Shop shop = getById(id);

        // 2. 封装逻辑过期时间
        RedisData redisData = new RedisData();
        redisData.setData(shop);
        redisData.setExpireTime(LocalDateTime.now().plusSeconds(expireSeconds));

        // 3. 写入redis
        stringRedisTemplate.opsForValue().set(RedisConstants.CACHE_SHOP_KEY + id
                , JSONUtil.toJsonStr(redisData));
    }

    /**
     * 使用逻辑过期解决缓存击穿
     *
     * @param id
     * @return
     */

    // 创建10个线程的线程池
    private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);

    public Shop queryWithLogicalExpire(Long id) {

        // 1. 从redis里查询商铺缓存
        // String key = "cache:shop:" + id;
        String key = RedisConstants.CACHE_SHOP_KEY + id;
        String shopJson = stringRedisTemplate.opsForValue().get(key);

        // 2. 判断缓存是否存在
        if (StrUtil.isBlank(shopJson)) {

            // 3. 不存在，直接返回null
            return null;
        }

        // 4. 命中，需要先把json反序列化为对象
        RedisData redisData = JSONUtil.toBean(shopJson, RedisData.class);
        JSONObject data = (JSONObject) redisData.getData();
        Shop shop = JSONUtil.toBean(data, Shop.class);
        LocalDateTime expireTime = redisData.getExpireTime();

        // 5. 判断是否过期
        if (expireTime.isAfter(LocalDateTime.now())) {

            // 5.1 未过期，直接返回店铺信息
            return shop;
        }

        // 5.2 已过期，需要缓存重建
        // 6. 缓存重建
        // 6.1 获取互斥锁
        String LockKey = RedisConstants.LOCK_SHOP_KEY + id;
        boolean isLock = tryLock(LockKey);

        // 6.2 判断是否获取锁成功 
        if (isLock) {
            // 6.3 成功，开启独立线程，实现缓存重建
            CACHE_REBUILD_EXECUTOR.submit(() -> {
                try {
                    // 重建缓存
                    saveShop2Redis(id, 20L);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                } finally {
                    // 释放锁
                    unLock(LockKey);
                }
            });
        }


        // 6.4 失败，返回过期的商铺信息
        return shop;
    }
}

缓存工具封装

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

方法1：将任意Java对象序列化为 json 并存储在 string 类型的 key 中，并且可以设置 TTL过期时间
方法2：将任意Java对象序列化为 json 并存储在 string 类型的 key 中，并且可以设置逻辑过期时间，用于处理缓存击穿问题
方法3：根据指定的key查询缓存，并反序列化为指定类型，利用缓存空值的方式解决缓存穿透问题
方法4：根据指定的key查询缓存，并反序列化为指定类型，需要利用逻辑过期解决缓存击穿问题

代码实现工具类：

@Slf4j
@Component
public class CacheClient {

    private final StringRedisTemplate stringRedisTemplate;

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

    public void set(String key, Object value, Long time, TimeUnit unit) {
        String jsonStr = JSONUtil.toJsonStr(value);
        stringRedisTemplate.opsForValue().set(key, jsonStr, time, unit);
    }

    /**
     * 封装逻辑过期时间
     *
     * @param key
     * @param value
     * @param time
     * @param unit
     */
    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)));

        String jsonStr = JSONUtil.toJsonStr(redisData);

        stringRedisTemplate.opsForValue().set(key, jsonStr);
    }

    /**
     * 缓存穿透工具类封装
     * <R> R 定义泛型R，使用泛型
     *
     * @param keyPrefix
     * @param id
     * @param dbFallBack 这里Function<ID, R> ID为传入的类型，R为返回值的类型
     * @param type       返回的类型
     * @param <R>        泛型的推断
     * @return
     */
    public <R, ID> R queryWithPassThrough(String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallBack, Long time, TimeUnit unit) {

        // 1. 从redis里查询商铺缓存
        String key = keyPrefix + id;
        String json = stringRedisTemplate.opsForValue().get(key);

        // 2. 判断缓存是否存在
        if (StrUtil.isNotBlank(json)) {

            // 3. 存在，直接返回
            return JSONUtil.toBean(json, type);
        }

        // 判断命中的是否为空值
        if (json != null) {
            // 返回一个错误信息
            return null;
        }

        // 4. 不存在，根据id查询数据库
        R r = dbFallBack.apply(id);

        // 5. 不存在，直接返回错误
        if (r == null) {

            // 将空值写入redis
            stringRedisTemplate.opsForValue().set(key, "", 2L, TimeUnit.MINUTES);

            // 返回错误信息
            return null;
        }


        // 6. 存在，将数据写入redis
        this.set(key, r, time, unit);

        // 7. 返回
        return r;
    }

    /**
     * 获取锁
     *
     * @param key
     * @return
     */
    private boolean tryLock(String key) {
        Boolean flag = stringRedisTemplate.opsForValue()
                .setIfAbsent(key, "1", 10L, TimeUnit.SECONDS);

        return BooleanUtil.isTrue(flag);
    }

    /**
     * 释放锁
     *
     * @param key
     */
    private void unLock(String key) {
        stringRedisTemplate.delete(key);
    }

    // 创建10个线程的线程池
    private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(10);

    /**
     * 基于逻辑时间解决缓存击穿工具类
     * @param keyPrefix
     * @param id
     * @param type
     * @param dbFallback
     * @param time
     * @param unit
     * @return
     * @param <R>
     * @param <ID>
     */
    public <R, ID> R queryWithLogicalExpire(String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit) {

        // 1. 从redis里查询商铺缓存
        String key = keyPrefix + id;
        String json = stringRedisTemplate.opsForValue().get(key);

        // 2. 判断缓存是否存在
        if (StrUtil.isBlank(json)) {

            // 3. 不存在，直接返回null
            return null;
        }

        // 4. 命中，需要先把json反序列化为对象
        RedisData redisData = JSONUtil.toBean(json, RedisData.class);
        JSONObject data = (JSONObject) redisData.getData();
        R r = JSONUtil.toBean(data, type);
        LocalDateTime expireTime = redisData.getExpireTime();

        // 5. 判断是否过期
        if (expireTime.isAfter(LocalDateTime.now())) {

            // 5.1 未过期，直接返回店铺信息
            return r;
        }

        // 5.2 已过期，需要缓存重建
        // 6. 缓存重建
        // 6.1 获取互斥锁
        String LockKey = RedisConstants.LOCK_SHOP_KEY + id;
        boolean isLock = tryLock(LockKey);

        // 6.2 判断是否获取锁成功
        if (isLock) {
            // 6.3 成功，开启独立线程，实现缓存重建
            CACHE_REBUILD_EXECUTOR.submit(() -> {
                try {
                    // 重建缓存
                    // 先查数据库
                    R r1 = dbFallback.apply(id);

                    // 再写入缓存
                    this.setWithLogicalExpire(key, r1, time, unit);

                } catch (Exception e) {
                    throw new RuntimeException(e);
                } finally {
                    // 释放锁
                    unLock(LockKey);
                }
            });
        }

        // 6.4 失败，返回过期的商铺信息
        return r;
    }
}

测试工具类：

@Resource
private StringRedisTemplate stringRedisTemplate;
@Resource
private CacheClient cacheClient;
@Override
public Result queryById(Long id) {
    
    // 缓存穿透工具类测试
    // Shop shop = queryWithPassThrough(id);
    Shop shop = cacheClient.queryWithPassThrough(RedisConstants.CACHE_SHOP_KEY, id, Shop.class
            , id2 -> getById(id2), RedisConstants.CACHE_SHOP_TTL, TimeUnit.MINUTES);
    
    // 用互斥锁来解决缓存击穿
    // Shop shop = queryWithMutex(id);
    
    // 用逻辑过期解决缓存击穿工具类测试
    cacheClient.queryWithLogicalExpire(RedisConstants.CACHE_SHOP_KEY, id, Shop.class
            , this::getById, RedisConstants.CACHE_SHOP_TTL, TimeUnit.MINUTES);
    
    if (shop == null) {
        return Result.fail("店铺不存在！");
    }
    
    // 返回
    return Result.ok(shop);
}

优惠券秒杀

全局唯一 ID

当用户抢购优惠券时，就会生成订单并保存到 tb_voucher_order 这张表中，而订单表如果 使用数据库自增 ID 就存在一些问题：

id 的规律性太明显
受单表数据量的限制

全局 ID 生成器

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

全局ID生成器特征

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

ID的组成部分：

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

Redis 生产全局唯一ID：

@Component
public class RedisIdWorker {

    /**
     * 开始时间戳
     */
    private static final long BEGIN_TIMESTAMP = 1721174400;

     /**
     * 序列号位数
     */
    private static final long COUNT_BITS = 32;


    private StringRedisTemplate stringRedisTemplate;

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

    public long nextId(String keyPrefix) {
        // 1. 生成时间戳
        LocalDateTime now = LocalDateTime.now();
        long nowSecond = now.toEpochSecond(ZoneOffset.UTC);
        long timestamp = nowSecond - BEGIN_TIMESTAMP;

        // 2. 生成序列号
        // 2.1 获取当前日期，精确到天
        String date = now.format(DateTimeFormatter.ofPattern("yyyy:MM:dd"));
        // 2.2 自增长
        Long count = stringRedisTemplate.opsForValue().increment("icr:" + keyPrefix + ":" + date);
        
        // 3. 拼接并返回
        
        return timestamp << COUNT_BITS | count;
    }

    // 生成 BEGIN_TIMESTAMP = 1721174400
//    public static void main(String[] args) {
//        LocalDateTime time = LocalDateTime.of(2024, 7, 17, 0, 0, 0);
//        long second = time.toEpochSecond(ZoneOffset.UTC);
//        System.out.println(second);
//    }
}

总结：

全局唯一ID生成策略：

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

Redis自增ID策略：

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

实现优惠券秒杀下单

数据库表：

tb_voucher：优惠券的基本信息，优惠金额、使用规则等
tb_seckill_voucher：优惠券的库存、开始抢购时间，结束抢购时间。特价优惠券才需要填写这些信息

**新增普通卷代码： **VoucherController

@PostMapping
public Result addVoucher(@RequestBody Voucher voucher) {
    voucherService.save(voucher);
    return Result.ok(voucher.getId());
}

新增秒杀卷代码：

VoucherController

@PostMapping("seckill")
public Result addSeckillVoucher(@RequestBody Voucher voucher) {
    voucherService.addSeckillVoucher(voucher);
    return Result.ok(voucher.getId());
}

VoucherServiceImpl

@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());
}

优惠券秒杀的下单功能流程

秒杀下单时需要判断两点：

秒杀是否开始或结束，如果尚未开始或已经结束则无法下单
库存是否充足，不足则无法下单

下单核心逻辑分析：

当用户开始进行下单，应当去查询优惠卷信息，查询到优惠卷信息，判断是否满足秒杀条件

比如时间是否充足，如果时间充足，则进一步判断库存是否足够，如果两者都满足，则扣减库存，创建订单，然后返回订单 id，如果有一个条件不满足则直接结束。

// VoucherOrderServiceImpl 代码

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    
    @Autowired
    private ISeckillVoucherService seckillVoucherService;
    
    @Autowired
    private RedisIdWorker redisIdWorker;
    
    @Override
    @Transactional
    public Result seckillVoucher(Long voucherId) {
        
        // 1. 查询优惠券
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);

        // 2. 判断秒杀是否开始
        LocalDateTime beginTime = voucher.getBeginTime();
        if (beginTime.isAfter(LocalDateTime.now())) {
            // 尚未开始
            return Result.fail("秒杀尚未开始");
        }
 
        // 3. 判断秒杀是否结束
        LocalDateTime endTime = voucher.getEndTime();
        if (endTime.isBefore(LocalDateTime.now())) {
            // 已经结束
            return Result.fail("秒杀已经结束");
        }

        // 4. 判断库存是否充足
        Integer stock = voucher.getStock();
        if (stock < 1) {
            // 库存不足
            return Result.fail("库存不足");
        }

        // 5. 扣减库存
        boolean success = seckillVoucherService.update()
                .setSql("stock = stock - 1")
                .eq("voucher_id", voucherId)
                .update();
        if (!success) {
            // 扣减失败
            return Result.fail("库存不足");
        }

        // 6. 创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        // 6.1 订单id
        long orderId = redisIdWorker.nextId("order");
        voucherOrder.setId(orderId);

        // 6.2 用户id
        Long userId = UserHolder.getUser().getId();
        voucherOrder.setId(userId);

        // 6.3 代金券id
        voucherOrder.setVoucherId(voucherId);

        save(voucherOrder);
        
        // 7. 返回订单id
        return Result.ok(orderId);
    }
}

超卖问题

超卖问题出现原因

假设线程1过来查询库存，判断出来库存大于1，正准备去扣减库存，但是还没有来得及去扣减，此时线程2过来，线程2也去查询库存，发现这个数量一定也大于1，那么这两个线程都会去扣减库存，最终多个线程相当于一起去扣减库存，此时就会出现库存的超卖问题

超卖问题是典型的多线程安全问题，针对这一问题的常见解决方案就是加锁，而对于加锁，通常有两种解决方案：

悲观锁 vs 乐观锁

悲观锁：

添加同步锁，让线程串行执行，比如 syn，和 lock 都是悲观锁的代表，同时，悲观锁中又可以再细分为公平锁，非公平锁，可重入锁等等
优点：简单粗暴
缺点：性能一般

乐观锁：

不加锁，在更新时判断是否有其他线程在修改
优点：性能好
缺点：存在成功率低的问题
乐观锁的关键是判断之前查询得到的数据是否有被修改过，常见的方式有两种：
- 版本号法
  - 会有一个版本号，每次操作数据会对版本号+1，再提交回数据时，会去校验是否比之前的版本大1 ，如果大1 ，则进行操作成功，这套机制的核心逻辑在于，如果在操作过程中，版本号只比原来大1 ，那么就意味着操作过程中没有人对它进行过修改，它的操作就是安全的，如果不大1，则数据被修改过
- CAS法
  - CAS 就是 Compare And Set（先比较再更新），是前面方案的升级，将版本号取消，用库存代替版本号比较

使用CAS法修改之前的代码解决超卖问题：

// ...
// 5. 扣减库存
boolean success = seckillVoucherService.update()
        .setSql("stock = stock - 1") // set stock = stock - 1
        .eq("voucher_id", voucherId)  // where id = ? and stock = ?
        .eq("stock", stock) // 加上一个对stock值的判断
        .update();

if (!success) {
    
    // 扣减失败
    return Result.fail("库存不足");
}

上面代码存在的问题：只要我扣减库存时的库存和之前我查询到的库存是一样的，就意味着没有人在中间修改过库存，那么此时就是安全的，但是以上这种方式通过测试发现会有很多失败的情况
失败的原因：在使用乐观锁过程中假设100个线程同时都拿到了100的库存，然后大家一起去进行扣减，但是100个人中只有1个人能扣减成功，其他的人在处理时，他们在扣减时，库存已经被修改过了，所以此时其他线程都会失败

对产生失败原因代码的修改：

由于成功的概率太低，所以乐观锁需要变一下，改成 stock 大于 0 即可

// 5. 扣减库存
boolean success = seckillVoucherService.update()
        .setSql("stock = stock - 1") // set stock = stock - 1
        .eq("voucher_id", voucherId)  // where id = ? and stock > 0
        .gt("stock", 0) // 加上一个对stock值的判断
        .update();

if (!success) {
    
    // 扣减失败
    return Result.fail("库存不足");
}

一人一单

需求：修改秒杀业务，要求同一个优惠券，一个用户只能下一单

一人一单流程图

// ...

// 6. 一人一单
Long UserId = UserHolder.getUser().getId();

// 6.1 查询订单
int count = query()
        .eq("user_id", UserId)
        .eq("voucher_id", voucherId)
        .count();

// 6.2 判断是否存在
if (count > 0) {
    
    // 用户已经购买过
    return Result.fail("该用户已经购买过");
}

// 扣减库存

**存在问题：**现在的问题还是和之前一样，并发过来，查询数据库，都不存在订单，所以我们还是需要加锁，但是 乐观锁比较适合更新数据，而 现在是插入数据，所以需要使用悲观锁操作

解决办法：首先我们的初始方案是封装了一个 createVoucherOrder 方法，同时为了确保他线程安全，在方法上添加了一把 synchronized 锁

@Transactional
public synchronized Result createVoucherOrder(Long voucherId) {
    // 6. 一人一单
    Long UserId = UserHolder.getUser().getId();
    
    // 6.1 查询订单
    int count = query()
            .eq("user_id", UserId)
            .eq("voucher_id", voucherId)
            .count();
    
    // 6.2 判断是否存在
    if (count > 0) {
        // 用户已经购买过
        return Result.fail("该用户已经购买过");
    }
    
    // 7. 扣减库存
    boolean success = seckillVoucherService.update()
            .setSql("stock = stock - 1") // set stock = stock - 1
            .eq("voucher_id", voucherId)  // where id = ? and stock > 0
            .gt("stock", 0) // 加上一个对stock值的判断
            .update();
    if (!success) {
        // 扣减失败
        return Result.fail("库存不足");
    }
    
    // 8. 创建订单
    VoucherOrder voucherOrder = new VoucherOrder();
    
    // 8.1 订单id
    long orderId = redisIdWorker.nextId("order");
    voucherOrder.setId(orderId);
    
    // 8.2 用户id
    Long userId = UserHolder.getUser().getId();
    voucherOrder.setId(userId);
    
    // 8.3 代金券id
    voucherOrder.setVoucherId(voucherId);
    save(voucherOrder);
    
    // 9. 返回订单id
    return Result.ok(orderId);
}

存在的问题：但是这样添加锁，锁的粒度太粗了，在使用锁过程中，控制 锁粒度 是一个非常重要的事情，因为如果锁的粒度太大，会导致每个线程进来都会锁住，所以需要去控制锁的粒度

@Transactional
public Result createVoucherOrder(Long voucherId) {
    // 6. 一人一单
    Long UserId = UserHolder.getUser().getId();
    
    // intern() 会去返回字符串值相同的地址，保证锁必须是同一把
    synchronized (UserId.toString().intern()) {
        
        // 6.1 查询订单
        int count = query()
                .eq("user_id", UserId)
                .eq("voucher_id", voucherId)
                .count();
        
        // 6.2 判断是否存在
        if (count > 0) {
            // 用户已经购买过
            return Result.fail("该用户已经购买过");
        }
        
        // 7. 扣减库存
        boolean success = seckillVoucherService.update()
                .setSql("stock = stock - 1") // set stock = stock - 1
                .eq("voucher_id", voucherId)  // where id = ? and stock > 0
                .gt("stock", 0) // 加上一个对stock值的判断
                .update();
        if (!success) {
            // 扣减失败
            return Result.fail("库存不足");
        }
        
        // 8. 创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        
        // 8.1 订单id
        long orderId = redisIdWorker.nextId("order");
        voucherOrder.setId(orderId);
        
        // 8.2 用户id
        Long userId = UserHolder.getUser().getId();
        voucherOrder.setId(userId);
        
        // 8.3 代金券id
        voucherOrder.setVoucherId(voucherId);
        save(voucherOrder);
        
        // 9. 返回订单id
        return Result.ok(orderId);
    }
}

存在的问题：但是以上代码还是存在问题，问题的原因在于当前方法被 spring 的事务控制，如果在方法内部加锁，可能会导致当前方法事务还没有提交，但是锁已经释放也会导致问题，所以选择将当前方法整体包裹起来，确保事务不会出现问题

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

synchronized (userId.toString().intern()) {
    return createVoucherOrder(voucherId, userId);
}

存在的问题：以上做法依然有问题，因为想要调用的方法，其实是 this. 的方式调用的，事务想要生效，还得利用代理来生效，所以这个地方，需要获得原始的事务对象，来操作事务

Long userId = UserHolder.getUser().getId();
synchronized (userId.toString().intern()) {
    
    // 获得IVoucherOrderService的代理对象（解决事务）
    IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
    return proxy.createVoucherOrder(voucherId);
}

同时导入依赖

<dependency>
    <groupId>org.aspectj</groupId>
    <artifactId>aspectjweaver</artifactId>
</dependency>

并且在引导类添加注解使代理对象暴露

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

    public static void main(String[] args) {
        SpringApplication.run(HmDianPingApplication.class, args);
    }

}

集群环境下的并发问题

单体项目锁的实现流程

集群下锁的实现流程

由于现在部署了多个 tomcat，每个 tomcat 都有一个属于自己的 jvm，那么假设在服务器 A 的 tomcat 内部，有两个线程，这两个线程由于使用的是同一份代码，那么他们的锁对象是同一个，是可以实现互斥的，但是如果现在是服务器 B 的 tomcat 内部，又有两个线程，但是它们的锁对象写的虽然和服务器A一样，但是锁对象却不是同一个，所以线程3 和线程4 可以实现互斥，但是却无法和线程1 和线程2 实现互斥，这就是集群环境下，syn 锁失效的原因，在这种情况下，就需要使用 分布式锁 来解决这个问题

分布式锁

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

分布式锁的核心思想就是 让大家都使用同一把锁，只要大家使用的是同一把锁，那么就能锁住线程，不让线程进行，让程序串行执行，这就是分布式锁的核心思路

分布式锁

分布式锁的特征：

可见性：多个线程都能看到相同的结果，
- 注意：这个地方说的可见性并不是并发编程中指的内存可见性，只是说 多个进程之间都能感知到变化 的意思
互斥：互斥是分布式锁的最基本的条件，使得程序串行执行
高可用：程序不易崩溃，时时刻刻都保证较高的可用性
高性能：由于加锁本身就让性能降低，所有对于分布式锁本身需要他就 较高的加锁性能和释放锁性能
安全性：安全也是程序中必不可少的一环

常见的分布式锁有三种：

MySQL：mysql本身就带有锁机制，但是由于mysql性能本身一般，所以采用分布式锁的情况下，其实使用mysql作为分布式锁比较少见

Redis：redis作为分布式锁是非常常见的一种使用方式，现在企业级开发中基本都使用 redis 或者 zookeeper 作为分布式锁，利用 setnx 这个方法，如果插入 key 成功，则表示获得到了锁，如果有人插入成功，其他人插入失败则表示无法获得到锁，利用这套逻辑来实现分布式锁

Zookeeper：zookeeper 也是企业级开发中较好的一个实现分布式锁的方案

常见的分布式锁

基于 Redis 的分布式锁

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

获取锁：

互斥：确保只能有一个线程获取锁

# 添加锁
setnx lock thread1

# 添加过期时间，避免服务宕机
expire lock 10

# 添加锁，NX是互斥，EX是设置超时时间
set lock thread1 NX EX 10

非阻塞：尝试一次，成功返回true，失败返回false

释放锁：
- 手动释放
- 超时释放：获取锁时添加一个超时时间
  1
  2
  # 释放锁
  del lock

Redis分布式锁流程

实现 Redis 分布式锁版本 1

需求：定义一个类，实现下面接口，利用 Redis 实现分布式锁功能

锁的基本接口

利用 setnx 方法进行加锁，同时增加过期时间，防止死锁，此方法可以保证加锁和增加过期时间具有原子性

public class SimpleRedisLock implements ILock {
    /**
     * 尝试获取锁
     *
     * @param timeoutSec
     * @return
     */

    // 前缀
    private static final String key_prefix = "lock:";

    // 业务名称
    private String name;

    private StringRedisTemplate stringRedisTemplate;

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

    @Override
    public boolean tryLock(long timeoutSec) {

        // 获取线程的表示
        long threadId = Thread.currentThread().getId();

        // 获取锁
        // setIfAbsent 表示 NX
        Boolean success = stringRedisTemplate
                .opsForValue()
                .setIfAbsent(key_prefix + name, threadId + "", timeoutSec, TimeUnit.SECONDS);

        // 避免空指针的风险
        return Boolean.TRUE.equals(success);
    }

    /**
     * 释放锁
     */
    @Override
    public void unlock() {
        // 释放锁
        stringRedisTemplate.delete(key_prefix + name);
    }
}

修改业务代码：

//...

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

// 创建锁对象
SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);

// 获取锁
boolean isLock = lock.tryLock(1200);

// 判断是否获取锁成功
if (!isLock) {
    
    // 获取锁失败，返回错误信息或者重试
    return Result.fail("不允许重复下单");
}
try {
    
    // 获得IVoucherOrderService的代理对象（解决事务）
    IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
    return proxy.createVoucherOrder(voucherId);
} finally {
    
    // 释放锁
    lock.unlock();
}

Redis 分布式锁误删情况说明

出现的问题：持有锁的线程在锁的内部出现了阻塞，导致他的锁自动释放，这时其他线程，线程2 来尝试获得锁，就拿到了这把锁，然后线程2 在持有锁执行过程中，线程1 反应过来，继续执行，而线程1 执行过程中，走到了删除锁逻辑（因为取到的key值为前缀+userId），此时就会把本应该属于线程2 的锁进行删除，这就是误删别人锁的情况说明

Redis分布式锁误删情况说明

解决方案：解决方案就是在 每个线程释放锁的时候，去判断一下当前这把锁是否属于自己，如果属于自己，则不进行锁的删除，假设还是上边的情况，线程1卡顿，锁自动释放，线程2进入到锁的内部执行逻辑，此时线程1反应过来，然后删除锁，但是线程1，一看当前这把锁不是属于自己，于是不进行删除锁逻辑，当线程2走到删除锁逻辑时，如果没有卡过自动释放锁的时间点，则判断当前这把锁是属于自己的，于是删除这把锁

业务流程

需求：修改之前的分布式锁实现

满足：

在获取锁时存入线程标示（可以用 UUID 表示）
在释放锁时先获取锁中的线程标示，判断是否与当前线程标示一致
- 如果一致则释放锁
- 如果不一致则不释放锁

核心逻辑：在存入锁时，放入自己线程的标识，在删除锁时，判断当前这把锁的标识是不是自己存入的，如果是，则进行删除，如果不是，则不进行删除。

public class SimpleRedisLock implements ILock {
    /**
     * 尝试获取锁
     *
     * @param timeoutSec
     * @return
     */

    // 锁的名称前缀
    private static final String KEY_PREFIX = "lock:";
    
    // 线程表示id前缀
    private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";

    // 业务名称
    private String name;

    private StringRedisTemplate stringRedisTemplate;

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

    @Override
    public boolean tryLock(long timeoutSec) {

        // 获取线程的表示
        String threadId = ID_PREFIX + Thread.currentThread().getId();

        // 获取锁
        // setIfAbsent 表示 NX
        Boolean success = stringRedisTemplate
                .opsForValue()
                .setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);

        // 避免空指针的风险
        return Boolean.TRUE.equals(success);
    }

    /**
     * 释放锁
     */
    @Override
    public void unlock() {
        
        // 获取线程标识
        String threadId = ID_PREFIX + Thread.currentThread().getId();
        
        // 获取锁中的标识
        String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);

        // 判断标识是否一致
        if (threadId.equals(id)) {
            // 释放锁
            stringRedisTemplate.delete(KEY_PREFIX + name);
        }

    }
}

Redis 分布式锁的原子性问题

Redis分布式锁的原子性问题

出现的问题：线程1 现在持有锁之后，在执行业务逻辑过程中，它正准备删除锁，而且已经走到了条件判断的过程中，比如它已经拿到了当前这把锁确实是属于它自己的，正准备删除锁，但是此时它的锁到期了，那么此时线程2进来，但是线程1他会接着往后执行，当他卡顿结束后，他直接就会执行删除锁那行代码，相当于条件判断并没有起到作用，这就是删锁时的原子性问题，之所以有这个问题，是因为线程1的拿锁，比锁，删锁，实际上并不是原子性的，必须要防止刚才的情况发生，可以用 Lua 脚本解决

Redis 的 Lua 脚本

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

可以使用 Lua 去操作 redis，又能保证它的原子性，这样就可以实现拿锁比锁删锁是一个原子性动作

Redis 提供的调用函数，语法如下：

1	redis.call('命令名称', 'key', '其它参数', ...)

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

1 2	# 执行 set name jack redis.call('set', 'name', 'jack')

例如：要先执行 set name Rose，再执行 get name，则脚本如下：

# 先执行 set name jack
redis.call('set', 'name', 'Rose')
# 再执行 get name
local name = redis.call('get', 'name')
# 返回
return name

写好脚本以后，需要用 Redis 命令来调用脚本，调用脚本的常见命令如下：

用Redis命令来调用脚本

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

调用脚本

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

调用脚本

释放锁的业务流程是：

获取锁中的线程标示
判断是否与指定的标示（当前线程标示）一致
如果一致则释放锁（删除）
如果不一致则什么都不做

如果用 Lua 脚本来表示则是：

-- 锁的key
-- local key = KEYS[1]

-- 当前线程标识
-- local threadId = ARGV[1]

-- 获取锁中的线程标识 get key
local id = redis.call('get', KEYS[1])

-- 比较线程标识与锁的标识是否一致
if (id == ARGV[1]) then
    -- 释放锁 del key
    redis.call('del', KEYS[1])
end
-- 不一致，则直接返回
return 0

利用 Java 代码调用 Lua 脚本改造分布式锁

需求：基于 Lua 脚本实现分布式锁的释放锁逻辑

提示：RedisTemplate 调用 Lua 脚本的 API 如下：

Lua脚本的API

/**
 * 释放锁
 */
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 void unlock() {
    
    // 调用Lua脚本
    stringRedisTemplate.execute(UNLOCK_SCRIPT
            , Collections.singletonList(KEY_PREFIX + name),
            ID_PREFIX + Thread.currentThread().getId());
}

总结：

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

利用 set nx ex 获取锁，并设置过期时间，保存线程标示
释放锁时先判断线程标示是否与自己一致，一致则删除锁

特性：

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

利用添加过期时间，防止死锁问题的发生，但是有了过期时间之后，可能出现误删别人锁的问题，这个问题开始是利用删之前通过拿锁，比锁，删锁这个逻辑来解决的，也就是删之前判断一下当前这把锁是否是属于自己的，但是现在还有原子性问题，也就是没法保证拿锁比锁删锁是一个原子性的动作，最后通过 lua 表达式来解决这个问题

分布式锁 - Redission

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

重入问题：重入问题是指获得锁的线程可以再次进入到相同的锁的代码块中，可重入锁的意义在于防止死锁，比如：HashTable 这样的代码中，它的方法都是使用 synchronized 修饰的，假如它在一个方法内，调用另一个方法，那么此时如果是不可重入的，就死锁了，所以可重入锁它的主要意义是防止死锁，synchronized 和 Lock 锁都是可重入的
不可重试：是指目前的分布式只能尝试一次，合理的情况是：当线程在获得锁失败后，它应该能再次尝试获得锁
超时释放：因为在加锁时增加了过期时间，这样可以防止死锁，但是如果卡顿的时间超长，虽然采用了lua表达式防止删锁的时候，误删别人的锁，但是毕竟没有锁住，有安全隐患
主从一致性：如果 Redis 提供了主从集群，当向集群写数据时，主机需要异步的将数据同步给从机，而万一在同步过去之前，主机宕机了，就会出现死锁问题

分布式锁 - redission 功能介绍

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

也就是说：redission 是一个在 redis 基础上实现的分布式工具的集合

官网地址：https://redisson.org

地址：https://github.com/redisson/redisson

Redission 入门案例

流程：

引入依赖

<dependency>
	<groupId>org.redisson</groupId>
	<artifactId>redisson</artifactId>
	<version>3.13.6</version>
</dependency>

配置 Redisson 客户端

@Configuration
public class RedissonConfig {

    @Bean
    public RedissonClient redissonClient() {

        // 1. 配置
        Config config = new Config();
        // useSingleServer()：使用单节点模式
        config.useSingleServer()
                .setAddress("redis://192.168.88.129:6379")
                .setPassword("leadnews");

        // 2. 创建redisson对象
        return Redisson.create(config);
    }
}

使用 Redission 的分布式锁

@Resource
private RedissionClient redissonClient;

@Test
void testRedisson() throws Exception{
    // 获取锁(可重入)，指定锁的名称
    RLock lock = redissonClient.getLock("anyLock");
    // 尝试获取锁，参数分别是：获取锁的最大等待时间(期间会重试)，锁自动释放时间，时间单位
    boolean isLock = lock.tryLock(1,10,TimeUnit.SECONDS);
    // boolean isLock = lock.tryLock(); 无参默认最大等待时间为-1，立刻返回
    // 判断获取锁成功
    if(isLock){
        try{
            System.out.println("执行业务");          
        }finally{
            // 释放锁
            lock.unlock();
        }
        
    } 
}

修改 VoucherOrderServiceImpl 代码，注入 RedissonClient

@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {


    @Autowired
    private ISeckillVoucherService seckillVoucherService;

    @Autowired
    private RedisIdWorker redisIdWorker;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Resource
    private RedissonClient redissonClient;

    @Override
    public Result seckillVoucher(Long voucherId) {

        // 1. 查询优惠券
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);

        // 2. 判断秒杀是否开始
        LocalDateTime beginTime = voucher.getBeginTime();
        if (beginTime.isAfter(LocalDateTime.now())) {
            // 尚未开始
            return Result.fail("秒杀尚未开始");
        }

        // 3. 判断秒杀是否结束
        LocalDateTime endTime = voucher.getEndTime();
        if (endTime.isBefore(LocalDateTime.now())) {
            // 已经结束
            return Result.fail("秒杀已经结束");
        }

        // 4. 判断库存是否充足
        Integer stock = voucher.getStock();
        if (stock < 1) {
            // 库存不足
            return Result.fail("库存不足");
        }

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

        // 创建锁对象
        // SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
        RLock lock = redissonClient.getLock("lock:order:" + userId);

        // 获取锁
        boolean isLock = lock.tryLock();

        // 判断是否获取锁成功
        if (!isLock) {
            // 获取锁失败，返回错误信息或者重试
            return Result.fail("不允许重复下单");
        }

        try {

            // 获得IVoucherOrderService的代理对象（解决事务）
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
        } finally {

            // 释放锁
            lock.unlock();
        }

    }
	// ....
}

分布式锁 - redission可重入锁原理

redission可重入锁原理

在 Lock 锁中，它是借助于底层的一个 voaltile 的 state 变量来记录重入的状态的，比如：当前没有人持有这把锁，那么 state=0，假如有人持有这把锁，那么 state=1，如果持有这把锁的人再次持有这把锁，那么state 就会 +1；如果是对于 synchronized 而言，它在c语言代码中会有一个 count，原理和 state 类似，也是重入一次就加一，释放一次就 -1，直到减少成 0 时，表示当前这把锁没有被人持有
在redission中，也支持支持可重入锁，在分布式锁中，它采用 hash 结构用来存储锁，其中大key表示表示这把锁是否存在，用小key表示当前这把锁被哪个线程持有

获取锁的 lua 脚本

local key = KEYS[1]; -- 锁的key
local threadId = ARGV[1]; -- 线程唯一标识
local releaseTime = ARGV[2]; -- 锁的自动释放时间
-- 判断是否存在
if(redis.call('exists', key) == 0) then
    -- 不存在, 获取锁    
    redis.call('hset', key, threadId, '1'); 
    -- 设置有效期
    redis.call('expire', key, releaseTime); 
    return 1; -- 返回结果
end;
-- 锁已经存在，判断threadId是否是自己
if(redis.call('hexists', key, threadId) == 1) then
    -- 不存在, 获取锁，重入次数+1
    redis.call('hincrby', key, threadId, '1'); 
    -- 设置有效期
    redis.call('expire', key, releaseTime); 
    return 1; -- 返回结果
end;
return 0; -- 代码走到这里,说明获取锁的不是自己，获取锁失败

释放锁的 lua 脚本

local key = KEYS[1]; -- 锁的key
local threadId = ARGV[1]; -- 线程唯一标识
local releaseTime = ARGV[2]; -- 锁的自动释放时间
-- 判断当前锁是否还是被自己持有
if (redis.call('HEXISTS', key, threadId) == 0) then
    return nil; -- 如果已经不是自己，则直接返回
end; 
-- 是自己的锁，则重入次数-1
local count = redis.call('HINCRBY', key, threadId, -1);
-- 判断是否重入次数是否已经为0 
if (count > 0) then
    -- 大于0说明不能释放锁，重置有效期然后返回
    redis.call('EXPIRE', key, releaseTime);
    return nil;
else  -- 等于0说明可以释放锁，直接删除
    redis.call('DEL', key);
    return nil;
end;

分布式锁 - redission锁重试和WatchDog机制

抢锁过程中，获得当前线程，通过 tryAcquire 进行抢锁，该抢锁逻辑和之前逻辑相同

先判断当前这把锁是否存在，如果不存在，插入一把锁，返回 null
判断当前这把锁是否是属于当前线程，如果是，则返回 null

这里没看懂…

分布式锁 - redission锁的MutiLock原理

为了提高 redis 的可用性，会搭建集群或者主从，现在以主从为例

此时去写命令，写在主机上，主机会将数据同步给从机，但是假设在主机还没有来得及把数据写入到从机去的时候，此时主机宕机，哨兵会发现主机宕机，并且选举一个 slave 变成 master，而此时新的 master 中实际上并没有锁信息，此时锁信息就已经丢掉了

分布式锁的主从一致性问题

为了解决这个问题，redission 提出来了 MutiLock 锁，使用这把锁就不使用主从了，每个节点的地位都是一样的，这把锁加锁的逻辑需要写入到每一个主丛节点上，只有所有的服务器都写入成功，此时才是加锁成功，假设现在某个节点挂了，那么去获得锁的时候，只要有一个节点拿不到，都不能算是加锁成功，就保证了加锁的可靠性

MutiLock锁

MutiLock 加锁原理 ：当设置了多个锁时，redission 会将多个锁添加到一个集合中，然后用 while 循环去不停去尝试拿锁，但是会有一个总共的加锁时间，这个时间是用需要加锁的个数 * 1500ms ，假设有3个锁，那么时间就是 4500ms，假设在这 4500ms 内，所有的锁都加锁成功，那么此时才算是加锁成功，如果在4500ms有线程加锁失败，则会再次去进行重试

MutiLock 加锁原理

总结：

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

Redis 秒杀优化

Redis秒杀业务之前的流程

用户发起请求，此时会请求 nginx，nginx 会访问到 tomcat，在 tomcat 中的程序，会进行串行操作，分成如下步骤：
1. 查询优惠卷
2. 判断秒杀库存是否足够
3. 查询订单
4. 校验是否是一人一单
5. 扣减库存
6. 创建订单
目前的业务1、3、5、6步骤需要查询数据库，而且还需要分布式锁，现在程序的性能低，并发的能力也比较弱

优化思路

优化的整体思路：
1. 当用户下单之后，判断库存是否充足只需要去 redis 中去根据 key 找对应的 value 是否大于0即可，如果不充足，则直接结束，如果充足，继续在 redis 中判断用户是否可以下单，如果 set 集合中没有这条数据，说明可以下单，如果 set 集合中没有这条记录，则将 userId 和优惠卷存入到 redis 中，并且返回0，整个过程需要保证是原子性的，可以使用 lua 来操作
2. 当以上判断逻辑走完之后，还可以判断当前redis中返回的结果是否是0 ，如果是0，则表示可以下单，则将之前说的信息存入到到queue中去，然后返回，然后再来个线程异步的下单，前端可以通过返回的订单id来判断是否下单成功

redis中的优化操作

用redis中的String结构存储剩余库存，用set结构存储购买过优惠券的userId

秒杀优化 - Redis完成秒杀资格判断

需求：

新增秒杀优惠券的同时，将优惠券信息保存到 Redis 中
基于 Lua 脚本，判断秒杀库存、一人一单，决定用户是否抢购成功
如果抢购成功，将优惠券id和用户id封装后存入阻塞队列
开启线程任务，不断从阻塞队列中获取信息，实现异步下单功能

流程：

修改新增优惠券的代码，在 VoucherServiceImpl 中

@Resource
private StringRedisTemplate stringRedisTemplate;

@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:" + voucher.getId()
            , voucher.getStock().toString());
    
}

完整 lua 表达式

-- 1. 参数列表

-- 1.1 优惠券id
local voucherId = ARGV[1]

-- 1.2 用户id
local userId = ARGV[2]

-- 2. 数据key
-- 2.1 库存key
-- .. 相当于String中的 +
local stockKey = 'seckill:stock:' .. voucherId

-- 2.2 订单key
local orderKey = 'seckill:order:' .. voucherId

-- 3. 脚本业务
-- 3.1 判断库存是否充足 get stockKey
if (tonumber(redis.call('get', stockKey)) <= 0) then
    -- 3.2 库存不足，返回1
    return 1
end

-- 3.2 判断用户是否下单
-- sismember：判断是不是集合成员 sismember orderKey userId
if (redis.call("sismember", orderKey, userId) == 1) then
    -- 3.3 存在说明是重复下单
    return 2
end

-- 3.3 扣减库存 incrby stockKey -1
redis.call("incrby", stockKey, -1)

-- 3.4 下单（保存用户）sadd orderKey userId
redis.call('sadd', orderKey, userId)

-- 0表示成功，1、2表示失败
return 0

当以上lua表达式执行完毕后，剩下的就是来完成需求3、4

/**
 * 基于lua脚本的优化
 *
 * @param voucherId
 * @return
 */
@Autowired
private StringRedisTemplate stringRedisTemplate;

private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

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()
    );
    
    // 2. 判断结果是否为 0
    int res = result.intValue();
    if (res != 0) {
    
        // 2.1 不为0，代表没有购买资格
        return Result.fail(res == 1 ? "库存不足" : "不能重复下单");
    }
    
    // 2.2 为0，代表有购买资格，把下单信息保存到阻塞队列
    // 获取订单id
    long orderId = redisIdWorker.nextId("order");
    // TODO 保存阻塞队列
    
    // 3. 返回订单id
    return Result.ok(orderId);
}

秒杀优化 - 基于阻塞队列实现秒杀优化

修改下单动作，现在去下单时，是通过 lua 表达式去原子执行判断逻辑，如果判断出来不为0 ，则要么是库存不足，要么是重复下单，返回错误信息，如果是0，则把下单的逻辑保存到队列中去，然后异步执行。

修改代码 VoucherOrderServiceImpl

@Service
@Slf4j
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {


    @Autowired
    private ISeckillVoucherService seckillVoucherService;

    @Autowired
    private RedisIdWorker redisIdWorker;


    @Resource
    private RedissonClient redissonClient;

    /**
     * 基于lua脚本的优化
     *
     * @param voucherId
     * @return
     */

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    // 设置一个阻塞队列
    private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);

    // 创建一个线程池
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

    /**
     * @PostConstruct：在创建完类之后执行这段代码
     */
    @PostConstruct
    private void init() {
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }

    // 内部类实现线程池
    private class VoucherOrderHandler implements Runnable {

        @Override
        public void run() {
            while (true) {

                try {
                    // 1. 获取队列中的订单信息
                    VoucherOrder order = orderTasks.take();

                    // 2. 创建订单
                    handleVoucherOrder(order);
                } catch (Exception e) {
                    log.error("处理订单异常", e);
                }
            }
        }
    }

    /**
     * 创建订单
     *
     * @param voucherOrder
     */
    private void handleVoucherOrder(VoucherOrder voucherOrder) {

        // 获取用户
        Long userId = voucherOrder.getUserId();

        // 创建锁对象
        RLock lock = redissonClient.getLock("lock:order:" + userId);

        // 获取锁
        boolean isLock = lock.tryLock();

        // 判断是否获取锁成功
        if (!isLock) {

            // 获取锁失败，返回错误信息或者重试
            log.error("不允许重复下单");
            return;
        }
        try {
            // 注意：由于是spring的事务是放在threadLocal中，此时的是多线程，事务会失效
            proxy.createVoucherOrder(voucherOrder);
        } finally {
            // 释放锁
            lock.unlock();
        }

    }

    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;

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

    // 创建一个全局变量，接收代理对象
    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()
        );

        // 2. 判断结果是否为 0
        int res = result.intValue();
        if (res != 0) {

            // 2.1 不为0，代表没有购买资格
            return Result.fail(res == 1 ? "库存不足" : "不能重复下单");
        }

        // 2.2 为0，代表有购买资格，把下单信息保存到阻塞队列
        VoucherOrder voucherOrder = new VoucherOrder();
        // 获取订单id
        long orderId = redisIdWorker.nextId("order");
        // 设置订单id
        voucherOrder.setId(orderId);
        // 设置用户id
        voucherOrder.setUserId(userId);
        // 设置代金券id
        voucherOrder.setVoucherId(voucherId);
        // 2.1 创建阻塞队列
        orderTasks.add(voucherOrder);

        // 3. 获取代理对象
        proxy = (IVoucherOrderService) AopContext.currentProxy();

        // 4. 返回订单id
        return Result.ok(orderId);

    }

    @Transactional
    public void createVoucherOrder(VoucherOrder voucherOrder) {

        // 5. 一人一单
        Long userId = voucherOrder.getUserId();

        // intern() 会去返回字符串值相同的地址，保证锁必须是同一把
        // synchronized (UserId.toString().intern()) {
        // 6.1 查询订单
        int count = query()
                .eq("user_id", userId)
                .eq("voucher_id", voucherOrder)
                .count();

        // 6.2 判断是否存在
        if (count > 0) {
            // 用户已经购买过
            log.error("用户已经购买过一次了");
            return;
        }

        // 7. 扣减库存
        boolean success = seckillVoucherService.update()
                .setSql("stock = stock - 1") // set stock = stock - 1
                .eq("voucher_id", voucherOrder)  // where id = ? and stock > 0
                .gt("stock", 0) // 加上一个对stock值的判断
                .update();
        if (!success) {
            // 扣减失败
            log.error("库存不足！");
            return;
        }

        // 创建订单
        save(voucherOrder);

    }
}

总结：

秒杀业务的优化思路是：

先利用 Redis 完成库存余量、一人一单判断，完成抢单业务
再将下单业务放入阻塞队列，利用独立线程异步下单

基于阻塞队列的异步秒杀存在的问题：

jvm内存限制问题
数据安全问题

Redis 消息队列

消息队列

消息队列（Message Queue）：字面意思就是存放消息的队列。最简单的消息队列模型包括3个角色：

消息队列：存储和管理消息，也被称为 消息代理（Message Broker）
生产者：发送消息到消息队列
消费者：从消息队列获取消息并处理消息

消息队列

使用队列的好处在于：解耦（降低系统中不同部分之间的依赖性，使系统的各个组件能够相对独立地进行开发、维护和演化）。在本案例中，我们下单之后，利用 redis 去进行校验下单条件，再通过队列把消息发送出去，然后再启动一个线程去消费这个消息，完成解耦，同时也加快响应速度
可以使用一些现成的mq，比如：kafka、rabbitmq等等，但是呢，如果没有安装mq，也可以直接使用redis提供的mq方案，降低我们的部署和学习成本

Redis提供了三种不同的方式来实现消息队列：

list 结构：基于 List 结构模拟消息队列
PubSub：基本的点对点消息模型
Stream：比较完善的消息队列模型

基于List结构的消息队列

Redis 的 list 数据结构是一个双向链表，很容易模拟出队列效果

基于List结构模拟消息队列

队列是入口和出口不在一边，因此可以利用：LPUSH 结合 RPOP、或者 RPUSH 结合 LPOP 来实现
不过要注意的是，当队列中没有消息时 RPOP 或 LPOP 操作会返回 null，并不像JVM的阻塞队列那样会阻塞并等待消息。因此这里应该使用 BRPOP 或者 BLPOP 来实现阻塞效果

基于List的消息队列的优缺点：

优点：
1. 利用 Redis 存储，不受限于 JVM 内存上限
2. 基于 Redis 的持久化机制，数据安全性有保证
3. 可以满足消息有序性
缺点：
1. 无法避免消息丢失
2. 只支持单消费者

基于PubSub结构的消息队列

PubSub（发布订阅）是 Redis2.0 版本引入的消息传递模型。顾名思义，消费者可以订阅一个或多个channel，生产者向对应 channel 发送消息后，所有订阅者都能收到相关消息

相比较List结构的消息队列，这种消息队列可以发送给多个消费者

基于PubSub结构的消息队列

SUBSCRIBE channel [channel] ：订阅一个或多个频道
PUBLISH channel msg ：向一个频道发送消息
PSUBSCRIBE pattern[pattern] ：订阅与 pattern 格式匹配的所有频道

基于PubSub的消息队列的优缺点：

优点：
1. 采用发布订阅模型，支持多生产、多消费
缺点：
1. 不支持数据持久化
  - 如果没有对应的频道，则在发送到该频道后会丢失数据
2. 无法避免消息丢失
3. 消息堆积有上限，超出时数据丢失

基于Stream结构的消息队列 - 单消费模式

Stream 是 Redis 5.0 引入的一种新数据类型，可以实现一个功能非常完善的消息队列

发送消息的命令：

XADD

发送消息命令

eg：

发送消息举例

读取消息的命令：

XREAD

读取消息的命令:XREAD

eg：

使用XREAD读取第一个消息

注意：当我们指定起始ID为 $ 时，代表 读取最新的消息，如果在处理一条消息的过程中，又有超过1条以上的消息到达队列，则下次获取时也只能获取到最新的一条，会出现 漏读消息 的问题

Stream 类型消息队列的 XREAD 命令特点：

消息可回溯（回溯：消息不会消失，永久保存）
一个消息可以被多个消费者读取
可以阻塞读取
有 消息漏读 的风险

基于Stream结构的消息队列 - 消费者组模式

消费者组（Consumer Group）：将多个消费者划分到一个组中，监听同一个队列

特点：

常见命令：

创建消费者组：
1
XGROUP CREATE key groupName ID [MKSTREAM]
- key：队列名称
- groupName：消费者组名称
- ID：起始ID标示，$ 代表队列中最后一个消息，0 代表队列中第一个消息
- MKSTREAM：队列不存在时自动创建队列
删除指定的消费者组：
1
XGROUP DESTORY key groupName

给指定的消费者组添加消费者：

1	XGROUP CREATECONSUMER key groupname consumername

删除消费者组中的指定消费者：

1	XGROUP DELCONSUMER key groupname consumername

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

Redis项目 仿大众点评

项目初始化

短信登录

基于 session 实现短信登录

集群的 session 共享问题

基于 Redis 实现共享 session 登录

登录 拦截器 的 优化

商户查询缓存

缓存 的概述

添加 redis 缓存

缓存 更新策略

缓存 穿透

缓存 雪崩

缓存 击穿（热点 key）

基于 互斥锁 方式解决缓存击穿问题

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

缓存 工具封装

优惠券秒杀

全局唯一 ID

实现优惠券秒杀下单

超卖问题

一人一单

集群环境下的并发问题

分布式锁

基于 Redis 的分布式锁

实现 Redis 分布式锁版本 1

Redis 分布式锁误删情况说明

Redis 分布式锁的原子性问题

Redis 的 Lua 脚本

利用 Java 代码调用 Lua 脚本改造分布式锁

分布式锁 - Redission

分布式锁 - redission 功能介绍

Redission 入门案例

分布式锁 - redission可重入锁原理

分布式锁 - redission锁重试和WatchDog机制

分布式锁 - redission锁的MutiLock原理

Redis 秒杀优化

秒杀优化 - Redis完成秒杀资格判断

秒杀优化 - 基于阻塞队列实现秒杀优化

Redis 消息队列

消息队列

基于List结构的消息队列

基于PubSub结构的消息队列

基于Stream结构的消息队列 - 单消费模式

基于Stream结构的消息队列 - 消费者组模式

Redis项目仿大众点评

登录拦截器的优化

缓存的概述

缓存更新策略

缓存穿透

缓存雪崩

缓存击穿（热点 key）

基于互斥锁方式解决缓存击穿问题

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

缓存工具封装