尚硅谷 Redis 查询缓存工具封装和秒杀整合Lua代码实现

超、凢脫俗 2024-03-17 14:27 56阅读 0赞

### 查询缓存工具类 ###

*  缓存穿透：数据库和缓存都不存在
 *  缓存雪崩：大量 key 在同一时刻失效
 *  缓存击穿：热点 key 突然失效

/**
     * @Author: chenyang
     * @DateTime: 2023/6/21 14:17
     * @Description:
     */
    @Component
    @RequiredArgsConstructor
    public class CacheClient {
    
        public static final long CACHE_NULL_TTL = 5L;
    
        public static final long CACHE_LOCK_TTL = 5L;
    
        public static final String CACHE_LOCK_KEY_PREFIX = "cache:lock:";
    
        private static final ExecutorService CACHE_REBUILD_EXECUTOR = Executors.newFixedThreadPool(5);
    
        private final StringRedisTemplate stringRedisTemplate;
    
        private final ObjectMapper objectMapper;
    
    
        public void set(String key, Object value, Long time, TimeUnit unit) {
            stringRedisTemplate.opsForValue().set(key, toJsonStr(value), time, unit);
        }
    
        public void setWithLogicExpire(String key, Object value, Long time, TimeUnit unit) {
            RedisData data = new RedisData();
            data.setData(value);
            data.setExpireTime(LocalDateTime.now().plusSeconds(unit.toSeconds(time)));
            stringRedisTemplate.opsForValue().set(key, toJsonStr(data));
        }
    
        /**
         * 解决缓存穿透
         */
        public <R, ID> R queryWithCacheThrough(String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit) {
            String key = keyPrefix + id;
            String json = stringRedisTemplate.opsForValue().get(key);
            if (StringUtils.isNotBlank(json)) {
                return toBean(json, type);
            }
            // 判断命中的是否是空值
            if (Objects.nonNull(json)) {
                return null;
            }
            R r = dbFallback.apply(id);
            if (Objects.isNull(r)) {
                stringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, unit);
                return null;
            }
            this.set(key, r, time, unit);
            return r;
        }
    
        /**
         * 互斥锁解决缓存击穿
         */
        public <ID, R> R queryWithMutex(String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit) {
            String key = keyPrefix + id;
            String json = stringRedisTemplate.opsForValue().get(key);
            if (StringUtils.isNotBlank(json)) {
                return this.toBean(json, type);
            }
            // 空格处理
            if (Objects.nonNull(json)) {
                return null;
            }
    
            String lockKey = CACHE_LOCK_KEY_PREFIX + id;
    
            R r;
            try {
                boolean successLock = tryLock(lockKey);
                if (!successLock) {
                    Thread.sleep(50);
                    this.queryWithMutex(keyPrefix, id, type, dbFallback, time, unit);
                }
                r = dbFallback.apply(id);
                if (Objects.isNull(r)) {
                    stringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.SECONDS);
                    return null;
                }
                stringRedisTemplate.opsForValue().set(key, toJsonStr(r), time, unit);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            } finally {
                unlock(lockKey);
            }
            return r;
        }
    
    
        /**
         * 逻辑过期解决缓存击穿
         */
        public <ID, R> R queryWithLogicExpire(String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit) {
            String key = keyPrefix + id;
            String json = stringRedisTemplate.opsForValue().get(key);
            if (StringUtils.isBlank(json)) {
                return null;
            }
    
            RedisData redisData = toBean(json, RedisData.class);
            R r = toBean(redisData.getData(), type);
            LocalDateTime expireTime = redisData.getExpireTime();
            if (LocalDateTime.now().isBefore(expireTime)) {
                return r;
            }
    
            String lockKey = CACHE_LOCK_KEY_PREFIX + id;
    
            boolean successLock = tryLock(lockKey);
            if (!successLock) {
                CACHE_REBUILD_EXECUTOR.submit(() -> {
                    try {
                        R newR = dbFallback.apply(id);
                        this.setWithLogicExpire(key, newR, time, unit);
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    } finally {
                        unlock(lockKey);
                    }
                });
            }
            return r;
        }
    
    
        /**
         * 加锁
         */
        private boolean tryLock(String key) {
            Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", CACHE_LOCK_TTL, TimeUnit.SECONDS);
            return Boolean.TRUE.equals(result);
        }
    
        /**
         * 解锁
         */
        private void unlock(String key) {
            stringRedisTemplate.delete(key);
        }
    
        private String toJsonStr(Object obj) {
            String res;
            try {
                res = objectMapper.writeValueAsString(obj);
            } catch (JsonProcessingException e) {
                throw new RuntimeException(e);
            }
            return res;
        }
    
    
        private <E> E toBean(String json, Class<E> type) {
            E e;
            try {
                e = objectMapper.readValue(json, type);
            } catch (JsonProcessingException ex) {
                throw new RuntimeException(ex);
            }
            return e;
        }
    
        private <E> E toBean(Object obj, Class<E> type) {
            E e;
            try {
                e = objectMapper.readValue((JsonParser) obj, type);
            } catch (IOException ex) {
                throw new RuntimeException(ex);
            }
            return e;
        }
    }

User u1 = cacheClient.queryWithCacheThrough(CACHE_USER_KEY, id, User.class, userService::getById, 10L, TimeUnit.SECONDS);
    User u2 = cacheClient.queryWithLogicExpire(CACHE_USER_KEY, id, User.class, userService::getById, 10L, TimeUnit.SECONDS);
    User u3 = cacheClient.queryWithMutex(CACHE_USER_KEY, id, User.class, userService::getById, 10L, TimeUnit.SECONDS);

### 秒杀 ###

#### 乐观锁解决超卖问题 ####

@Override
         @Transactional(rollbackFor = Exception.class)
         public Boolean secKillVoucher(Long id) {
             
             TbSeckillVoucher secKillVoucher = secKillVoucherService.getById(id);
     
             checkVoucher(secKillVoucher);
     
             boolean success = secKillVoucherService.update(Wrappers.<TbSeckillVoucher>lambdaUpdate()
                     .setSql("stock = stock - 1")
                     .eq(TbSeckillVoucher::getVoucherId, secKillVoucher.getVoucherId())
                     .gt(TbSeckillVoucher::getStock, 0)); // 乐观锁
     
             if (!success){
                 throw new RuntimeException("库存不足");
             }
     
             TbVoucherOrder voucherOrder = new TbVoucherOrder();
             long orderId = redisIdWorker.nextId("order");
             voucherOrder.setId(orderId);
             Long userId = 1L;
             voucherOrder.setUserId(userId);
             voucherOrder.setVoucherId(secKillVoucher.getVoucherId());
             return save(voucherOrder);
         }

#### 一人一单 ####

@Override
         @Transactional(rollbackFor = Exception.class)
         public Boolean secKillVoucher(Long id) {
     
             Long userId = 1L;
             TbSeckillVoucher secKillVoucher = secKillVoucherService.getById(id);
     
             checkVoucher(secKillVoucher);
     
             // 由于是插入数据，所以需要加悲观锁
             int count = count(Wrappers.<TbVoucherOrder>lambdaQuery()
                     .eq(TbVoucherOrder::getVoucherId, id)
                     .eq(TbVoucherOrder::getUserId, userId));
             
             if (count > 0){
                 throw new RuntimeException("用户已经购买过一次");
             }
     
             boolean success = secKillVoucherService.update(Wrappers.<TbSeckillVoucher>lambdaUpdate()
                     .setSql("stock = stock - 1")
                     .eq(TbSeckillVoucher::getVoucherId, secKillVoucher.getVoucherId())
                     .gt(TbSeckillVoucher::getStock, 0)); // 乐观锁
     
             if (!success){
                 throw new RuntimeException("库存不足");
             }
     
             TbVoucherOrder voucherOrder = new TbVoucherOrder();
             long orderId = redisIdWorker.nextId("order");
             voucherOrder.setId(orderId);
     
             voucherOrder.setUserId(userId);
             voucherOrder.setVoucherId(secKillVoucher.getVoucherId());
             return save(voucherOrder);
         }

**加锁**

@Override
        public Boolean secKillVoucher(Long id) {
    
            Long userId = 1L;
            TbSeckillVoucher secKillVoucher = secKillVoucherService.getById(id);
    
            checkVoucher(secKillVoucher);
            return createVoucherOrder(id, userId);
        }
    
    
        // 这样加锁的粒度太粗了
        @Transactional(rollbackFor = Exception.class)
        public synchronized Boolean createVoucherOrder(Long voucherId, Long userId){
            int count = count(Wrappers.<TbVoucherOrder>lambdaQuery()
                    .eq(TbVoucherOrder::getVoucherId, voucherId)
                    .eq(TbVoucherOrder::getUserId, userId));
    
            if (count > 0){
                throw new RuntimeException("用户已经购买过一次");
            }
    
            boolean success = secKillVoucherService.update(Wrappers.<TbSeckillVoucher>lambdaUpdate()
                    .setSql("stock = stock - 1")
                    .eq(TbSeckillVoucher::getVoucherId, voucherId)
                    .gt(TbSeckillVoucher::getStock, 0)); // 乐观锁
    
            if (!success){
                throw new RuntimeException("库存不足");
            }
    
            TbVoucherOrder voucherOrder = new TbVoucherOrder();
            long orderId = redisIdWorker.nextId("order");
            voucherOrder.setId(orderId);
    
            voucherOrder.setUserId(userId);
            voucherOrder.setVoucherId(voucherId);
            return save(voucherOrder);
        }

**优化锁的粒度**

@Transactional(rollbackFor = Exception.class)
        public Boolean createVoucherOrder(Long voucherId, Long userId){
          synchronized (userId.toString().intern()) {
              int count = count(Wrappers.<TbVoucherOrder>lambdaQuery()
                      .eq(TbVoucherOrder::getVoucherId, voucherId)
                      .eq(TbVoucherOrder::getUserId, userId));
    
              if (count > 0){
                  throw new RuntimeException("用户已经购买过一次");
              }
    
              boolean success = secKillVoucherService.update(Wrappers.<TbSeckillVoucher>lambdaUpdate()
                      .setSql("stock = stock - 1")
                      .eq(TbSeckillVoucher::getVoucherId, voucherId)
                      .gt(TbSeckillVoucher::getStock, 0)); // 乐观锁
    
              if (!success){
                  throw new RuntimeException("库存不足");
              }
    
              TbVoucherOrder voucherOrder = new TbVoucherOrder();
              long orderId = redisIdWorker.nextId("order");
              voucherOrder.setId(orderId);
    
              voucherOrder.setUserId(userId);
              voucherOrder.setVoucherId(voucherId);
              return save(voucherOrder);
              // 先释放锁，再提交事务，事务是被 Spring 管理的，事务的提交是在函数执行完之后，由 Spring 做的提交，锁在大括号执行结束之后已经释放了，锁释放了意味着其他线程已经可以进来了。
              //而此时因为事务尚未提交，有可能出现并发安全问题,应该是要在事务提交之后再去释放锁 
          }
        }

@Override
        public Boolean secKillVoucher(Long id) {
    
            Long userId = 1L;
            TbSeckillVoucher secKillVoucher = secKillVoucherService.getById(id);
    
            checkVoucher(secKillVoucher);
            // 保证事务的特性，同时也控制了锁的粒度
            // 问题：事务会失效
            synchronized (userId.toString().intern()){
                TbVoucherOrderService proxy = (TbVoucherOrderService) AopContext.currentProxy();
                return proxy.createVoucherOrder(id, userId);
            }
        }
    
    
        @Transactional(rollbackFor = Exception.class)
        @Override
        public Boolean createVoucherOrder(Long voucherId, Long userId){
            int count = count(Wrappers.<TbVoucherOrder>lambdaQuery()
                    .eq(TbVoucherOrder::getVoucherId, voucherId)
                    .eq(TbVoucherOrder::getUserId, userId));
    
            if (count > 0){
                throw new RuntimeException("用户已经购买过一次");
            }
    
            boolean success = secKillVoucherService.update(Wrappers.<TbSeckillVoucher>lambdaUpdate()
                    .setSql("stock = stock - 1")
                    .eq(TbSeckillVoucher::getVoucherId, voucherId)
                    .gt(TbSeckillVoucher::getStock, 0)); // 乐观锁
    
            if (!success){
                throw new RuntimeException("库存不足");
            }
    
            TbVoucherOrder voucherOrder = new TbVoucherOrder();
            long orderId = redisIdWorker.nextId("order");
            voucherOrder.setId(orderId);
    
            voucherOrder.setUserId(userId);
            voucherOrder.setVoucherId(voucherId);
            return save(voucherOrder);
    
        }
    
    
        private void checkVoucher(TbSeckillVoucher voucher){
            if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
                throw new RuntimeException("秒杀尚未开始");
            }
            // 3.判断秒杀是否已经结束
            if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
                throw new RuntimeException("秒杀已经结束");
            }
            // 4.判断库存是否充足
            if (voucher.getStock() < 1) {
                throw new RuntimeException("库存不足");
            }
        }

#### 分布式锁 ####

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

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

释放锁：

*  手动释放

*  超时释放：TTL

@Override
        public Boolean secKillVoucher(Long id) {
    
            Long userId = 1L;
            TbSeckillVoucher secKillVoucher = secKillVoucherService.getById(id);
    
            checkVoucher(secKillVoucher);
         
            String lockKey = "order:" + userId;
            boolean isSuccess = tryLock(lockKey, 5L);
            if (!isSuccess){
                System.out.println("一人只能领取一次！");
                return false;
            }
            try {
                TbVoucherOrderService proxy = (TbVoucherOrderService) AopContext.currentProxy();
                return proxy.createVoucherOrder(id, userId);
            } catch (IllegalStateException e) {
                throw new RuntimeException(e);
            } finally {
                unlock(lockKey);
            }
        }

public static final String KEY_PREFIX = "sec:kill:lock:";   
    
    private boolean tryLock(String name, Long timeoutSec){
        long threadId = Thread.currentThread().getId();
        Boolean isSuccess = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, String.valueOf(threadId), timeoutSec, TimeUnit.SECONDS);
        return Boolean.TRUE.equals(isSuccess);
    }
    
    private void unlock(String name){
        stringRedisTemplate.delete(KEY_PREFIX + name);
    }

**问题：会出现误删情况**

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

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

**修改锁**

private boolean tryLock(String name, Long timeoutSec){
            // 因为 thread id 是自增的，在分布式中可能出出现id相同的情况
            String threadId = ID_PREFIX + Thread.currentThread().getId();
            Boolean isSuccess = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);
            return Boolean.TRUE.equals(isSuccess);
        }
    
        private void unlock(String name){
            String currentThreadId =ID_PREFIX + Thread.currentThread().getId();
            String threadId = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
            if (currentThreadId.equals(threadId)){
                stringRedisTemplate.delete(KEY_PREFIX + name);
            }
        }

问题：判断和释放锁不是一个原子操作，在准备删除锁之时线程执行时间到期，这时候（锁过期，TTL = -1），另外一个线程获取锁成功，线程一继续执行删除了锁

#### 使用 Lua 脚本保证原子性 ####

@SpringBootTest
    class CouponApplicationTests {
    
        public static final String KEY_PREFIX = "order:sec:";
        public static final String THREAD_PREFIX = "abc-";
        private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;
        private static final DefaultRedisScript<Long> TRY_LOCK_SCRIPT;
        
        static {
            TRY_LOCK_SCRIPT = new DefaultRedisScript<>();
            TRY_LOCK_SCRIPT.setLocation(new ClassPathResource("tryLock.lua"));
            TRY_LOCK_SCRIPT.setResultType(Long.class);
    
            UNLOCK_SCRIPT = new DefaultRedisScript<>();
            UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
            UNLOCK_SCRIPT.setResultType(Long.class);
        }
        
        @Autowired
        private StringRedisTemplate stringRedisTemplate;
    
        @Test
        void tryLockTest() {
            long threadId = Thread.currentThread().getId();
            long userId = 9962L;
            long timeout = 360L;
            String key = KEY_PREFIX + userId;
            // 线程id前面拼接一个 uuid，避免分布式系统中 id 重复,由于是在测试，就写死了
            String value = THREAD_PREFIX + threadId;
            // 加锁成功返回 null
            // 加锁失败返回锁的 ttl, 便于重试
            Long ttl = stringRedisTemplate.execute(TRY_LOCK_SCRIPT, Collections.singletonList(key), value,  String.valueOf(timeout));
    
            if (Objects.isNull(ttl)){
                System.out.println("获取锁成功");
            }else {
                System.out.println("获取锁失败：ttl=" + ttl);
            }
        }
    
        @Test
        void unlockTest() {
            long threadId = Thread.currentThread().getId();
            long userId = 9962L;
            long timeout = 360L;
            String key = KEY_PREFIX + userId;
            String value = THREAD_PREFIX + threadId;
            String channelName = "unlockChannel";
            Long result = stringRedisTemplate.execute(UNLOCK_SCRIPT, Arrays.asList(key, channelName), value, String.valueOf(timeout));
            if (Objects.isNull(result)){
                System.out.println("锁不存在或已过期");
                return;
            }
            String msg = result.equals(0L) ? "锁已释放" : "可重入次数：" + result;
            System.out.println(msg);
        }
    
    }

-- tryLock.lua
    if redis.call('exists', KEYS[1]) == 0 then
        -- 不存在，设置锁
        redis.call('hset', KEYS[1], ARGV[1], 1)
        redis.call('expire', KEYS[1], ARGV[2])
        return nil
    end
    
    -- 重入锁
    if redis.call('hexists', KEYS[1], ARGV[1]) == 1 then
        redis.call('HINCRBY', KEYS[1], ARGV[1], 1)
        redis.call('expire', KEYS[1], ARGV[2])
        return nil
    end
    
    return redis.call('TTL', KEYS[1])

-- unlock.lua
    if (redis.call("hexists", KEYS[1], ARGV[1]) == 0) then
        return nil;
    end
    
    local count = redis.call('hincrby', KEYS[1], ARGV[1], -1);
    
    if count > 0 then
        redis.call('expire', KEYS[1], ARGV[2]);
        return count;
    else
        redis.call('del', KEYS[1]);
        redis.call('publish', KEYS[2], ARGV[1]);
        return 0;
    end
    
    return nil;

后续扩展：加锁失败时返回锁的 ttl，根据ttl 来判断是否要重试，具体思路可以查看 **Redisson**

问题：为什么不适用redis的事务来保证一致性？