Redis秒杀使用-白红宇的个人博客

Redis秒杀使用

发布日期：2022-09-29 16:46:51 浏览次数：2 分类：博客文章

本文共 60162 字，大约阅读时间需要 200 分钟。

1. 缓存穿透

客户端请求的数据在缓存中和数据库中都不存在，这样缓存永远不会生效，请求都会直接打到数据库。

1.1 解决方案

1. 缓存Null值   -> 优点:实现方便  缺点: 额外内存消耗，可能造成短期的不一致2. 布隆过滤器    -> 优点: 内存占用较少，没有多余的key    缺点: 实现复杂，存在误判的可能3. 增强id的复杂度，避免被猜测到id的规律4. 做好数据的基础格式校验5. 加强用户权限校验6. 做好热点参数的限流

2. 缓存雪崩

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

2.1 解决方案

1. 给不同的key设置不同的过期时间  -> 随机时间2. 利用redis集群提高服务的可用性  -> 利用redis哨兵机制3. 给缓存业务添加降级限流策略4. 给业务添加多级缓存

3. 缓存击穿

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

3.1 常见的解决方案

3.1.1 互斥锁

3.1.2 逻辑过期

3.2 解决方案对比

3.3 基于互斥锁的方式解决

修改根据id查询商品的接口，基于互斥锁方式来解决缓存击穿问题

3.3.1 代码

package com.hmdp.service.impl;import cn.hutool.core.util.BooleanUtil;import cn.hutool.core.util.StrUtil;import cn.hutool.json.JSONUtil;import com.hmdp.dto.Result;import com.hmdp.entity.Shop;import com.hmdp.mapper.ShopMapper;import com.hmdp.service.IShopService;import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;import com.hmdp.utils.RedisConstants;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.stereotype.Service;import org.springframework.transaction.annotation.Transactional;import javax.annotation.Resource;import java.util.concurrent.TimeUnit;@Servicepublic class ShopServiceImpl extends ServiceImpl
   
     implements IShopService {    @Resource    private StringRedisTemplate stringRedisTemplate;    /**     * 根据id查询商铺信息     * @param id     * @return     */    @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 id     * @return     */    public Shop queryWithMutex(Long id) {        String key = RedisConstants.CACHE_SHOP_KEY + id;        // 1. 从 redis 中查询商品缓存        String shopJson = stringRedisTemplate.opsForValue().get(key);        // 2. 判断是否存在        if (StrUtil.isNotBlank(shopJson)) {            // 3. 存在， 直接返回结果            return JSONUtil.toBean(shopJson, Shop.class);        }        // 判断命中的是否是空值        if(shopJson != null) {            // 返回错误信息            return null;        }        // 4. 实现缓存重建        // 4.1 获取互斥锁        String lockKey = "lock:shop:" + id;        Shop shop = null;        try {            boolean isLock = tryLock(lockKey);            // 4.2 判断是否获取成功            if(!isLock) {                // 4.3 失败，则休眠并重试                Thread.sleep(100);                return queryWithMutex(id);            }            // 4.4 成功，根据id查询数据库            shop = getById(id);            // todo 模拟重建延时            //Thread.sleep(200);            // 5. 不存在，返回错误            if (shop == null) {                // 将空值写入redis                stringRedisTemplate.opsForValue().set(key, "", RedisConstants.CACHE_NULL_TTL, TimeUnit.MINUTES);                // 返回错误信息                return null;            }            // 6. 存在，写入redis            stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop), RedisConstants.CACHE_SHOP_TTL, TimeUnit.MINUTES);        } catch (InterruptedException e) {            e.printStackTrace();        } finally {            // 7. 释放互斥锁            unlock(lockKey);        }        // 8. 返回结果        return shop;    }    /**     * 利用空值解决缓存穿透问题     * @param id     * @return     */    public Shop queryWithPassThrough(Long id) {        String key = RedisConstants.CACHE_SHOP_KEY + id;        // 1. 从 redis 中查询商品缓存        String shopJson = stringRedisTemplate.opsForValue().get(key);        // 2. 判断是否存在        if (StrUtil.isNotBlank(shopJson)) {            // 3. 存在， 直接返回结果            return JSONUtil.toBean(shopJson, Shop.class);        }        // 判断命中的是否是空值        if(shopJson != null) {            // 返回错误信息            return null;        }        // 4. 不存在，根据id查询数据库        Shop shop = getById(id);        // 5. 不存在，返回错误        if (shop == null) {            // 将空值写入redis            stringRedisTemplate.opsForValue().set(key, "", RedisConstants.CACHE_NULL_TTL, TimeUnit.MINUTES);            // 返回错误信息            return null;        }        // 6. 存在，写入redis        stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop), RedisConstants.CACHE_SHOP_TTL, TimeUnit.MINUTES);        // 7. 返回结果        return shop;    }    /**     * 尝试获取锁     * @param key     * @return     */    private boolean tryLock(String key) {        Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);        return BooleanUtil.isTrue(flag);    }    /**     * 释放锁     * @param key     */    private void unlock(String key) {        stringRedisTemplate.delete(key);    }}

3.4 基于逻辑过期方式解决

修改根据id查询商品的业务，基于逻辑过期方式来解决缓存击穿问题

3.4.1 代码

package com.hmdp.service.impl;import cn.hutool.core.util.BooleanUtil;import cn.hutool.core.util.StrUtil;import cn.hutool.json.JSONObject;import cn.hutool.json.JSONUtil;import com.hmdp.dto.Result;import com.hmdp.entity.Shop;import com.hmdp.mapper.ShopMapper;import com.hmdp.service.IShopService;import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;import com.hmdp.utils.RedisConstants;import com.hmdp.utils.RedisData;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.stereotype.Service;import org.springframework.transaction.annotation.Transactional;import javax.annotation.Resource;import java.time.LocalDateTime;import java.util.concurrent.*;@Servicepublic class ShopServiceImpl extends ServiceImpl
   
     implements IShopService {    @Resource    private StringRedisTemplate stringRedisTemplate;    /**     * 根据id查询商铺信息     * @param id     * @return     */    @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);    }    /**     * 定义线程池， 创建10个线程     */    private static final ExecutorService CACHE_REBUILD_EXECUTOR = new ThreadPoolExecutor(10, 10, 0L, TimeUnit.MILLISECONDS,            new LinkedBlockingQueue
    
     (10), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());    /**     * 使用逻辑过期解决缓存击穿     * @param id     * @return     */    public Shop queryWithLogicalExpire(Long id) {        String key = RedisConstants.CACHE_SHOP_KEY + id;        // 1. 从 redis 中查询商品缓存        String shopJson = stringRedisTemplate.opsForValue().get(key);        // 2. 判断是否存在        if (StrUtil.isBlank(shopJson)) {            // 3. 不存在， 直接返回            return null;        }        // 4. 命中，将json反序列化为对象        RedisData redisData = JSONUtil.toBean(shopJson, RedisData.class);        Shop shop = JSONUtil.toBean((JSONObject) redisData.getData(), Shop.class);        LocalDateTime expireTime = redisData.getExpireTime();        // 5. 判断逻辑过期时间是否过期        if (LocalDateTime.now().isBefore(expireTime)) {            // 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 {                    // 重建缓存                    this.saveShop2Redis(id, 20L);                } catch (Exception e) {                    throw new RuntimeException(e);                } finally {                    // 释放锁                    unlock(lockKey);                }            });        }        // 6.4 返回过期的商铺信息        // 7. 返回结果        return shop;    }       /**     * 利用空值解决缓存穿透问题     * @param id     * @return     */    public Shop queryWithPassThrough(Long id) {        String key = RedisConstants.CACHE_SHOP_KEY + id;        // 1. 从 redis 中查询商品缓存        String shopJson = stringRedisTemplate.opsForValue().get(key);        // 2. 判断是否存在        if (StrUtil.isNotBlank(shopJson)) {            // 3. 存在， 直接返回结果            return JSONUtil.toBean(shopJson, Shop.class);        }        // 判断命中的是否是空值        if(shopJson != null) {            // 返回错误信息            return null;        }        // 4. 不存在，根据id查询数据库        Shop shop = getById(id);        // 5. 不存在，返回错误        if (shop == null) {            // 将空值写入redis            stringRedisTemplate.opsForValue().set(key, "", RedisConstants.CACHE_NULL_TTL, TimeUnit.MINUTES);            // 返回错误信息            return null;        }        // 6. 存在，写入redis        stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop), RedisConstants.CACHE_SHOP_TTL, TimeUnit.MINUTES);        // 7. 返回结果        return shop;    }    /**     * 尝试获取锁     * @param key     * @return     */    private boolean tryLock(String key) {        Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);        return BooleanUtil.isTrue(flag);    }    /**     * 释放锁     * @param key     */    private void unlock(String key) {        stringRedisTemplate.delete(key);    }    /**     * 存入商铺信息附带过期时间到redis     * @param id     */    public void saveShop2Redis(Long id, Long expireSeconds) {        // 1. 查询店铺数据        Shop shop = this.getById(id);        // todo 模拟重建延时        //try {        //    Thread.sleep(200);        //} catch (InterruptedException e) {        //    e.printStackTrace();        //}        // 2. 封装逻辑过期时间        RedisData redisData = new RedisData();        redisData.setExpireTime(LocalDateTime.now().plusSeconds(expireSeconds));        redisData.setData(shop);        // 3. 写入redis        stringRedisTemplate.opsForValue()                .set(RedisConstants.CACHE_SHOP_KEY + id, JSONUtil.toJsonStr(redisData));    }}

4. 缓存工具封装

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

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

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

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

package com.hmdp.utils;import cn.hutool.core.util.BooleanUtil;import cn.hutool.core.util.StrUtil;import cn.hutool.json.JSONObject;import cn.hutool.json.JSONUtil;import com.hmdp.entity.Shop;import com.sun.xml.internal.bind.v2.model.core.ID;import lombok.extern.slf4j.Slf4j;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.stereotype.Component;import java.time.LocalDateTime;import java.util.concurrent.*;import java.util.function.Function;@Slf4j@Componentpublic class CacheClient {    private final StringRedisTemplate stringRedisTemplate;    public CacheClient(StringRedisTemplate stringRedisTemplate) {        this.stringRedisTemplate = stringRedisTemplate;    }    /**     * 将任意的java对象序列化为json并存储在String类型的key中,     * 并且可以设置TLL过期时间     * @param key 缓存key     * @param value 缓存对象     * @param time 过期时间     * @param timeUnit 时间单位     */    public void set(String key, Object value, Long time, TimeUnit timeUnit) {        stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(value), time, timeUnit);    }    /**     * 将任意的java对象序列化为json并存储在String类型的key中,     * 并且可以设置TLL过期时间，并且可以设置逻辑过期时间，用于处理缓存击穿问题     * @param key 缓存key     * @param value 缓存对象     * @param time 过期时间     * @param timeUnit 时间单位     */    public void setWithLogicalExpire(String key, Object value, Long time, TimeUnit timeUnit) {        // 设置逻辑过期时间        RedisData redisData = new RedisData();        redisData.setExpireTime(LocalDateTime.now().plusSeconds(timeUnit.toSeconds(time)));        redisData.setData(value);        // 写入 redis        stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(redisData));    }    /**     * 根据指定的key查询缓存，并反序列化为指定类型,     * 利用缓存空值的方式解决缓存穿透问题     * @param keyPrefix key前缀     * @param id 查询的id     * @param type 返回值类型     * @param dbFallBack 查询数据库的回调函数     * @param time 过期时间     * @param timeUnit 时间单位     * @param 
   
     返回值类型     * @param 
    
      id类型     * @return     */    public 
     
       R queryWithPassThrough(            String keyPrefix,            ID id,            Class
      
        type,            Function
       
         dbFallBack,            Long time,            TimeUnit timeUnit    ) {        String key = keyPrefix + id;        // 1. 从 redis 中查询缓存        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, "", RedisConstants.CACHE_NULL_TTL, TimeUnit.MINUTES);            // 返回错误信息            return null;        }        // 6. 存在，写入redis        this.set(key, r, time, timeUnit);        // 7. 返回结果        return r;    }    /**     * 定义线程池， 创建10个线程     */    private static final ExecutorService CACHE_REBUILD_EXECUTOR = new ThreadPoolExecutor(10, 10, 0L, TimeUnit.MILLISECONDS,            new LinkedBlockingQueue
        
         (10), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy()); /** * 使用逻辑过期解决缓存击穿问题 * @param keyPrefix key前缀 * @param id 查询的id * @param type 返回值类型 * @param dbFallBack 查询数据库的回调函数 * @param time 过期时间 * @param timeUnit 时间单位 * @param 
         
           返回值类型 * @param 
          
            id类型 * @return */ public 
           
             R queryWithLogicalExpire( String keyPrefix, ID id, Class
            
              type, Function
             
               dbFallBack, Long time, TimeUnit timeUnit ) { String key = keyPrefix + id; // 1. 从 redis 中查询缓存 String json = stringRedisTemplate.opsForValue().get(key); // 2. 判断是否存在 if (StrUtil.isBlank(json)) { // 3. 不存在， 直接返回 return null; } // 4. 命中，将json反序列化为对象 RedisData redisData = JSONUtil.toBean(json, RedisData.class); R r = JSONUtil.toBean((JSONObject) redisData.getData(), type); LocalDateTime expireTime = redisData.getExpireTime(); // 5. 判断逻辑过期时间是否过期 if (LocalDateTime.now().isBefore(expireTime)) { // 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 { // 重建缓存 // 1. 查询数据库 R r1 = dbFallBack.apply(id); // 2. 写入redis this.setWithLogicalExpire(key, r1, time, timeUnit); } catch (Exception e) { throw new RuntimeException(e); } finally { // 释放锁 unlock(lockKey); } }); } // 7. 返回结果 return r; } /** * 尝试获取锁 * @param key * @return */ private boolean tryLock(String key) { Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS); return BooleanUtil.isTrue(flag); } /** * 释放锁 * @param key */ private void unlock(String key) { stringRedisTemplate.delete(key); }}

5. 全局ID生成器

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

5.1 工具类

package com.hmdp.utils;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.stereotype.Component;import java.time.LocalDateTime;import java.time.ZoneOffset;import java.time.format.DateTimeFormatter;/** * @Description: Redis id生成器 * @Date: 2022/5/7 13:00 */@Componentpublic class RedisIdWorker {    /**     * 开始时间戳     */    public static final long BEGIN_TIMESTAMP = 1640995200L;    /**     * 序列号位数     */    public static final int COUNT_BITS = 32;    private StringRedisTemplate stringRedisTemplate;    public RedisIdWorker(StringRedisTemplate stringRedisTemplate) {        this.stringRedisTemplate = stringRedisTemplate;    }    /**     * 唯一id生成     * @param keyPrefix 不同业务的key前缀     * @return     */    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;    }}

5.2 总结

5.2.1 全局唯一ID生成策略

1. UUID2. Redis自增3. snowflack算法4. 数据库自增

5.2.2 Redis自增ID策略

1. 每天一个key，方便统计订单量

6. 优惠券秒杀

判断两点:

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

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

6.1 普通抢购优惠券的代码

/** * 抢购优惠券 * @param voucherId * @return */@Override@Transactionalpublic Result seckillVoucher(Long voucherId) {    // 1. 查询优惠券    SeckillVoucher voucher = seckillVoucherService.getById(voucherId);    // 2. 判断秒杀是否开始    if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {        // 秒杀未开始        return Result.fail("秒杀尚未开始");    }    // 3. 判断秒杀是否结束    if (voucher.getEndTime().isBefore(LocalDateTime.now())) {        // 秒杀已经结束        return Result.fail("秒杀已经结束");    }    // 4. 判断库存是否充足    if (voucher.getStock() < 1) {        // 库存不足        return Result.fail("库存不足");    }    // 5. 扣减库存    boolean success = seckillVoucherService.update(new LambdaUpdateWrapper
   
    ()                                                   .eq(SeckillVoucher::getVoucherId, voucherId)                                                   .set(SeckillVoucher::getStock, voucher.getStock() - 1));    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.setUserId(userId);    // 6.3 优惠券id    voucherOrder.setVoucherId(voucherId);    // 存库    this.save(voucherOrder);    // 7. 返回订单id    return Result.ok(orderId);}

6.2 出现超卖问题

超卖问题就是典型的多线程安全问题

6.3 常见解决方案

加锁

6.3.1 乐观锁解决思路

乐观锁的关键是判断之前查询得到的数据是否有被修改过。常见的方式有两种:

6.3.2 悲观锁

略，加synchronized 或者 Lock即可

6.4 乐观锁 (CAS)解决

package com.hmdp.service.impl;import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;import com.hmdp.dto.Result;import com.hmdp.entity.SeckillVoucher;import com.hmdp.entity.VoucherOrder;import com.hmdp.mapper.VoucherOrderMapper;import com.hmdp.service.ISeckillVoucherService;import com.hmdp.service.IVoucherOrderService;import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;import com.hmdp.utils.RedisIdWorker;import com.hmdp.utils.UserHolder;import org.springframework.stereotype.Service;import org.springframework.transaction.annotation.Transactional;import javax.annotation.Resource;import java.time.LocalDateTime;@Servicepublic class VoucherOrderServiceImpl extends ServiceImpl
   
     implements IVoucherOrderService {    @Resource    private ISeckillVoucherService seckillVoucherService;    @Resource    private RedisIdWorker redisIdWorker;    /**     * 抢购优惠券     * @param voucherId     * @return     */    @Override    @Transactional    public Result seckillVoucher(Long voucherId) {        // 1. 查询优惠券        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);        // 2. 判断秒杀是否开始        if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {            // 秒杀未开始            return Result.fail("秒杀尚未开始");        }        // 3. 判断秒杀是否结束        if (voucher.getEndTime().isBefore(LocalDateTime.now())) {            // 秒杀已经结束            return Result.fail("秒杀已经结束");        }        // 4. 判断库存是否充足        if (voucher.getStock() < 1) {            // 库存不足            return Result.fail("库存不足");        }        // 5. 扣减库存        boolean success = seckillVoucherService.update(new LambdaUpdateWrapper
    
     ()                .eq(SeckillVoucher::getVoucherId, voucherId)                // 利用CAS 解决超卖问题   where id = ? and stock = ?                //.eq(SeckillVoucher::getStock, voucher.getStock())                // 利用CAS 解决超卖问题   where id = ? and stock > 0                .gt(SeckillVoucher::getStock, 0)                .setSql("stock = stock - 1"));        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.setUserId(userId);        // 6.3 优惠券id        voucherOrder.setVoucherId(voucherId);        // 存库        this.save(voucherOrder);        // 7. 返回订单id        return Result.ok(orderId);    }}

6.5 总结

解决方案

1. 悲观锁: 添加同步锁，让线程串行执行 - 优点: 简单粗暴 - 缺点: 性能一般2. 乐观锁：不加锁，在更新时判断是否有其他线程在修改 => 通过数据库层面解决 => InnoDB存储引擎通过多版本并发控制（MVCC，Multiversion Concurrency Control）机制解决并发修改问题。 - 优点: 性能好 - 缺点: 存在成功率低的问题       对数据库的压力很大

6.6 一人一单

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

6.6.1 解决一个人只能下一单（单机）

package com.hmdp.service.impl;import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;import com.hmdp.dto.Result;import com.hmdp.entity.SeckillVoucher;import com.hmdp.entity.VoucherOrder;import com.hmdp.mapper.VoucherOrderMapper;import com.hmdp.service.ISeckillVoucherService;import com.hmdp.service.IVoucherOrderService;import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;import com.hmdp.utils.RedisIdWorker;import com.hmdp.utils.UserHolder;import org.springframework.stereotype.Service;import org.springframework.transaction.annotation.Transactional;import javax.annotation.Resource;import java.time.LocalDateTime;@Servicepublic class VoucherOrderServiceImpl extends ServiceImpl
   
     implements IVoucherOrderService {    @Resource    private ISeckillVoucherService seckillVoucherService;    @Resource    private RedisIdWorker redisIdWorker;    /**     * 抢购优惠券     *     * @param voucherId     * @return     */    @Override    public Result seckillVoucher(Long voucherId) {        // 1. 查询优惠券        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);        // 2. 判断秒杀是否开始        if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {            // 秒杀未开始            return Result.fail("秒杀尚未开始");        }        // 3. 判断秒杀是否结束        if (voucher.getEndTime().isBefore(LocalDateTime.now())) {            // 秒杀已经结束            return Result.fail("秒杀已经结束");        }        // 4. 判断库存是否充足        if (voucher.getStock() < 1) {            // 库存不足            return Result.fail("库存不足");        }        return createVoucherOrder(voucherId);    }    /**     * 创建订单     * @param voucherId     * @return     */    @Transactional    public Result createVoucherOrder(Long voucherId) {        // 5. 一人只能抢一份        Long userId = UserHolder.getUser().getId();        synchronized (userId.toString().intern()) {            // 5.1 查询订单            long count = this.count(new LambdaQueryWrapper
    
     ()                    .eq(VoucherOrder::getUserId, userId)                    .eq(VoucherOrder::getVoucherId, voucherId));            // 5.2 判断是否已经抢过            if (count > 0) {                // 用户已经购买过                return Result.fail("用户已经购买过一次");            }            // 6. 扣减库存            boolean success = seckillVoucherService.update(new LambdaUpdateWrapper
     
      ()                    .eq(SeckillVoucher::getVoucherId, voucherId)                    // 利用CAS 解决超卖问题   where id = ? and stock = ?                    //.eq(SeckillVoucher::getStock, voucher.getStock())                    // 利用CAS 解决超卖问题   where id = ? and stock > 0                    .gt(SeckillVoucher::getStock, 0)                    .setSql("stock = stock - 1"));            if (!success) {                // 扣减库存失败                return Result.fail("扣减库存失败");            }            // 7. 创建订单            VoucherOrder voucherOrder = new VoucherOrder();            // 7.1 订单id            long orderId = redisIdWorker.nextId("order");            voucherOrder.setId(orderId);            // 7.2 用户id            voucherOrder.setUserId(userId);            // 7.3 优惠券id            voucherOrder.setVoucherId(voucherId);            // 存库            this.save(voucherOrder);            // 8. 返回订单id            return Result.ok(orderId);        }    }}

6.6.2 一人一单的并发安全问题（集群）

通过加锁可以解决在单机情况下的一人一单安全问题，但是在集群模式下就不行了.

6.6.2.1 将服务启动两份，端口分别为 8081 和 8082

6.6.2.2 修改`nginx`的`nginx.conf`文件，配置反向代理和负载均衡

然后执行 ./nginx -s reload

现在，用户的请求会在这两个节点上负载均衡

6.6.3 集群下一人多单并发问题

如果一个用户同一秒类，发送两次以上请求，请求都会进入到创建订单业务代码中，导致一个人可以下多个订单的并发问题

出现问题的原理图:

8081:

8082:

导致一人多单的问题

6.6.4 解决方案 -> 分布式锁

7. 分布式锁

7.1 分布式锁原理

7.1.1 什么是分布式锁

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

7.1.2 分布式锁的实现

分布式锁的核心是实现多进程之间互斥，而满足这一点的方式有很多，常见的有三种:

7.2 基于Redis的分布式锁

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

7.2.1 获取锁

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

# 添加锁，EX是设置超时间、NX如果不存在才执行(互斥)SET lock thread1 EX 10 NX

7.2.2 释放锁

- 手动释放- 超时释放: 获取锁时添加一个超时时间

# 释放锁，删除即可DEL key

7.2.3 【初级】分布式锁

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

package com.hmdp.utils;/** * @author codertl */public interface ILock {    /**     * 尝试获取锁     * @param timeoutSeconds 锁持有的超时时间，过期后自动释放锁     * @return true代表获取锁成功，false代表获取锁失败     */    boolean tryLock(long timeoutSeconds);    /**     * 释放锁     */    void unlock();}

实现类:

package com.hmdp.utils;import org.springframework.data.redis.core.StringRedisTemplate;import java.util.concurrent.TimeUnit;public class SimpleRedisLock implements ILock{    /**     * 锁的key名字     */    private String name;    private StringRedisTemplate stringRedisTemplate;    /**     * 锁的前缀     */    public static final String KEY_PREFIX = "lock:";    public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {        this.name = name;        this.stringRedisTemplate = stringRedisTemplate;    }    /**     * 尝试获取锁     * @param timeoutSeconds 锁持有的超时时间，过期后自动释放锁     * @return true代表获取锁成功，false代表获取锁失败     */    @Override    public boolean tryLock(long timeoutSeconds) {        // 获取线程标识        long threadId = Thread.currentThread().getId();        // 获取锁        Boolean success = stringRedisTemplate.opsForValue()                .setIfAbsent(KEY_PREFIX + name, threadId + "", timeoutSeconds, TimeUnit.SECONDS);        return Boolean.TRUE.equals(success);    }    /**     * 释放锁     */    @Override    public void unlock() {        // 释放锁完成        stringRedisTemplate.delete(KEY_PREFIX + name);    }}

修改创建订单的逻辑:

/** * 创建订单 * @param voucherId * @return */@Transactionalpublic Result createVoucherOrder(Long voucherId) {    // 5. 一人只能抢一份    Long userId = UserHolder.getUser().getId();    // 创建锁对象    SimpleRedisLock redisLock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);    // 尝试获取锁    boolean isLock = redisLock.tryLock(1200);    // 判断是否获取锁    if (!isLock) {        // 获取锁失败，直接返回失败/重试        return Result.fail("不允许重复抢购!!");    }    try {        // 5.1 查询订单        long count = this.count(new LambdaQueryWrapper
   
    ()                                .eq(VoucherOrder::getUserId, userId)                                .eq(VoucherOrder::getVoucherId, voucherId));        // 5.2 判断是否已经抢过        if (count > 0) {            // 用户已经购买过            return Result.fail("用户已经购买过一次");        }        // 6. 扣减库存        boolean success = seckillVoucherService.update(new LambdaUpdateWrapper
    
     ()                                                       .eq(SeckillVoucher::getVoucherId, voucherId)                                                       // 利用CAS 解决超卖问题   where id = ? and stock = ?                                                       //.eq(SeckillVoucher::getStock, voucher.getStock())                                                       // 利用CAS 解决超卖问题   where id = ? and stock > 0                                                       .gt(SeckillVoucher::getStock, 0)                                                       .setSql("stock = stock - 1"));        if (!success) {            // 扣减库存失败            return Result.fail("扣减库存失败");        }        // 7. 创建订单        VoucherOrder voucherOrder = new VoucherOrder();        // 7.1 订单id        long orderId = redisIdWorker.nextId("order");        voucherOrder.setId(orderId);        // 7.2 用户id        voucherOrder.setUserId(userId);        // 7.3 优惠券id        voucherOrder.setVoucherId(voucherId);        // 存库        this.save(voucherOrder);        // 8. 返回订单id        return Result.ok(orderId);    } finally {        // 释放锁        redisLock.unlock();    }}

7.2.4 【初级】分布式锁存在问题

业务阻塞导致锁被超时自动释放

基于上面的情况，当线程1业务执行完成之后，释放锁，可能会释放到其他线程的锁

解决方法

业务逻辑解决流程:

7.2.5 解决分布式锁->【误删】问题

修改之前的分布式锁实现

在获取锁时，存入线程标识(可以用UUID标识)

在释放锁时，先获取锁中的线程标识，判断是否与当前线程标识一致

如果一致则释放锁

如果不一致则不释放锁

package com.hmdp.utils;import cn.hutool.core.lang.UUID;import org.springframework.data.redis.core.StringRedisTemplate;import java.util.concurrent.TimeUnit;public class SimpleRedisLock implements ILock{    /**     * 锁的key名字     */    private String name;    private StringRedisTemplate stringRedisTemplate;    /**     * 锁的前缀     */    public static final String KEY_PREFIX = "lock:";    /**     * 锁的内容前缀     */    public static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";    public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {        this.name = name;        this.stringRedisTemplate = stringRedisTemplate;    }    /**     * 尝试获取锁     * @param timeoutSeconds 锁持有的超时时间，过期后自动释放锁     * @return true代表获取锁成功，false代表获取锁失败     */    @Override    public boolean tryLock(long timeoutSeconds) {        // 获取线程标识        String threadId = ID_PREFIX + Thread.currentThread().getId();        // 获取锁        Boolean success = stringRedisTemplate.opsForValue()                .setIfAbsent(KEY_PREFIX + name, threadId, timeoutSeconds, 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);        }    }}

7.2.6 分布式锁的原子性问题

【注】：当前依旧存在问题，判断锁标识和释放锁，并不是原子性操作。

1. 如果线程1当判断锁成功后，在释放锁之前阻塞，然后锁超时释放了。

2. 这个时候，别的线程获取到锁，当前的线程1阻塞结束，进行释放锁，依旧会存在锁误删的问题

7.2.7 Lua脚本解决多条命令的原子性问题

Redis提供了Lua脚本功能，在一个脚本中编写多条redis命令，确保多条命令执行时的原子性。

Lua是一种编程语言，它的语法可以参考:

- 这里介绍Redis提供的调用函数，语法如下:

-- 执行redis命令redis.call('命令名称', 'key', '其他参数', ...)

例如:

-- 执行 set name jackredis.class('set', 'name', 'jack')

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

-- 先执行 set name jackredis.call('set', 'name', 'jack')-- 再执行 get namelocal name = redis.call('get', 'name')-- 返回return name

写好脚本之后，需要使用Redis命令来调用脚本，调用脚本的常见命令如下:

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

# 调用脚本  ""当中为脚本内容  0为脚本需要的key类型的参数个数EVAL "return redis.call('set', 'name', 'jack')" 0

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

# 调用脚本 KEYS[1] => name  ARGV[1] => Rose   1为脚本需要的key类型的参数个数EVAL "return redis.call('set', KEYS[1], ARGV[1])" 1 name Rose

7.2.7.1 基于Redis的分布式锁的Lua脚本

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

7.2.8 Java调用Lua脚本改造分布式锁

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

修改释放锁的实现

脚本内容在resource目录下新建unlock.lua文件:

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

java代码:

package com.hmdp.utils;import cn.hutool.core.lang.UUID;import org.springframework.core.io.ClassPathResource;import org.springframework.data.redis.core.RedisCallback;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.data.redis.core.script.DefaultRedisScript;import org.springframework.data.redis.core.script.RedisScript;import java.util.Collections;import java.util.concurrent.TimeUnit;/** * @Author: tl * @Description: * @Date: 2022/5/8 */public class SimpleRedisLock implements ILock{    /**     * 锁的key名字     */    private String name;    private StringRedisTemplate stringRedisTemplate;    /**     * 锁的前缀     */    public static final String KEY_PREFIX = "lock:";    /**     * 锁的内容前缀     */    public static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";    /**     * 加载Lua脚本     */    public static final DefaultRedisScript
   
     UNLOCK_SCRIPT;    static {        UNLOCK_SCRIPT = new DefaultRedisScript<>();        // 脚本文件位置        UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));        // 脚本返回值        UNLOCK_SCRIPT.setResultType(Long.class);    }    public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {        this.name = name;        this.stringRedisTemplate = stringRedisTemplate;    }    /**     * 尝试获取锁     * @param timeoutSeconds 锁持有的超时时间，过期后自动释放锁     * @return true代表获取锁成功，false代表获取锁失败     */    @Override    public boolean tryLock(long timeoutSeconds) {        // 获取线程标识        String threadId = ID_PREFIX + Thread.currentThread().getId();        // 获取锁        Boolean success = stringRedisTemplate.opsForValue()                .setIfAbsent(KEY_PREFIX + name, threadId, timeoutSeconds, TimeUnit.SECONDS);        return Boolean.TRUE.equals(success);    }    /**     * 释放锁     */    @Override    public void unlock() {       // 调用 Lua 脚本释放锁       stringRedisTemplate.execute(               UNLOCK_SCRIPT,               Collections.singletonList(KEY_PREFIX + name),               ID_PREFIX + Thread.currentThread().getId()       );    }}

7.2.9 总结

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

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

特性

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

7.3 Redisson实现分布式锁

基于 set nx 实现的分布式锁存在下面的问题:

7.3.1 概述

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

官方网址:

GitHub地址:

7.3.2 Redisson快速入门

1.引入依赖


   
       
    
     org.redisson
        
    
     redisson
        
    
     3.14.0

2.配置Redisson客户端

package com.hmdp.config;import org.redisson.Redisson;import org.redisson.api.RedissonClient;import org.redisson.config.Config;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;@Configurationpublic class RedissonConfig {    @Bean    public RedissonClient redissonClient() {        // 配置类        Config config = new Config();        // 添加redis地址，这里添加了单点的地址，也可以使用 config.useClusterServers() 方法添加集群模式的地址        config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123345");        // 创建RedissonClient对象        return Redisson.create(config);    }}

3.使用Redisson的分布式锁

package com.hmdp.service.impl;import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;import com.hmdp.dto.Result;import com.hmdp.entity.SeckillVoucher;import com.hmdp.entity.VoucherOrder;import com.hmdp.mapper.VoucherOrderMapper;import com.hmdp.service.ISeckillVoucherService;import com.hmdp.service.IVoucherOrderService;import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;import com.hmdp.utils.RedisIdWorker;import com.hmdp.utils.SimpleRedisLock;import com.hmdp.utils.UserHolder;import org.redisson.api.RLock;import org.redisson.api.RedissonClient;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.stereotype.Service;import org.springframework.transaction.annotation.Transactional;import javax.annotation.Resource;import java.time.LocalDateTime;@Servicepublic class VoucherOrderServiceImpl extends ServiceImpl
   
     implements IVoucherOrderService {    @Resource    private ISeckillVoucherService seckillVoucherService;    @Resource    private RedisIdWorker redisIdWorker;    @Resource    private RedissonClient redissonClient;    /**     * 抢购优惠券     *     * @param voucherId     * @return     */    @Override    public Result seckillVoucher(Long voucherId) {        // 1. 查询优惠券        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);        // 2. 判断秒杀是否开始        if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {            // 秒杀未开始            return Result.fail("秒杀尚未开始");        }        // 3. 判断秒杀是否结束        if (voucher.getEndTime().isBefore(LocalDateTime.now())) {            // 秒杀已经结束            return Result.fail("秒杀已经结束");        }        // 4. 判断库存是否充足        if (voucher.getStock() < 1) {            // 库存不足            return Result.fail("库存不足");        }        return createVoucherOrder(voucherId);    }    /**     * 创建订单     * @param voucherId     * @return     */    @Transactional    public Result createVoucherOrder(Long voucherId) {        // 5. 一人只能抢一份        Long userId = UserHolder.getUser().getId();        // 使用Redisson创建锁对象        RLock redisLock = redissonClient.getLock("lock:order:" + userId);        // 使用Redisson尝试获取锁  参数1: 获取锁的最大等待时间(期间会重试) 参数2: 锁自动释放的事件 参数3：时间单位        boolean isLock = redisLock.tryLock();        // 判断是否获取锁        if (!isLock) {            // 获取锁失败，直接返回失败/重试            return Result.fail("不允许重复抢购!!");        }        try {            // 5.1 查询订单            long count = this.count(new LambdaQueryWrapper
    
     ()                    .eq(VoucherOrder::getUserId, userId)                    .eq(VoucherOrder::getVoucherId, voucherId));            // 5.2 判断是否已经抢过            if (count > 0) {                // 用户已经购买过                return Result.fail("用户已经购买过一次");            }            // 6. 扣减库存            boolean success = seckillVoucherService.update(new LambdaUpdateWrapper
     
      ()                    .eq(SeckillVoucher::getVoucherId, voucherId)                    // 利用CAS 解决超卖问题   where id = ? and stock = ?                    //.eq(SeckillVoucher::getStock, voucher.getStock())                    // 利用CAS 解决超卖问题   where id = ? and stock > 0                    .gt(SeckillVoucher::getStock, 0)                    .setSql("stock = stock - 1"));            if (!success) {                // 扣减库存失败                return Result.fail("扣减库存失败");            }            // 7. 创建订单            VoucherOrder voucherOrder = new VoucherOrder();            // 7.1 订单id            long orderId = redisIdWorker.nextId("order");            voucherOrder.setId(orderId);            // 7.2 用户id            voucherOrder.setUserId(userId);            // 7.3 优惠券id            voucherOrder.setVoucherId(voucherId);            // 存库            this.save(voucherOrder);            // 8. 返回订单id            return Result.ok(orderId);        } finally {            // 使用Redisson释放锁            redisLock.unlock();        }    }}

7.3.3 Redisson可重入锁原理

根据重入次数来记录可重入锁

核心: 利用redis的Hash结构，记录获取锁的线程以及记录的次数

package com.hmdp;import lombok.extern.slf4j.Slf4j;import org.junit.jupiter.api.BeforeEach;import org.junit.jupiter.api.Test;import org.redisson.api.RLock;import org.redisson.api.RedissonClient;import org.springframework.boot.test.context.SpringBootTest;import javax.annotation.Resource;@Slf4j@SpringBootTestpublic class RedissonTest {    @Resource    private RedissonClient redissonClient;    private RLock lock;    @BeforeEach    void setUp() {        lock = redissonClient.getLock("order");    }    @Test    void method1() {        // 尝试获取锁        boolean isLock = lock.tryLock();        if (!isLock) {            log.error("获取锁失败 ....1");            return;        }        try {            log.info("获取锁成功 ....1");            // 执行业务代码            method2();            log.info("执行业务代码 ....1");        } finally {            log.warn("准备释放锁 ....1");            // 释放锁            lock.unlock();        }    }    @Test    void method2() {        // 尝试获取锁        boolean isLock = lock.tryLock();        if (!isLock) {            log.error("获取锁失败 ....2");            return;        }        try {            log.info("获取锁成功 ....2");            // 执行业务代码            log.info("执行业务代码 ....2");        } finally {            log.warn("准备释放锁 ....2");            // 释放锁            lock.unlock();        }    }}

7.3.3.1 Redisson获取锁核心源码

7.3.3.2 Redisson释放锁核心源码

7.3.4 Redisson的锁重试和WatchDog机制

7.3.4.1总结

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

7.3.5 Redisson解决分布式锁主从一致性问题

使用multiLock 联锁解决分布式锁主从一致性问题

案例:我这里启动了三个redis节点

配置:

package com.hmdp.config;import org.redisson.Redisson;import org.redisson.api.RedissonClient;import org.redisson.config.Config;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;/** * @Author: tl * @Description: * @Date: 2022/5/8 */@Configurationpublic class RedissonConfig {    @Bean    public RedissonClient redissonClient() {        // 配置类        Config config = new Config();        // 添加redis地址，这里添加了单点的地址，也可以使用 config.useClusterServers() 方法添加集群模式的地址        config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("123345");        // 创建RedissonClient对象        return Redisson.create(config);    }    @Bean    public RedissonClient redissonClient2() {        // 配置类        Config config = new Config();        // 添加redis地址，这里添加了单点的地址，也可以使用 config.useClusterServers() 方法添加集群模式的地址        config.useSingleServer().setAddress("redis://127.0.0.1:6380").setPassword("123345");        // 创建RedissonClient对象        return Redisson.create(config);    }    @Bean    public RedissonClient redissonClient3() {        // 配置类        Config config = new Config();        // 添加redis地址，这里添加了单点的地址，也可以使用 config.useClusterServers() 方法添加集群模式的地址        config.useSingleServer().setAddress("redis://127.0.0.1:6381").setPassword("123345");        // 创建RedissonClient对象        return Redisson.create(config);    }}

7.3.5.1 总结

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

8. Redis优化秒杀

8.1 改进秒杀业务

提高并发性能:

需求:

新增秒杀优惠券的同时，将优惠券信息保存到Redis中

基于Lua脚本，判断秒杀库存、一人一单，决定用户是否抢购成功

如果抢购成功，将优惠券id和用户id封装后存入阻塞队列

开启线程任务，不断从阻塞队列中获取信息，实现异步下单功能

8.1.2 Lua脚本完成秒杀资格判断

resource文件夹下创建 Seckill.lua

-- 1. 参数列表-- 1.1 优惠券idlocal voucherId = ARGV[1]-- 1.2 用户idlocal userId = ARGV[2]-- 2. 数据key-- 2.1 库存keylocal stockKey = 'seckill:stock:' .. voucherId-- 2.2 订单keylocal orderKey = 'seckill:order:' .. voucherId-- 3. 脚本业务-- 3.1 判断库存是否足够  get stockKeyif(tonumber(redis.call('get', stockKey)) <= 0)then    -- 3.2 库存不足，返回1    return 1end-- 3.2 判断用户是否已经下单 SISMEMBER orderKey userIdif(redis.call('sismember', orderKey, userId) == 1)then    -- 3.3 重复下单，返回2    return 2end-- 3.4 扣除库存 incrby stockKey -1redis.call('incrby', stockKey, -1)-- 3.5 下单(保存用户) sadd orderKey userIdredis.call('sadd', orderKey, userId)return 0

业务代码:

package com.hmdp.service.impl;import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;import com.hmdp.dto.Result;import com.hmdp.entity.SeckillVoucher;import com.hmdp.entity.VoucherOrder;import com.hmdp.mapper.VoucherOrderMapper;import com.hmdp.service.ISeckillVoucherService;import com.hmdp.service.IVoucherOrderService;import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;import com.hmdp.utils.RedisIdWorker;import com.hmdp.utils.UserHolder;import org.redisson.api.RLock;import org.redisson.api.RedissonClient;import org.springframework.core.io.ClassPathResource;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.data.redis.core.script.DefaultRedisScript;import org.springframework.stereotype.Service;import org.springframework.transaction.annotation.Transactional;import javax.annotation.Resource;import java.util.Collections;@Servicepublic class VoucherOrderServiceImpl extends ServiceImpl
   
     implements IVoucherOrderService {    @Resource    private ISeckillVoucherService seckillVoucherService;    @Resource    private RedisIdWorker redisIdWorker;    @Resource    private RedissonClient redissonClient;    @Resource    private StringRedisTemplate stringRedisTemplate;    /**     * 加载Lua脚本     */    public static final DefaultRedisScript
    
      SECKILL_SCRIPT;    static {        SECKILL_SCRIPT = new DefaultRedisScript<>();        // 脚本文件位置        SECKILL_SCRIPT.setLocation(new ClassPathResource("Seckill.lua"));        // 脚本返回值        SECKILL_SCRIPT.setResultType(Long.class);    }    /**     * 抢购优惠券     *     * @param voucherId     * @return     */    @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. 判断结果是否为0        // 2.1 不为0，代表没有购买资格        if (r != 0) {            return Result.fail(r == 1 ? "库存不足" : "不能重复下单");        }        // 2.2 为0，有购买资格，把下单信息保存到阻塞队列中        long orderId = redisIdWorker.nextId("order");        // TODO: 2022/5/9 这里需要保存到阻塞队列        // 3. 返回订单id        return Result.ok(orderId);    }}

8.1.3 抢购成功，将优惠券id和用户id存入阻塞队列

private BlockingQueue
   
     orderTasks = new ArrayBlockingQueue<>(1024 * 1024);/** * 抢购优惠券 * * @param voucherId * @return */@Overridepublic 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. 判断结果是否为0    // 2.1 不为0，代表没有购买资格    if (r != 0) {        return Result.fail(r == 1 ? "库存不足" : "不能重复下单");    }    // 2.2 为0，有购买资格，把下单信息保存到阻塞队列中    VoucherOrder voucherOrder = new VoucherOrder();    // 2.3 订单id    long orderId = redisIdWorker.nextId("order");    voucherOrder.setId(orderId);    // 2.4 用户id    voucherOrder.setUserId(userId);    // 2.5 优惠券id    voucherOrder.setVoucherId(voucherId);    // 2.6 放入阻塞队列    orderTasks.add(voucherOrder);    // 3. 返回订单id    return Result.ok(orderId);}

8.1.4 开启线程任务，不断从阻塞队列中获取信息，实现异步下单功能

package com.hmdp.service.impl;import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;import com.hmdp.dto.Result;import com.hmdp.entity.SeckillVoucher;import com.hmdp.entity.VoucherOrder;import com.hmdp.mapper.VoucherOrderMapper;import com.hmdp.service.ISeckillVoucherService;import com.hmdp.service.IVoucherOrderService;import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;import com.hmdp.utils.RedisIdWorker;import com.hmdp.utils.UserHolder;import lombok.extern.slf4j.Slf4j;import org.redisson.api.RLock;import org.redisson.api.RedissonClient;import org.springframework.core.io.ClassPathResource;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.data.redis.core.script.DefaultRedisScript;import org.springframework.stereotype.Service;import org.springframework.transaction.annotation.Transactional;import javax.annotation.PostConstruct;import javax.annotation.Resource;import java.util.Collections;import java.util.concurrent.*;@Slf4j@Service@Transactional(rollbackFor = Exception.class)public class VoucherOrderServiceImpl extends ServiceImpl
   
     implements IVoucherOrderService {    @Resource    private ISeckillVoucherService seckillVoucherService;    @Resource    private RedisIdWorker redisIdWorker;    @Resource    private RedissonClient redissonClient;    @Resource    private StringRedisTemplate stringRedisTemplate;    /**     * 加载Lua脚本     */    public static final DefaultRedisScript
    
      SECKILL_SCRIPT;    static {        SECKILL_SCRIPT = new DefaultRedisScript<>();        // 脚本文件位置        SECKILL_SCRIPT.setLocation(new ClassPathResource("Seckill.lua"));        // 脚本返回值        SECKILL_SCRIPT.setResultType(Long.class);    }    private BlockingQueue
     
       orderTasks = new ArrayBlockingQueue<>(1024 * 1024);    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();    @PostConstruct    private void init() {        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());    }    private class VoucherOrderHandler implements Runnable {        @Override        public void run() {            while (true) {                try {                    // 1. 获取队列中的订单信息                    VoucherOrder voucherOrder = orderTasks.take();                    // 2. 创建订单                    createVoucherOrder(voucherOrder);                } catch (InterruptedException e) {                    log.error("处理订单异常", e);                }            }        }    }    /**     * 创建订单     * @param voucherOrder     * @return     */    public void createVoucherOrder(VoucherOrder voucherOrder) {        // 5. 一人只能抢一份        Long userId = voucherOrder.getUserId();        Long voucherId = voucherOrder.getVoucherId();        // 使用Redisson创建锁对象        RLock redisLock = redissonClient.getLock("lock:order:" + userId);        // 使用Redisson尝试获取锁        boolean isLock = redisLock.tryLock();        // 判断是否获取锁        if (!isLock) {            // 获取锁失败，直接返回失败/重试            log.error("不允许重复抢购!!");            return;        }        try {            // 5.1 查询订单            long count = this.count(new LambdaQueryWrapper
      
       ()                                    .eq(VoucherOrder::getUserId, userId)                                    .eq(VoucherOrder::getVoucherId, voucherId));            // 5.2 判断是否已经抢过            if (count > 0) {                // 用户已经购买过                log.error("用户已经购买过!!");                return;            }            // 6. 扣减库存            boolean success = seckillVoucherService.update(new LambdaUpdateWrapper
       
        ()                                                           .eq(SeckillVoucher::getVoucherId, voucherId)                                                           .gt(SeckillVoucher::getStock, 0)                                                           .setSql("stock = stock - 1"));            if (!success) {                // 扣减库存失败                log.error("扣减库存失败!!");                return ;            }            // 存库            this.save(voucherOrder);        } finally {            // 使用Redisson释放锁            redisLock.unlock();        }    }    /**     * 抢购优惠券     *     * @param voucherId     * @return     */    @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. 判断结果是否为0        // 2.1 不为0，代表没有购买资格        if (r != 0) {            return Result.fail(r == 1 ? "库存不足" : "不能重复下单");        }        // 2.2 为0，有购买资格，把下单信息保存到阻塞队列中        VoucherOrder voucherOrder = new VoucherOrder();        // 2.3 订单id        long orderId = redisIdWorker.nextId("order");        voucherOrder.setId(orderId);        // 2.4 用户id        voucherOrder.setUserId(userId);        // 2.5 优惠券id        voucherOrder.setVoucherId(voucherId);        // 2.6 放入阻塞队列        orderTasks.add(voucherOrder);        // 3. 返回订单id        return Result.ok(orderId);    }}

8.1.5 总结

优化思路

先利用Redis完成库存余量、一人一单判断，完成抢单业务

再讲下单业务放入阻塞队列，利用独立线程异步下单

基于阻塞队列的异步秒杀存在哪些问题?

内存限制问题

数据安全问题

9. Redis消息队列实现异步秒杀

9.1 什么是消息队列

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

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

9.2 基于List结构模拟消息队列

消息队列(Message Queue)，字面意思就是存放消息的队列。而Redis的List数据结构是一个双向链表，很容以模拟出队列效果。

队列：是入口和出口不在一遍，因此可以利用: LPUSH结合RPOP、或者RPUSH结合LPOP来实现

要注意的是，当队列中没有消息时 RPOP或LPOP操作会返回null，并不像JVM的阻塞队列那样会阻塞并等待消息。

因此这里应该使用 BRPOP 或者 BLPOP 来实现阻塞效果。

9.2.1基于list的消息队列的优缺点

9.2.1.1 优点

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

9.2.1.2 缺点

- 无法避免消息丢失- 只支持单消费者

9.4 基于PubSub的消息队列

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

- SubScribe channel [channel] ： 订阅一个或多个频道- Publish channel msg : 向一个频道发送消息- PsubScribe pattern[pattern] : 订阅与pattern格式匹配的所有频道

9.4.1 基于PubSub的优缺点

9.4.1.1 优点

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

9.4.1.2 缺点

- 不支持数据持久化- 无法避免消息丢失- 消息堆积有上限，超出时数据丢失

9.6 基于Stream的消息队列=>(单消费模式)

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

发送消息的命令 xadd:

读取消息的方式之一 xread:

例如: 使用xread读取第一个消息

9.6.1 Xread阻塞方式，读取消息

在业务开发中，我们可以循环的调用Xread阻塞方式来查询最新消息，从而实现持续监听队列的效果，伪代码如下:

9.6.2 Stream类型消息队列的Xread命令特点

- 消息可回溯- 一条消息可以被多个消费者读取- 可以阻塞读取- 有消息漏读的风险

9.7 基于Stream的消息队列=>(消费组模式)

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

创建消费者组:

其他常见命令

从消费者组读取消息

9.7.1 消费者监听消息的基本思路

9.7.2 XREADGROUP命令特点

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

9.8 Redis消息队列

9.9 基于Redis的Stream结构作为消息队列，实现异步秒杀下单

需求:

创建一个 Stream类型的消息队列命名为stream.orders

修改之前的秒杀下单Lua脚本，在认定有抢购资格后，直接向stream.orders中添加消息，内容包含voucherId、userId、orderId

项目启动时，开启一个线程任务，尝试获取stream.orders中的消息，完成下单

9.9.1 创建stream.orders

9.9.2 修改lua脚本和抢购优惠券的逻辑

-- 1. 参数列表-- 1.1 优惠券idlocal voucherId = ARGV[1]-- 1.2 用户idlocal userId = ARGV[2]-- 1.3 订单idlocal orderId = ARGV[3]-- 2. 数据key-- 2.1 库存keylocal stockKey = 'seckill:stock:' .. voucherId-- 2.2 订单keylocal orderKey = 'seckill:order:' .. voucherId-- 3. 脚本业务-- 3.1 判断库存是否足够  get stockKeyif(tonumber(redis.call('get', stockKey)) <= 0)then    -- 3.2 库存不足，返回1    return 1end-- 3.2 判断用户是否已经下单 SISMEMBER orderKey userIdif(redis.call('sismember', orderKey, userId) == 1)then    -- 3.3 重复下单，返回2    return 2end-- 3.4 扣除库存 incrby stockKey -1redis.call('incrby', stockKey, -1)-- 3.5 下单(保存用户) sadd orderKey userIdredis.call('sadd', orderKey, userId)-- 3.6 发送消息到队列中， XADD stream.orders * k1 v1 k2 v2 ...redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)return 0

/** * 抢购优惠券 * * @param voucherId * @return */@Overridepublic Result seckillVoucher(Long voucherId) {    Long userId = UserHolder.getUser().getId();    long orderId = redisIdWorker.nextId("order");    // 1. 执行Lua脚本    Long result = stringRedisTemplate.execute(        SECKILL_SCRIPT,        Collections.emptyList(),        voucherId.toString(),        userId.toString(),        String.valueOf(orderId)    );    int r = result.intValue();    // 2. 判断结果是否为0    // 2.1 不为0，代表没有购买资格    if (r != 0) {        return Result.fail(r == 1 ? "库存不足" : "不能重复下单");    }    // 3. 返回订单id    return Result.ok(orderId);}

9.9.3 获取stream.orders中的消息，完成下单

package com.hmdp.service.impl;import cn.hutool.core.bean.BeanUtil;import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;import com.hmdp.dto.Result;import com.hmdp.entity.SeckillVoucher;import com.hmdp.entity.VoucherOrder;import com.hmdp.mapper.VoucherOrderMapper;import com.hmdp.service.ISeckillVoucherService;import com.hmdp.service.IVoucherOrderService;import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;import com.hmdp.utils.RedisIdWorker;import com.hmdp.utils.UserHolder;import lombok.extern.slf4j.Slf4j;import org.redisson.api.RLock;import org.redisson.api.RedissonClient;import org.springframework.core.io.ClassPathResource;import org.springframework.data.redis.connection.stream.*;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.data.redis.core.script.DefaultRedisScript;import org.springframework.stereotype.Service;import org.springframework.transaction.annotation.Transactional;import javax.annotation.PostConstruct;import javax.annotation.Resource;import java.time.Duration;import java.util.Collections;import java.util.List;import java.util.Map;import java.util.concurrent.*;@Slf4j@Service@Transactional(rollbackFor = Exception.class)public class VoucherOrderServiceImpl extends ServiceImpl
   
     implements IVoucherOrderService {    @Resource    private ISeckillVoucherService seckillVoucherService;    @Resource    private RedisIdWorker redisIdWorker;    @Resource    private RedissonClient redissonClient;    @Resource    private StringRedisTemplate stringRedisTemplate;    /**     * 加载Lua脚本     */    public static final DefaultRedisScript
    
      SECKILL_SCRIPT;    static {        SECKILL_SCRIPT = new DefaultRedisScript<>();        // 脚本文件位置        SECKILL_SCRIPT.setLocation(new ClassPathResource("Seckill.lua"));        // 脚本返回值        SECKILL_SCRIPT.setResultType(Long.class);    }    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();    @PostConstruct    private void init() {        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());    }    private class VoucherOrderHandler implements Runnable {        @Override        public void run() {            String queueName = "stream.orders";            while (true) {                try {                    // 1. 获取消息队列中的订单信息 XREADGROUP GROUP G1 C1 COUNT 1 BLOCK 2000 STREAMS S1 >                    List
     
      
       > list = stringRedisTemplate.opsForStream().read(                            Consumer.from("g1", "c1"),                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),                            StreamOffset.create(queueName, ReadOffset.lastConsumed())                    );                    // 2. 判断订单信息是否为空                    if (list == null || list.isEmpty()) {                        // 2.1 如果为null，说明没有消息，继续下一次循环                        continue;                    }                    // 2.2 解析消息                    MapRecord
       
         record = list.get(0);                    Map
        
          value = record.getValue(); VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true); // 3. 创建订单 createVoucherOrder(voucherOrder); // 4. 确认消息 XACK s1 g1 id stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId()); } catch (Exception e) { log.error("处理订单异常", e); hanldependdingList(); } } } private void hanldependdingList() { String queueName = "stream.orders"; while (true) { try { // 1. 获取pending-list中的订单信息 XREADGROUP GROUP G1 C1 COUNT 1 BLOCK 2000 STREAMS S1 > List
         
          
           > list = stringRedisTemplate.opsForStream().read( Consumer.from("g1", "c1"), StreamReadOptions.empty().count(1), StreamOffset.create(queueName, ReadOffset.from("0")) ); // 2. 判断订单信息是否为空 if (list == null || list.isEmpty()) { // 2.1 如果为null，说明pendding-list没有异常消息，结束循环 break; } // 2.2 解析消息 MapRecord
           
             record = list.get(0); Map
            
              value = record.getValue(); VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true); // 3. 创建订单 createVoucherOrder(voucherOrder); // 4. 确认消息 XACK s1 g1 id stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId()); } catch (Exception e) { log.error("处理pendding订单异常", e); try { Thread.sleep(200); } catch (InterruptedException ex) { ex.printStackTrace(); } } } } } /** * 创建订单 * @param voucherOrder * @return */ public void createVoucherOrder(VoucherOrder voucherOrder) { // 5. 一人只能抢一份 Long userId = voucherOrder.getUserId(); Long voucherId = voucherOrder.getVoucherId(); // 使用Redisson创建锁对象 RLock redisLock = redissonClient.getLock("lock:order:" + userId); // 使用Redisson尝试获取锁 boolean isLock = redisLock.tryLock(); // 判断是否获取锁 if (!isLock) { // 获取锁失败，直接返回失败/重试 log.error("不允许重复抢购!!"); return; } try { // 5.1 查询订单 long count = this.count(new LambdaQueryWrapper
             
              () .eq(VoucherOrder::getUserId, userId) .eq(VoucherOrder::getVoucherId, voucherId)); // 5.2 判断是否已经抢过 if (count > 0) { // 用户已经购买过 log.error("用户已经购买过!!"); return; } // 6. 扣减库存 boolean success = seckillVoucherService.update(new LambdaUpdateWrapper
              
               () .eq(SeckillVoucher::getVoucherId, voucherId) .gt(SeckillVoucher::getStock, 0) .setSql("stock = stock - 1")); if (!success) { // 扣减库存失败 log.error("扣减库存失败!!"); return ; } // 存库 this.save(voucherOrder); } finally { // 使用Redisson释放锁 redisLock.unlock(); } } /** * 抢购优惠券 * * @param voucherId * @return */ @Override public Result seckillVoucher(Long voucherId) { Long userId = UserHolder.getUser().getId(); long orderId = redisIdWorker.nextId("order"); // 1. 执行Lua脚本 Long result = stringRedisTemplate.execute( SECKILL_SCRIPT, Collections.emptyList(), voucherId.toString(), userId.toString(), String.valueOf(orderId) ); int r = result.intValue(); // 2. 判断结果是否为0 // 2.1 不为0，代表没有购买资格 if (r != 0) { return Result.fail(r == 1 ? "库存不足" : "不能重复下单"); } // 3. 返回订单id return Result.ok(orderId); }}

转载地址：https://www.cnblogs.com/codertl/p/16266313.html 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！

上一篇：Redis笔记

下一篇：redis知识点总结(持续更新)

发表评论

关于作者

喝酒易醉，品茶养心，人生如梦，品茶悟道，何以解忧？唯有杜康！

-- 愿君每日到此一游！