缓存与分布式锁的应用

目录

• 缓存与分布式锁的应用
  • 1、缓存应用
    • 1、本地应用
    • 2、分布式场景中应用
    • 3、缓存相关问题
      • 3.1、缓存穿透
      • 3.2、缓存雪崩
      • 3.3、缓存击穿
  • 2、击穿解决方案-锁的应用
    • 1、本地锁（包括JUC包下）
    • 2、分布式锁
    • 3、Redisson——分布式中的JUC
    • 4、Redisson 开始

1、缓存应用

缓存使用场景：

为了提升系统性能，我们一般会将部分数据放入缓存中，加速访问，而数据库只承担数据的落盘工作

那么哪些数据适合放入缓存呢？

• 即时性，数据一致性要求不高的
• 访问量大且更新效率不高的

举例：电商类应用、商品分类、商品列表等适合放入缓存并加一个更新时间（由数据更新频率来定），后台发布一个商品，买家需要5分钟后才能看到新的商品一般是可以接受的

1、本地应用

    /**
     * 自定义缓存	
     */
    private Map cache = new HashMap<>();
    /**
     * 获取真实数据
     * 
     * @return
     */
    public NeedLockDataVO getDataVO() {
        NeedLockDataVO cacheDataVO = (NeedLockDataVO) cache.get("cacheDataVO");
        // 如果有，就返回
        if (cacheDataVO != null) {
            return cacheDataVO;
        }
        // 如果没有,获取到db中数据
        NeedLockDataVO needLockDataVO = doGetNeedLockDataVO();
        // put进缓存
        return (NeedLockDataVO) cache.put("cacheDataVO", needLockDataVO);
    }

    private NeedLockDataVO doGetNeedLockDataVO() {
        // TODO 繁琐业务逻辑代码
      	try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
        }
        return new NeedLockDataVO();
    }

我们这里的缓存组件利用了原生的Map。

在同一个项目，同一个JVM中，即本地保存一个副本的，可以称为本地缓存。

在单个服务应用中，我们使用本地缓存模式，缓存组件和应用如果永远只用同一个机器上部署，不会出现任何问题，并且效率很高。

但是我们思考下面一个场景：

如果在一个分布式的系统下，我们一个服务项目往往会部署十几台服务器上，每一个服务器都自带一个自己的一个本地缓存，这样会出现什么问题呢？

分布式场景下，单体应用自带的缓存仅对与自己所在服务器上生效。假设我们有一个商品服务，同时部署在多个服务器上，客户端发起了查询一个商品列表的请求，我们通过负载均衡找到第一台服务器，发现一号服务器的本地缓存中没有，就从数据库中查询出来，放到了一号服务器的本地缓存中。第二个客户端请求同样是查询这个商品列表，通过负载均衡，找到了第二台服务器，那么此时第二台服务器的本地缓存是不会有第一台的缓存信息的。如此会引来一下问题：

问题：

• 服务器各顾各的，每一个请求进来，都得查一遍放入自己的缓存中
• 如果对数据进行修改，一般还要修改缓存中的数据，假设我们通过负载均衡，只修改了一号服务器缓存数据，那么以后负载均衡到其他服务器上的请求所得到的数据，就会和一号服务器的数据有所不同，就会出现一个严重的问题：数据一致性问题

那么，我们分布式系统下，该如何使用缓存，解决数据一致性的问题呢？

2、分布式场景中应用

我们引入一个中间件redis，将缓存控制交给第三方来处理，所有读取｜写入缓存的数据都交由redis，本地不再做缓存

/**
     * StringRedisTemplate
     */
    @Resource
    StringRedisTemplate cache;

    /**
     * 获取真实数据
     *
     * @return
     */
    public NeedLockDataVO getDataVO() {
        //  从redis中取出
        String jsonObjectStr = cache.opsForValue().get("cacheDataVO");
        NeedLockDataVO cacheDataVO = JSONObject.parseObject(jsonObjectStr, new TypeReference(){});
        // 如果有，就返回
        if (cacheDataVO != null) {
            return cacheDataVO;
        }
        // 如果没有,获取到db中数据
        NeedLockDataVO needLockDataVO = doGetNeedLockDataVO();
        // 存入redis
        cache.opsForValue().set("cacheDataVO", JSON.toJSONString(needLockDataVO));
        return needLockDataVO;
    }

    private NeedLockDataVO doGetNeedLockDataVO() {
        // TODO 繁琐业务逻辑代码
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
        }
        return new NeedLockDataVO();
    }

接下来，我们利用jmeter压测一下会发现，redis后期会频繁报错：OutOfDirectMemoryError

产生的原因：

Redis自动配置

@Import({ LettuceConnectionConfiguration.class, JedisConnectionConfiguration.class })
public class RedisAutoConfiguration {

}

SpringBoot2.0以后默认使用Lettuce作为操作redis的客户端，它使用netty进行网络通信lettuce的bug导致堆外内存溢出，netty如果没有指定堆外内存，就会默认使用虚拟机的-Xms的值,可以使用-Dio.netty.maxDirectMemory进行设置,时间久了堆外内存溢出问题肯定还会出现,治标不治本。

那我们该如何解决这个棘手的问题呢？

Jedis 替换SpringBoot默认使用的Lettuce

首先排除Lettuce包：

				
            org.springframework.boot
            spring-boot-starter-data-redis
                
                    
                        io.lettuce
                        lettuce-core
                    
                
        
        
            redis.clients
            jedis
        

堆外内存溢出OutOfDirectMemoryError,完美解决!

我们通过加第三方缓存解决了缓存一致性的问题，可是我们设想一个场景,如果有两台服务器，A先请求更新db，B在之后更新db，但是A请求的服务器，网络出现问题导致延时，B请求的服务器在A请求的服务器之前首先操作缓存，那么我们按照常理来讲，应该是从缓存中查询到最后一次更新的数据，这就引来了另一个问题：缓存的最终一致性问题

那么，我们对这个"缓存最终一致性"问题又该如何解决呢？我们先从缓存存在的安全问题引出来解决方案！

3、缓存相关问题

3.1、缓存穿透

查询了一定不存在的数据-缓存穿透

缓存穿透：是指缓存和数据库中都不存在的数据，而用户不断发起请求，如发起查询id=“-1”的数据或id为特别大不存在的数据，这时候用户很可能是攻击者，攻击会导致数据库压力过大

解决方案：

• 接口层增加校验（用户鉴权、id作为基础校验，过滤id为-1的请求）
• 缓存中去不到的数据，也写入缓存中，以key-null形式保存（缓存时间设置短一些，设置太长会导致正常情况也无法使用），这样可以防止用同一个id暴力攻击
• **布隆过滤器：具体看大佬的文章

3.2、缓存雪崩

缓存中不同数据大批量过期-缓存穿透

缓存雪崩：是指缓存中数据大批量过期而且查询量巨大，引起数据库压力巨大，甚至宕机，与缓存击穿不同，雪崩是查询不同的数据都过期了

解决方案：

• 随机时间：在原有的过期时间的基础上，加上一个随机的时间，防止同一时间数据大批量过期
• 永不过期缓存：在情况允许的情况下，设置缓存数据永不过期
• redis高可用：预防redis宕机导致雪崩问题
• 限流降级：通过加锁和队列的方式进行对数据库的读取和写缓存，例如：对某个key只允许一个线程查询数据库，其他线程等待
• 数据预热：在正式部署之前，先把数据访问一遍加入缓存，设置不同的过期时间，让缓存失效的时间点尽量均匀

3.3、缓存击穿

热点key刚好失效-缓存击穿

缓存击穿：一个热点key在某个时间点失效的情况下，有大批量线程去查询该key，导致大批量线程去查询数据库，引起数据库压力巨大，甚至宕机

解决方案：

• 互斥锁：在并发的多个请求中，只有第一个请求线程能拿到锁并执行数据库查询操作，等到第一个线程将数据写入缓存后，其他线程直接走缓存
• 分布式锁：在分布式场景下，本地互斥锁不能保证只有一个线程去查询数据库，也可以使用分布式锁去避免击穿问题
• 热点数据不过期：直接将缓存设置为不过期，然后由定时任务去异步加载数据，更新缓存

关于缓存击穿，我们如何选定方案呢？

本质上我们是在并发场景很高的情况下，通过降低访问数据库的线程并发量，来达到避免缓存击穿的问题出现。

互斥锁VS分布式锁：

我们很多时候是通过集群部署多个相同的服务，本地互斥锁虽然不能严谨控制单个线程查询数据库，但是我们的目的是降低并发量，只要保证走到数据库的请求能大大降低即可，所以还有另一个思路是 JVM 锁，当然如果要保证缓存最终一致性的场景，我们还是需要用到分布式锁作为最终解决方案的！

JVM 锁保证了在单台服务器上只有一个请求走到数据库，通常来说已经足够保证数据库的压力大大降低，同时在性能上比分布式锁更好。
值得注意的是：无论是使用“分布式锁”，还是“JVM 锁”，加锁时要按 key 维度去加锁。

使用固定的key值加锁，这样会导致不同的 key 之间也会互相阻塞，造成性能严重损耗。

2、击穿解决方案-锁的应用

综合上面的结果，我们的redis缓存虽然提升了性能，但是在一些特殊场景下，仍会存在一些问题（缓存击穿与数据最终一致性）。

我们了解到分布式锁是可以通过单个线程访问数据库资源，解决上面两个问题的，那么我们接下来讨论一下“锁”相关的应用。

1、本地锁（包括JUC包下）

在我们引入解决方案之前，我们先看一个例子：

		/**
     * 获取真实数据
     *
     * @return
     */
    public NeedLockDataVO getDataVO() {
        String jsonObjectStr = cache.opsForValue().get("cacheDataVO");
        NeedLockDataVO cacheDataVO = JSONObject.parseObject(jsonObjectStr, new TypeReference(){});
        // 如果有，就返回
        if (cacheDataVO != null) {
            return cacheDataVO;
        }
        // 如果没有,获取到db中数据
        NeedLockDataVO needLockDataVO = doGetNeedLockDataVO();
        // 存入redis
        cache.opsForValue().set("cacheDataVO", JSON.toJSONString(needLockDataVO));
        return needLockDataVO;
    }

    private NeedLockDataVO doGetNeedLockDataVO() {
				// 数据本地加锁
        synchronized (this){
            // TODO 繁琐业务逻辑代码
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
            }
            return new NeedLockDataVO();
        }
    }

假设我们实例，交由Spring来管理，this获取的是同一个

不知道我们有没有发现以下问题：

1. 缓存的读取与存入均不在锁内，即便是单体服务器，并发情况下都会出现宕机风险问题。
2. 加锁是在本地，多个服务器下，仍然会有多个线程去访问数据库，缓存数据一致性仍然得不到解决

我们对“1”的问题，只需要在进入锁之后查一遍缓存即可。

代码片段更改如下：

private NeedLockDataVO doGetNeedLockDataVO() {
        synchronized (this){
          	// 再次查询缓存，预防宕机风险
            String jsonObjectStr = cache.opsForValue().get("cacheDataVO");
            NeedLockDataVO cacheDataVO = JSONObject.parseObject(jsonObjectStr, new TypeReference(){});
            // 如果有，就返回
            if (cacheDataVO != null) {
                return cacheDataVO;
            }
            // TODO 繁琐业务逻辑代码
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
            }
            return new NeedLockDataVO();
        }
    }

我们对“2”的问题，如何做一个分布式的锁来解决当前的隐患问题呢？

2、分布式锁

什么是？

当多个进程不在同一个系统中，用分布式锁控制多个进程对资源的访问。

分布式解决方案

针对分布式锁的实现，目前比较常用的有以下几种方案：

1. 基于数据库实现分布式锁
2. 基于缓存（redis，memcached，tair）实现分布式锁
3. 基于Zookeeper实现分布式锁

我们着重讨论一下基于缓存的分布式锁演进实现：

阶段一

  public Map> getCatalogJsonFromDbWithRedisLock() {
		//1：占分布式锁。去redis占坑
    Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "111");

    if (lock){
        //加锁成功。。。执行业务
        Map> dataFromDB = getDataFromDB();

        stringRedisTemplate.delete("lock");//删除锁
        return dataFromDB;

    }else {
        //加锁失败。。。。重试
        //休眠一百毫秒重试
        return getCatalogJsonFromDbWithRedisLock();//自旋方式
    }
}

问题：

• 如果我们现在在获取到锁以后，执行业务出现了异常，会导致锁没有释放，造成死锁

原因：加锁和解锁过程互不影响，不会整体回滚，没有对出现异常后锁做处理

解决方案：

• 为锁指定过期时间，到期自动解锁

阶段二

public Map> getCatalogJsonFromDbWithRedisLock() {
  	//1：占分布式锁。去redis占坑
    Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "111");

    if (lock){
        //加锁成功。。。执行业务
        //2：设置过期时间
        stringRedisTemplate.expire("lock",30, TimeUnit.SECONDS);
        Map> dataFromDB = getDataFromDB();

        stringRedisTemplate.delete("lock");//删除锁
        return dataFromDB;

    }else {

        //加锁失败。。。。重试
        //休眠一百毫秒重试
        return getCatalogJsonFromDbWithRedisLock();//自旋方式

    }

问题：

• 同样是如果因为异常原因，导致过期时间没有设置执行，造成死锁

原因：加锁和设置过期时间侧操作不是原子性

解决方案：

我们可以使用SET key value [EX seconds]，保证加锁和过期时间设置的原子性

阶段三

public Map> getCatalogJsonFromDbWithRedisLock() {
        //1：占分布式锁。去redis占坑
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "111",30, TimeUnit.SECONDS);
        if (lock){
            //加锁成功。。。执行业务
            //2：设置过期时间
//            stringRedisTemplate.expire("lock",30, TimeUnit.SECONDS);
            Map> dataFromDB = getDataFromDB();

            stringRedisTemplate.delete("lock");//删除锁
            return dataFromDB;
        }else {

            //加锁失败。。。。重试
            //休眠一百毫秒重试
            return getCatalogJsonFromDbWithRedisLock();//自旋方式

        }
    }

问题：

• 业务超时发现锁已经到期自动删除了，没锁可以删除了，怎么办？
• 业务用时很长，锁自动过期后，我们把别人的锁删除了，怎么办？其他的线程又进来怎么办？

原因：基于性能和网络的综合原因，我们不能保证超时时间永远小于过期时间，业务超时时间过长，会导锁混乱，甚至达不到加锁的目的。

解决方案：

我们要保证删除锁的时候，我们不可以删除别人的锁

阶段四

public Map> getCatalogJsonFromDbWithRedisLock() {
        //1：占分布式锁。去redis占坑
        String uuid= UUID.randomUUID().toString();
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "uuid",30, TimeUnit.SECONDS);
        if (lock){
            //加锁成功。。。执行业务
            //2：设置过期时间
//            stringRedisTemplate.expire("lock",30, TimeUnit.SECONDS);
            Map> dataFromDB = getDataFromDB();
//获取值对比和对比成功删除一定要是一个原子操作
            String lockValue = stringRedisTemplate.opsForValue().get("lock");
            if (uuid.equals(lockValue)){
                stringRedisTemplate.delete("lock");//删除锁
            }
            return dataFromDB;
        }else {
            //加锁失败。。。。重试
            //休眠一百毫秒重试
            return getCatalogJsonFromDbWithRedisLock();//自旋方式
        }
    }

问题：

• 我们在做 uuid.equals(lockValue) 之后，由于网络原因导致超时，还没有删除锁之前，其他线程更改了锁，导致我们虽然觉得是自己的值，删除的还是别人的锁，又会有很多线程进来抢占锁。
• 业务用时很长，锁自动过期后，我们把别人的锁删除了，怎么办？其他的线程又进来怎么办？

原因：删除锁没能保证原子性

解决方案：

保证删除锁的时候的原子性

阶段五

public Map> getCatalogJsonFromDbWithRedisLock() {

        //1：占分布式锁。去redis占坑
        String uuid= UUID.randomUUID().toString();
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "uuid",30, TimeUnit.SECONDS);

        if (lock){
            //加锁成功。。。执行业务
            Map> dataFromDB = getDataFromDB();
            String script="if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
                    "then\n" +
                    "    return redis.call(\"del\",KEYS[1])\n" +
                    "else\n" +
                    "    return 0\n" +
                    "end";
            //原子删除锁
            Integer lock1 = stringRedisTemplate.execute(new DefaultRedisScript(script, Integer.class), Arrays.asList("lock"), uuid);
            return dataFromDB;
        }else {
						//加锁失败。。。。重试
            //休眠一百毫秒重试
            return getCatalogJsonFromDbWithRedisLock();//自旋方式
        }
    }

问题：

• 仍然没有解决锁过期了的问题

原因：业务超时，还没有删除锁，锁就过期了，咋办？

解决方案：

加长锁时间

阶段6

 public Map> getCatalogJsonFromDbWithRedisLock() {
        //1：占分布式锁。去redis占坑
        String uuid= UUID.randomUUID().toString();
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", "uuid",30, TimeUnit.SECONDS);
        if (lock){
            //加锁成功。。。执行业务
            Map> dataFromDB;
            try {
              dataFromDB = getDataFromDB();
            }finally {
                String script="if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
                        "then\n" +
                        "    return redis.call(\"del\",KEYS[1])\n" +
                        "else\n" +
                        "    return 0\n" +
                        "end";

                //原子删除锁
                Integer lock1 = stringRedisTemplate.execute(new DefaultRedisScript(script, Integer.class), Arrays.asList("lock"), uuid);
            }
            return dataFromDB;
        }else {
            //加锁失败。。。。重试
            //休眠一百毫秒重试
            return getCatalogJsonFromDbWithRedisLock();//自旋方式
        }
    }

分布式锁总结：

• 设置足够长的过期时间
• 加锁和过期时间必须是原子性操作
• 删除锁也必须是原子性操作

3、Redisson——分布式中的JUC

在我们了解了分布式锁的演进过程后，能解决一般的场景问题，但是遇到一些复杂的场景，我们需要更高级的分布式锁，怎么办呢？Redis为我们提供了一站式解决方案——Redisson（Distributed locks with Redis）

Rediosson是什么：Redisson是一个在Redis的基础上实现的Java驻内存数据网格

Redisson官方地址

Redisson（中｜英文）文档链接

所有用法，我们均可翻阅Redisson文档

4、Redisson 开始

配置-参考中文文档

/**
 * @author lishanbiao
 * @Date 2021/11/22
 */
@Configuration
public class MyRedissonConfig {
    @Bean(destroyMethod="shutdown")
    RedissonClient redisson() throws IOException {
        Config config = new Config();
        config.useSingleServer()
                .setAddress("redis://127.0.0.1:6379");
        return Redisson.create(config);
    }
}

测试

		@Autowired
    private Redisson redisson;

		/**
     * hello world
     */
    @RequestMapping("/delete")
    // @RequiresPermissions("coupon:coupon:delete")
    public String helloWorld(@RequestBody Long[] ids) {
        // 获取锁
        RLock lock = redisson.getLock("my-lock");

        // 加锁 阻塞式等待……
        lock.lock();
        try {
            System.out.println("加锁成功，执行业务……" + Thread.currentThread().getId());
            Thread.sleep(3000);
        } catch (Exception e) {

        } finally {
            // 解锁
            System.out.println("释放锁……" + Thread.currentThread().getId());
            lock.unlock();
        }
        return "hello";
    }

思考：程序删除之前终端，会不会有死锁问题呢？

测试会发现，并不会（自己动手实践）。

原因：

翻看文档会发现，业务超长执行期间，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。默认情况下，看门狗的检查锁的超时时间是30秒钟，也可以通过修改Config.lockWatchdogTimeout来另行指定。

值得注意的是：如果为锁指定了时间，则会关闭看门狗功能，业务超长后，删除锁的程序就会报错

读写锁

这些我只写一些特性（具体请翻阅Redisson文档）：

只要有写锁的存在都必须等待

• 读 + 读：相当于无锁
• 读 + 写：写等待读锁，读完后执行
• 写 + 读：读等待
• 写 + 写：写等待

双写模式——写数据库的同时，去更新缓存

失效模式——写数据库的同时去删除缓存，等待下一次读取

我们根据上面两张图可以看出：

无论我们哪种模式，都会存在数据不一致的问题，但是我们可以怎么办？

• 如果用户纬度数据（用户不可能一会儿加单，一会儿删单），并发机率非常小，不用考虑这个问题，加上过期时间，只需要隔一段时间主动查询数据库即可
• 如果是目录，商品介绍等基础数据，对业务产生不了大影响，允许缓存的不一致，若想要考虑可以使用：cananl订阅的方式
• 缓存数据+过期时间：可以保证大部分的需求
• 通过加锁并发读写，适合于写少读多的特点

总结：

• 我们放入缓存的数据不应该是实时性、一致性要求超高的数据
• 不应该过度设计，增加系统的复杂性
• 遇到实时性要求高的数据，我们应该查数据库，即使慢一些

本人只做汇总，以上所有来自各个大佬们