Redis的实战应用

缓存穿透

缓存穿透是指客户端请求的数据在缓存中和数据库中都不存在，这样缓存永远不会生效，这些请求都会打到数据库。其中有几种常见的解决方案。

• 1.缓存空对象 优点：实现简单，维护方便 缺点：额外的内存消耗，可能造成短期的不一致

• 2.布隆过滤 优点：内存占用较少，没有多余key 缺点：实现复杂，存在误判可能

• 3.做好数据的基础格式校验，加强用户权限的校验，做好热点参数的限流

以下是解决缓存穿透的流程图

• 使用代码解决缓存穿透

public <R,ID> R queryWithPassThrough(
        String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit){
    String key = keyPrefix + id;
    // 1.从redis查询商铺缓存
    String json = stringRedisTemplate.opsForValue().get(key);
    // 2.判断是否存在 isNotBlank 在只有是字符串的时候才返回true
    if (StrUtil.isNotBlank(json)) {
        // 3.存在，直接返回
        return JSONUtil.toBean(json, type);
    }
    // 要判断json 是为null 还是 空字符串
    if (json != null) {
        // 证明是空字符串 直接返回
        return null;
    }

    // 4.进入到这里，证明json为null值，根据id查询数据库
    R r = dbFallback.apply(id);
    // 5.不存在，返回错误
    if (r == null) {
        // 将空值写入redis
        stringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.MINUTES);
        // 返回错误信息
        return null;
    }
    // 6.存在，写入redis
    this.set(key, r, time, unit);
    return r;
}
```

缓存雪崩

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

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

缓存击穿

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

• 1.互斥锁

• 2.逻辑过期

关于缓存击穿两种方案的对比

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

用代码实现互斥锁

public <R, ID> R queryWithMutex(
        String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit) {
    String key = keyPrefix + id;
    // 1.从redis查询商铺缓存
    String shopJson = stringRedisTemplate.opsForValue().get(key);
    // 2.判断是否存在
    if (StrUtil.isNotBlank(shopJson)) {
        // 3.存在，直接返回
        return JSONUtil.toBean(shopJson, type);
    }
    // 要判断json 是为null 还是 空字符串
    if (shopJson != null) {
        // 返回一个错误信息
        return null;
    }

    // 4.实现缓存重建
    // 4.1.获取互斥锁
    String lockKey = LOCK_SHOP_KEY + id;
    R r = null;
    try {
        boolean isLock = tryLock(lockKey);
        // 4.2.判断是否获取成功
        if (!isLock) {
            // 4.3.获取锁失败，休眠并重试
            Thread.sleep(50);
            return queryWithMutex(keyPrefix, id, type, dbFallback, time, unit);
        }
        // 4.4.获取锁成功，根据id查询数据库
        r = dbFallback.apply(id);
        // 5.不存在，返回错误
        if (r == null) {
            // 将空值写入redis
            stringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.MINUTES);
            // 返回错误信息
            return null;
        }
        // 6.存在，写入redis
        this.set(key, r, time, unit);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }finally {
        // 7.释放锁
        unlock(lockKey);
    }
    // 8.返回
    return r;
}

//这里采用的是redis里面的 setnx key value 实现锁
private boolean tryLock(String key) {
    Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);
    return BooleanUtil.isTrue(flag);
}

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

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

用代码实现逻辑过期方式

public <R, ID> R queryWithLogicalExpire(
        String keyPrefix, ID id, Class<R> type, Function<ID, R> dbFallback, Long time, TimeUnit unit) {
    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(expireTime.isAfter(LocalDateTime.now())) {
        // 5.1.未过期，直接返回店铺信息
        return r;
    }
    // 5.2.已过期，需要缓存重建
    // 6.缓存重建
    // 6.1.获取互斥锁
    String lockKey = LOCK_SHOP_KEY + id;
    boolean isLock = tryLock(lockKey);
    // 6.2.判断是否获取锁成功
    if (isLock){
        // 6.3.成功，开启独立线程，实现缓存重建
        CACHE_REBUILD_EXECUTOR.submit(() -> {
            try {
                // 查询数据库
                R newR = dbFallback.apply(id);
                // 重建缓存
                this.setWithLogicalExpire(key, newR, time, unit);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }finally {
                // 释放锁
                unlock(lockKey);
            }
        });
    }
    // 6.4.返回过期的商铺信息
    return r;
}
```

超卖问题

超卖问题是典型的多线程安全问题，针对这一问题的常见解决方案就是加锁

• 乐观锁的关键是判断之前查询得到的数据是否有被修改过，常见的方式有两种。 1.版本号法

2.CAS法

• 悲观锁是让线程串行执行，简单粗暴，但是性能一般

一人一单

要求一个人只能抢购一种商品，关键在于判断数据库中的订单表中，是否已经存在该商品和用户的订单。

但是一人一单很容易产生并发安全问题，其中解决的办法就是加锁

在单机情况下，我们加锁以后能够解决并发问题

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

//我们要锁住的是同一个实例 toString方法内部是要new出一个String对象 所以要用intern(),不然锁不住同一用户的多个线程
synchronized (userId.toString().intern()){
    int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();
    if (count > 0){
        //用户已经购买过了
        return Result.fail("用户已经购买过一次");
    }
    //4.扣减库存 限制库存为0
    boolean success = seckillVoucherService.update().setSql("stock = stock -1").eq("voucher_id", voucherId).gt("stock",0).update();
    if (!success){
        return Result.fail("库存不足");
    }
    //5.创建订单
    VoucherOrder voucherOrder = new VoucherOrder();
    long orderId = redisIdWorker.nextId("order");
    voucherOrder.setId(orderId);
    voucherOrder.setUserId(UserHolder.getUser().getId());
    voucherOrder.setVoucherId(voucherId);
    save(voucherOrder);
    return Result.ok(orderId);
}
```

但是目前很多项目都是分布式微服务，服务会部署在多个服务器上。因此这种锁在分布式的情况下，也会产生并发问题，因为部署在多个tomcat里面的jvm有各自的synchronized锁。因此我们需要一个分布式锁：满足分布式系统或集群模式下多进程可见并且互斥的锁。

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

基于Redis的分布式锁

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

• 获取锁：1.互斥：确保只能有一个线程获取锁。2.非阻塞:尝试一次，成功返回true，失败返回false。
• 释放锁：1.手动释放2.超时释放：获取时添加一个超时时间

基于Redis实现分布式锁初级版本

//加锁
@Override
public boolean tryLock(long timeoutSec) {
    // 获取线程标示
    String threadId = ID_PREFIX + Thread.currentThread().getId();
    // 获取锁 key:统一前缀KEY_PREFIX + name(传入的是针对同一个的用户id)，value:线程id
    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);
}
//调用  这里传进去的key 对于同一个用户不同线程来说，是相同的
SimpleRedisLock redisLock = new SimpleRedisLock("lock:" + userId,stringRedisTemplate);

分析初级版本可能发生的并发情况：

1.当线程1获取到锁以后，执行业务代码，但是产生了业务阻塞。

2.业务阻塞时间超过了我们设定的超时时间，那么这把锁就被自动释放了。

3.紧接着线程2又来获取到锁，并执行业务。

4.在线程2执行业务的过程中，线程1业务完成，便会执行del删除锁的操作（由于存的是同一用户的key），便将线程2的锁给删除掉。

5.线程3又来获取锁，此时线程2还在执行业务，并且线程3也能获取锁执行任务，这样便产生了并发问题。

改进分布式锁

由于我们是针对同一用户上的锁，所以我们对于锁的key设置的是同一key，所以不同线程针对同一用户的key是相同的，为了防止不同线程删除掉对方的key，所以我们要在value中设置自己线程的值，并且在进行解锁的时候，应该取出value值与当前线程的值进行比较。

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

分布式锁的原子性问题

在上述进行if(threadId.equals(id))判断与删除锁的行为，是两个动作，并不能保证其原子性，即在判断完以后，可能会发生阻塞，然后又发生超时释放锁，而后又删除其他线程的锁。所以我们要保证其判断和删除锁的原子性，即一起执行。我们便要采用lua脚本去操作redis。

lua脚本

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

提前读取脚本

//提前读取脚本
private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;
static {
    UNLOCK_SCRIPT = new DefaultRedisScript<>();
    UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
    UNLOCK_SCRIPT.setResultType(Long.class);
}

调用lua脚本

@Override
public void unlock() {
    // 调用lua脚本
    stringRedisTemplate.execute(UNLOCK_SCRIPT, Collections.singletonList(KEY_PREFIX + name), ID_PREFIX + Thread.currentThread().getId());
}

www.bilibili.com/video/BV1cr…