谷粒商城-缓存使用分布式锁SpringCache（5天）

缓存使用

1.1.1 哪些数据适合放入缓存
即时性、数据一致性要求不高的
访问量大且更新频率不高的数据（读多，写少）
例如：电商类应用，商品分类，商品列表等适合缓存
本地缓存
使用Map进行本地缓存
本地缓存在分布式下的问题
集群下的本地缓存不共享，存在于jvm中【并且负载均衡到新的机器后会重新查询】
数据一致性：如果一台机器修改了数据库+缓存，但是集群下其他机器的缓存未修改所以分布式情况下不使用本地缓存

redis的应用三级分类业务实现缓存

查询缓存中是否有数据
String catelogJSON = redisTemplate.opsForValue().get(“catelogJSON”);
将分类存为JSON数据，因为JSON数据是全平台兼容的
如何理解redis中的序列化和反序列化
在这里插入图片描述

如何使对象保存到redis

1.使用序列化方法这需要对象定义时实现Serializable接口
2.将对象数据使用转换为Json数据
Json.toJsonString()

RedisTemplate底层原理

redis对外内存溢出问题解决

springboot2.0以后默认使用lettuce作为操作redis的客户端。他使用netty进行网络通信
lettuce的bug导致netty堆外内存溢出 -Xmx300m；netty如果没有指定堆外内存，默认使用-Xmx300m，跟jvm设置的一样
Dio.netty.maxDirectMemory调大堆外内存，真正的原因在于netty没有及时释放资源
解决方案
升级lettuce客户端（推荐）
切换使用jedis客户端

缓存出现的问题
1.2.1 缓存穿透
缓存穿透：指查询一个数据库和缓存库都不存在的数据，每次查询都要查缓存库和数据库，一秒钟查一万次就要访问一万次数据库，这将导致数据库压力过大。如果我们在第一次查的时候就将查到的null加入缓存库并设置过期时间，这时一秒钟查一万次都不会再查数据库了，因为缓存库查到值了。

风险：利用不存在的数据进行攻击，数据库瞬时压力增大，最终导致崩溃缓存
解决：
采用：数据库查的空值放入缓存，并加入短暂过期时间
布隆过滤器（请求先查布隆过滤器、再查缓存库、数据库）
例如字符串"null"放进Redis。
Redis存的值是fastjson转换后的对象字符串，null转字符串后是字符串“null”，存到Redis里是能查到的。
1.2.2 缓存雪崩
缓存雪崩：缓存雪崩是指在我们设置缓存时key采用了相同的过期时间，导致缓存在某一时刻同时失效，请求全部转发到DB，DB瞬时压力过重雪崩。
解决：
原有的失效时间基础上增加一个随机值，比如1-5分钟随机，这样每一个缓存的过期时间的重复率就会降低，就很难引发集体失效的事件。
降级和熔断
采用哨兵或集群模式，从而构建高可用的Redis服务
如果已经发生缓存血崩：熔断、降级
1.2.3 缓存击穿【分布式锁】
一条数据过期了，还没来得及存null值解决缓存穿透，高并发情况下导致所有请求到达DB
解决：加分布式锁,获取到锁，先查缓存，其他人就有数据，不用去DB
缓存击穿：
对于一些设置了过期时间的key，如果这些key可能会在某些时间点被超高并发地访问，是一种非常“热点”的数据。
如果这个key在大量请求同时进来前正好失效，那么所有对这个key的数据查询都落到db，我们称为缓存击穿
解决：
采用：加互斥锁。大量并发只让一个去查，其他人等待，查到以后释放锁，其他人获取到锁，先查缓存，就会有数据，不用去db
热点数据不设置过期时间

本地锁解决缓存击穿

synchronized只锁当前进程
BUG：本地锁时序问题
由于将数据写入redis这一步骤没有加锁导致数据库被查询了两次

压测时，第一个线程刚释放锁，还没来得及将结果放入Redis缓存，第二个线程就拿到锁。
解决办法:
把存入缓存的操作放在锁中
本地锁缺点
本地锁只能锁住当前进程，高并发下，集群下有一百台机器，就会放一百个请求进锁查数据库。
分布式情况下，要用分布式锁。

分布式锁

什么是分布式情况
在分布式系统中，每个节点（计算机）都拥有自己的存储空间和计算能力，它们之间通过网络进行通信。
分布式情况下，服务部署在多台机器下时，本地锁只能锁住当前进程

分布式锁的原理

分布式锁最重要的是要保证占锁与删锁的原子性
在这里插入图片描述
占锁可以都去redis占有
使用redis实现分布式锁
NX – 只有键key不存在的时候才会设置key的值。实现分布式锁
多个进程使用set lock NX占有锁只有一个会占有成功
手动释放锁-出现死锁

设置超时自动删除
在这里插入图片描述
最终解决
1.加锁，只有键key不存在的时候才会设置key的值，加锁时将value设为UUID并构造方法设置锁的300秒自动过期。
2.如果加锁成功…执行业务后，使用lua脚本（保证原子性）判断当前锁的value是不是当前线程的UUID，是的话删除锁。
3.如果加锁失败，休眠100ms重试。
在这里插入图片描述
注意：
1.加锁保障原子性，删锁保证原子性。
2.为了避免死锁，加锁和设置锁的自动过期必须是原子操作，使用构造方法设置过期时间，过期时间要久一点，作为删锁失败的保险操作，这里设置成300秒。
3.为了防止删成上个线程的锁，将锁的value设成当前线程的UUID；
4.判断锁的值是不是当前线程的UUID，以及删除锁，这两个操作必须是原子操作，使用lua脚本判断锁和删除锁。
只能保证当前服务（非分布式）情况加锁成功

Redisson分布式锁单节点配置

redis官方推荐Redisson
注册一个Redission Client

可重入锁

可重入锁（ReentranRank）到底是什么？
允许同一个线程多次获得同一把锁。这种锁的主要特点是避免了同一个线程在尝试再次获取已持有的锁时产生死锁。
锁计数：可重入锁内部维护一个计数器，以跟踪同一线程对锁的获取次数。每当线程重新获取这个锁时，计数器就会增加；当线程释放锁时，计数器就会减少。只有当计数器回到零时，锁才真正释放，其他线程才能获取它。
实现

lock.lock();

如何保证代码出问题，所仍被正常释放
Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。默认情况下，看门狗的检查锁的超时时间是30秒钟 该做法有死锁风险？？执行删锁逻辑的时候，自己的锁已经被删了

避免看门狗三十秒之后删除锁该业务

指定锁的过期时间，看门狗不会自动续期：

//在自定义锁的存在时间时不会自动解锁
lock.lock(30, TimeUnit.SECONDS);
注意：

设置的自动解锁时间一定要稳稳地大于业务时间

分布式下 lock()方法的两大特点：

1、会有一个看门狗机制，在我们业务运行期间，将我们的锁自动续期
2、为了防止死锁，加的锁设置成30秒的过期时间，不让看门狗自动续期，如果业务宕机，没有手动调用解锁代码，30s后redis也会对他自动解锁。

公平锁

它保证了当多个Redisson客户端线程同时请求加锁时，优先分配给先发出请求的线程。所有请求线程会在一个队列中排队，当某个线程出现宕机时，Redisson会等待5秒后继续下一个线程，也就是说如果前面有5个线程都处于等待状态，那么后面的线程会等待至少25秒

RLock fairLock = redisson.getFairLock("anyLock");
// 最常见的使用方法
fairLock.lock();

读写锁（ReadWriteLock）可重入读写锁

分布式可重入读写锁，允许同时有多个读锁和一个写锁处于加锁状态。
Redisson的读写锁实现了JUC.locks.ReadWriteLock接口，读读不互斥，读写互斥，写写互斥。
写锁会阻塞读锁，读锁会阻塞写锁，但是读锁和读锁不会互相阻塞

RReadWriteLock rwlock = redisson.getReadWriteLock("anyRWLock");
// 最常见的使用方法
rwlock.readLock().lock();
// 或
rwlock.writeLock().lock();

闭锁（CountDownLatch）

redisson.getCountDownLatch();

可以理解为门栓，使用若干个门栓将当前方法阻塞，只有当全部门栓都被放开时，当前方法才能继续执行。
以下代码只有gogogo被调用5次后 lockDoor（）才能继续执行
在这里插入图片描述

信号量（Semaphore）

redisson.getSemaphore("semaphore")

@GetMapping("/park")
    @ResponseBody
    public String park() {
        RSemaphore park = redissonClient.getSemaphore("park");
        try {
            park.acquire(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "停车，占一个车位1";
    }
 
    @GetMapping("/go")
    @ResponseBody
    public String go() {
        RSemaphore park = redissonClient.getSemaphore("park");
        park.release(1);
        return "开走，放出一个车位1";
    }

信号量与闭锁的区别
他们都是标志位为0时解锁
但是信号量的标志位可以加,但是闭锁不能,闭锁是能减,直到标志位为0解锁

缓存数据一致性问题(redis 与数据库数据一致性问题)

双写模式

**双写模式:**在数据库进行写操作的同时对缓存也进行写操作,确保缓存数据与数据库数据的一致性
**脏数据问题：**在A修改数据库后，更新缓存时延迟高，在延迟期间，B已经有修改数据并更新缓存，过了一会A才更新缓存完毕。此时数据库里是B修改的内容，缓存库里是A修改的内容。
解决方式：加锁但redis中数据有过期时间最终会删除脏数据保持数据一致性
在这里插入图片描述

失效模式

**失效模式:**在数据库进行更新操作时,删除原来的缓存,再次查询数据库就可以更新最新数据
存在问题（写多读少时读数据更新缓存时不能保证是最新写入的数据）
脏数据问题：当两个请求同时修改数据库，A已经更新成功并删除缓存时又有读数据的请求进来，这时候发现缓存中无数据就去数据库中查询并放入缓存，在放入缓存前第二个更新数据库的请求B成功，这时候留在缓存中的数据依然是A更新的数据
在这里插入图片描述

解决方法
1、缓存的所有数据都有过期时间，数据过期下一次查询触发主动更新
2、读写数据的时候(并且写的不频繁)，加上分布式的读写锁。

总结：

• 我们能放入缓存的数据本就不应该是实时性、一致性要求超高的。所以缓存数据的时候加上过期时间，保证每天拿到当前最新数据即可。
• 我们不应该过度设计，增加系统的复杂性
• 遇到实时性、一致性要求高的数据，就应该查数据库，即使慢点。

SpringCache-简化缓存操作

在这里插入图片描述

使用CacheManager管理Cache

在这里插入图片描述
示例
getLevel1Categorys方法加上@Cacheable(“category”)注解

/**
     * 查询一级分类。
     * 父ID是0， 或者  层级是1
     */
    @Cacheable("category") //写入缓存
    @Override
    public List<CategoryEntity> getLevel1Categorys() {
        System.out.println("调用了 getLevel1Categorys  查询了数据库........【一级分类】");
        return baseMapper.selectList(new QueryWrapper<CategoryEntity>().eq("parent_cid", 0));
    }

测试结果
指定一个名字，放入哪个分区@Cacheable({“category”})
1）当前方法的结果需要缓存，如果缓存中有，方法不被调用
2）默认缓存数据的key： category::SimpleKey []
3）默认使用jdk序列化机制，将序列化后的数据存到redis
4）默认过期时间-1，永不过期

自定义数据的key 数据格式过期时间

在这里插入图片描述


@EnableConfigurationProperties(CacheProperties.class)
@EnableCaching
@Configuration
public class MyCacheConfig {
 
//    @Autowired
//    CacheProperties cacheProperties;
 
    /**
     * 需要将配置文件中的配置设置上
     * 1、使配置类生效
     * 1）开启配置类与属性绑定功能EnableConfigurationProperties
     *
     * @ConfigurationProperties(prefix = "spring.cache")  public class CacheProperties
     * 2）注入就可以使用了
     * @Autowired CacheProperties cacheProperties;
     * 3）直接在方法参数上加入属性参数redisCacheConfiguration(CacheProperties redisProperties)
     * 自动从IOC容器中找
     * <p>
     * 2、给config设置上
     */
    @Bean
    RedisCacheConfiguration redisCacheConfiguration(CacheProperties cacheProperties) {
        RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig();
        config = config.serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(new StringRedisSerializer()));
        //指定缓存序列化方式为json
        config = config.serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(new GenericJackson2JsonRedisSerializer()));
        // 配置文件生效：RedisCacheConfiguration
        CacheProperties.Redis redisProperties = cacheProperties.getRedis();
        //设置配置文件中的各项配置，如过期时间,如果此处以下的代码没有配置,配置文件中的配置不会生效
        if (redisProperties.getTimeToLive() != null) {
            config = config.entryTtl(redisProperties.getTimeToLive());
        }
        if (redisProperties.getKeyPrefix() != null) {
            config = config.prefixKeysWith(redisProperties.getKeyPrefix());
        }
        if (!redisProperties.isCacheNullValues()) {
            config = config.disableCachingNullValues();
        }
        if (!redisProperties.isUseKeyPrefix()) {
            config = config.disableKeyPrefix();
        }
        return config;
    }
}

@CacheEvict使用将数据从缓存中删除

@Transactional
    @CacheEvict(value = {"category"},key ="'getLevel1Categorys'")//删除指定key值的缓存
    @Override
    public void updateCascade(CategoryEntity category) {
        this.updateById(category);
        categoryBrandRelationService.updateCategory(category.getCatId(),category.getName());
    }

删除其他关联的缓存
删除带category前缀的所有缓存allEntries = true

@Transactional
    @CacheEvict(value = {"category"},allEntries = true)   //调用该方法(updateCascade)会删除缓存category下的所有cache
    @Override
    public void updateCascade(CategoryEntity category) {
        this.updateById(category);
        categoryBrandRelationService.updateCategory(category.getCatId(),category.getName());
    }

@Caching 的使用

作用:在数据修改时需要对多个缓存进行操作时使用

 @Transactional
    @Caching(evict = {
            @CacheEvict(value = {"category"},key ="'getLevel1Categorys'"),
            @CacheEvict(value = {"category"},key ="'getCatalogJson'")
    })
    @Override
    public void updateCascade(CategoryEntity category) {
        this.updateById(category);
        categoryBrandRelationService.updateCategory(category.getCatId(),category.getName());
    }

@CacheEvict：触发将数据从缓存删除的操作【删除缓存】【可实现失效模式】
@CachePut：不影响方法执行更新缓存【更新缓存】【可实现双写模式】
@Caching：组合以上多个操作【实现双写+失效模式】

SpringCache的不足

1、读模式:
缓存穿透:查询一个DB不存在的数据。解决:缓存空数据;ache-null-values=true【布隆过滤器】
缓存击穿:大量并发进来同时查询一个正好过期的数据。解决:加锁; 默认未加锁【sync = true】本地锁

 @Cacheable(value = "category",key = "#root.method.name",sync = true)

缓存雪崩:大量的key同时过期。解决:加上过期时间。: spring.cache.redis.time-to-live= 360000s
2.写模式:（缓存与数据库一致)（没有解决）
1)、读写加锁。
2)、引入canal，感知mysql的更新去更新缓存
3)、读多写多，直接去查询数据库就行
总结:
常规数据（读多写少，即时性，一致性要求不高的数据）﹔完全可以使用Spring-Cache，写模式(只要缓存的数据有过期时间就可以）