0
点赞
收藏
分享

微信扫一扫

谷粒商城_09_分布式缓存+分布式锁

weipeng2k 2022-02-22 阅读 80
分布式锁缓存分布式spring boot后端

文章目录

  • 本地缓存：和微服务同一个进程。缺点：分布式时
  • 分布式缓存：缓存中间件

在这里插入图片描述

本地缓存

本地缓存问题：每个微服务都要有缓存服务、数据更新时只更新自己的缓存，造成缓存数据不一致

解决方案：分布式缓存，微服务共用 缓存中间件

  • 采用map来存缓存，直接放到内存中，但是本地缓存就会在分布式服务中出现问题，各个服务的缓存都是不一样的

分布式缓存

在这里插入图片描述

在这里插入图片描述

redis

整合redis的步骤

  • 引入依赖
  • 配置redis信息
  • 直接使用spring自动配置的RedisTemplate

1、product导入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

2、配置redis主机地址

spring:
  redis:
    host: localhost
    port: 6379
    # 可以指定账号密码

3、自动注入了RedisTemplate，优化菜单获取业务getCatalogJson

  • 测试StringRedisTemplate
@Autowired // 注入StringRedisTemplate
StringRedisTemplate stringRedisTemplate;

@Test
public void test(){
    // 相当于拿到redis的map
    ValueOperations<String, String> ops = stringRedisTemplate.opsForValue();
    // 保存 set
    // 获取 get
}
  • 业务getCatalogJson加入缓存逻辑
ValueOperations<String, String> ops = stringRedisTemplate.opsForValue();
// 每次从缓存中获取
String catalogJson = ops.get("catalogJson");
// 判断redis中是否有此缓存
if (catalogJson == null) {
    // 没有，则从数据库中查询
    Map<String, List<Catalog2Vo>> categoriesDb = getCategoriesDb();
    // 先将数据库的数据转为json在放入缓存中，这也就是【序列化】
    // JSON跨语言，跨平台兼容
    String toJSONString = JSON.toJSONString(categoriesDb);
    // 查到数据放到redis中
    ops.set("catalogJson",toJSONString);
    return categoriesDb;
}
// 如果缓存有，先【反序列化】，再返回
Map<String, List<Catalog2Vo>> listMap = JSON.parseObject(catalogJson, new TypeReference<Map<String, List<Catalog2Vo>>>() {});
return listMap;

堆外内存溢出

  • 当进行压力测试时后期后出现堆外内存溢出OutOfDirectMemoryError

产生原因：

1、springboot2.0以后默认使用lettuce作为操作redis的客户端，它使用netty进行网络通信

2、lettuce的bug导致netty堆外内存溢出。netty如果没有指定堆外内存，默认使用Xms的值，可以使用-Dio.netty.maxDirectMemory来设置

解决方案：由于是lettuce的bug造成，不要直接使用-Dio.netty.maxDirectMemory去调大虚拟机堆外内存。

  1. 升级lettuce客户端

  2. 切换使用jedis、导入依赖即可

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
    <exclusions>
        <exclusion>
            <groupId>io.lettuce</groupId>
            <artifactId>lettuce-core</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
</dependency>

lettuce和jedis是操作redis的底层客户端，RedisTemplate是再次封装

压力测试

在这里插入图片描述

缓存失效

缓存穿透

在这里插入图片描述

缓存穿透是指缓存和数据库中都没有的数据，而用户不断发起请求，如发起为id为10000的数据或id为特别大不存在的数据。这时的用户很可能是攻击者，攻击会导致数据库压力过大。

解决方案：

  • 缓存我们的null对象，但是要设置短暂的过期时间，防止以后有这个数据，而缓存中还是缓存的null

缓存雪崩

在这里插入图片描述

缓存雪崩是指在我们设置缓存时key采用了相同的过期时间，导致缓存在某一时刻缓存大面积的同时失效，请求全部转发到DB，DB瞬时压力过重雪崩。

解决方案：

  • 避免雪崩：缓存数据的过期时间设置随机，防止同一时间大量数据过期现象发生。
  • 如果缓存数据库是分布式部署，将热点数据均匀分布在不同缓存数据库中。
  • 设置热点数据永远不过期。
  • 出现雪崩：降级 熔断
  • 事前：尽量保证整个 redis 集群的高可用性，发现机器宕机尽快补上。选择合适的内存淘汰策略。
  • 事中：本地ehcache缓存 + hystrix限流&降级，避免MySQL崩掉
  • 事后：利用 redis 持久化机制保存的数据尽快恢复缓存

缓存击穿

在这里插入图片描述

三者不同的是：

  • 缓存击穿指并发查同一条数据。缓存击穿是指缓存中没有但数据库中有的数据（一般是缓存时间到期），这时由于并发用户特别多，同时读缓存没读到数据，又同时去数据库去取数据，引起数据库压力瞬间增大，造成过大压力
  • 缓存雪崩是不同数据都过期了，很多数据都查不到从而查数据库。
  • 缓存穿透也是指并发查同一条数据，但是穿透是数据库也没有此数据

解决方案：

  • 设置热点数据永远不过期。
  • 加互斥锁

缓存击穿：加锁

本地锁

在这里插入图片描述

1、本地锁synchronized(this)、JUC(Lock)

  • 竞争到锁后，确认缓存中没有，再去查数据库，将数据放入缓存中，再去释放锁。

  • Springboot所有的组件都是单例的，那么锁this就可以把所有的请求都锁住，但是这是单服务上的，如果是多服务就有多个实例，那么synchronized就不管用了

复制微服务

1、右键点击服务，copy configuration

在这里插入图片描述

2、修改配置

  • 在program arguments: --server.port=10002

分布式锁

在这里插入图片描述

分布式项目时，但本地锁只能锁住当前服务，需要分布式锁

redis分布式锁的原理：

  • setnx命令，同一时刻只能设置成功一个，没获取到锁阻塞或者sleep一会，原子性操作。前提，锁的key是一定的，value可以变
# redis命令 加速
set lock 1 NX

// 成功返回true，失败了就自旋重新尝试加锁
Boolean lock = RedisTemplate.opsForValue().setIfAbsent("lock","1");
// 删除锁
RedisTemplate.delete("lock");

分布式锁问题及解决：

  • 设置好了锁，万一服务出现宕机，没有执行删除锁逻辑，这就造成了死锁

    • 解决：设置过期时间：并且保证设置过期时间和加锁是原子操作

      # redis 300毫秒过期时间
set lock 1 EX 300 NX

      // 1秒
Boolean lock = RedisTemplate.opsForValue().setIfAbsent("lock", "1",500, TimeUnit.SECONDS);

  • 业务还没执行完锁就过期了，别人拿到锁，自己执行完去删了别人的锁

    • 解决：锁续期（redisson有看门狗），删锁的时候明确是自己的锁。如设置锁的value等于UUID

  • 判断uuid对了，但是将要删除的时候锁过期了，别人拿到锁设置了新值，我们拿到的是自己的uuid，这时去删除就会删除别人的锁

    • 解决：删除锁必须保证原子性（保证判断UUID和删锁是原子操作）。官方http://redis.cn/commands/set.html：redis+Lua脚本完成，脚本是原子的

    • // 官网脚本
if redis.call("get",KEYS[1]) == ARGV[1]  // KEYS获取的值，也就是key的集合。ARGV我们指定的值，也就是uuid
then
	return redis.call("del",KEYS[1]) // 如果获取的值等于我们指定的值，就删除，删除成功返回1
else
    return 0 // 失败返回0
end;

    • // 定义脚本
String script = "if redis.call(\"get\",KEYS[1]) == ARGV[1] then\n" +
    "    return redis.call(\"del\",KEYS[1])\n" +
    "else\n" +
    "    return 0\n" +
    "end";
// 执行脚本，成功返回1，失败返回0
RedisTemplate.execute(
    new DefaultRedisScript<Long>(script, Long.class), // 脚本参数，Long返回值类型
    Arrays.asList("lock"), // 键key的集合
    uuid); // 我们定义的uuid

最终代码：

public Map<String, List<Catalog2Vo>> getCatalogJsonDbWithRedisLock() {
    // 设置UUID
    String uuid = UUID.randomUUID().toString();
    ValueOperations<String, String> ops = stringRedisTemplate.opsForValue();
    // 设置锁并加上过期时间
    Boolean lock = ops.setIfAbsent("lock", uuid,500, TimeUnit.SECONDS);
    // 判断锁
    if (lock) {
        Map<String, List<Catalog2Vo>> categoriesDb = getCategoryMap();
        String lockValue = ops.get("lock");
        // get和delete原子操作
        String script = "if redis.call(\"get\",KEYS[1]) == ARGV[1] then\n" +
            "    return redis.call(\"del\",KEYS[1])\n" +
            "else\n" +
            "    return 0\n" +
            "end";
        stringRedisTemplate.execute(
            new DefaultRedisScript<Long>(script, Long.class), // 脚本和返回类型
            Arrays.asList("lock"), // 参数
            lockValue); // 参数值，锁的值
        return categoriesDb;
    }else {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 睡眠0.1s后，重新调用，自旋
        return getCatalogJsonDbWithRedisLock();
    }
}

上面的lua脚本写法每次用分布式锁时比较麻烦，我们可以采用redisson现有框架

Redis分布式锁：https://redis.io/topics/distlock

Redisson

https://github.com/redisson/redisson

官方文档

环境搭建

1、导入依赖：这个用作学习，后面可以使用redisson-spring-boot-starter

<!-- https://mvnrepository.com/artifact/org.redisson/redisson -->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.16.7</version>
</dependency>

2、开启配置：https://github.com/redisson/redisson/wiki/

@Configuration
public class MyRedisConfig {

    @Value("${ipAddr}") // 从我们的配置中获取redis地址，安全
    private String ipAddr;

    // 所有对redission的使用都是通过redissonClient对象
    // 如果是多个redis集群，也可以通过Config来配置
    @Bean(destroyMethod = "shutdown")
    public RedissonClient redisson() {
        Config config = new Config();
        // 创建单例模式的配置，也就是单个redis
        config.useSingleServer().setAddress("redis://" + ipAddr + ":6379");
        // 根据Config创建出RedissonClient实例
        return Redisson.create(config);
    }
}

// 使用直接注入RedissonClient即可

可重入锁（Reentrant Lock）

分布式锁：github.com/redisson/redisson/wiki/8.-分布式锁和同步器

方法A中调用方法B。AB都需要同一锁，此时可重入锁就可以重入，A就可以调用B。不可重入锁时，A调用B将死锁

// 参数为锁名字，获取一把锁，只要锁的名字一样就是同一把锁
RLock lock = redissonClient.getLock("CatalogJson-Lock");// 该锁实现了JUC.locks.lock接口
// 加锁
lock.lock();// 阻塞等待，默认30s，如果其他一个线程想去抢占锁，但是由于上一个线程业务超长，redisson有看门狗会自动续期，会继续等待
// 业务代码。。。。。  
// 解锁放到finally 
lock.unlock();

// 如果此线程宕机，没有释放锁，但是有看门狗，会为我们自动解锁

基于Redis的Redisson分布式可重入锁RLock Java对象实现了java.util.concurrent.locks.Lock接口。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。

锁的续期：大家都知道，如果负责储存这个分布式锁的Redisson节点宕机以后，而且这个锁正好处于锁住的状态时，这个锁会出现锁死的状态。为了避免这种情况的发生，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。默认情况下，看门狗的检查锁的超时时间是30秒钟（每到20s就会自动续借成30s，是1/3的关系），也可以通过修改Config.lockWatchdogTimeout来另行指定。

// 加锁以后10秒钟自动解锁，看门狗不续命，所以自动解锁的时间一定要大于业务时间
// 无需调用unlock方法手动解锁
lock.lock(10, TimeUnit.SECONDS);

// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
if (res) {
   try {
     ...
   } finally {
       lock.unlock();
   }
}

Redisson同时还为分布式锁提供了异步执行的相关方法：

RLock lock = redisson.getLock("anyLock");
lock.lockAsync();
lock.lockAsync(10, TimeUnit.SECONDS);
Future<Boolean> res = lock.tryLockAsync(100, 10, TimeUnit.SECONDS);

读写锁（ReadWriteLock）

分布式可重入读写锁允许同时有多个读锁和一个写锁处于加锁状态。保证一定能读到最新数据

RReadWriteLock rwlock = redisson.getReadWriteLock("anyRWLock");
// 最常见的使用方法
rwlock.readLock().lock();
// 或
rwlock.writeLock().lock();
// 10秒钟以后自动解锁
// 无需调用unlock方法手动解锁
rwlock.readLock().lock(10, TimeUnit.SECONDS);
// 或
rwlock.writeLock().lock(10, TimeUnit.SECONDS);

// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
boolean res = rwlock.readLock().tryLock(100, 10, TimeUnit.SECONDS);
// 或
boolean res = rwlock.writeLock().tryLock(100, 10, TimeUnit.SECONDS);
...
lock.unlock();
  • 读+读：并发读，相当于无锁
  • 读+写：等待读锁释放才能加写锁
  • 写+读：等待写锁释放才能加读锁
  • 写+写：阻塞等待释放
  • 只要有写的存在都要等待

上锁时在redis的状态

HashWrite-Lock
key:mode  value:read
key:sasdsdffsdfsdf... value:1

信号量（Semaphore）

信号量为存储在redis中的一个数字，当这个数字大于0时，即可以调用acquire()方法增加数量，也可以调用release()方法减少数量，但是当调用release()之后小于0的话方法就会阻塞，直到数字大于0

基于Redis的Redisson的分布式信号量（Semaphore）Java对象RSemaphore采用了与java.util.concurrent.Semaphore相似的接口和用法。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。

  • 可以用来限流操作
RSemaphore semaphore = redisson.getSemaphore("semaphore");
semaphore.acquire();
// 或
semaphore.acquireAsync();
semaphore.acquire(23);
semaphore.tryAcquire(); // 不成功直接返回false
// 或
semaphore.tryAcquireAsync();
semaphore.tryAcquire(23, TimeUnit.SECONDS);
// 或
semaphore.tryAcquireAsync(23, TimeUnit.SECONDS);
semaphore.release(10);
semaphore.release();
// 或
semaphore.releaseAsync();


// semaphore会初始化信号量，acquire一下就会减信号量，release就会加信号量，当信号量为0去acquire会阻塞
@GetMapping("/park")
@ResponseBody
public String park() {
    RSemaphore park = redissonClient.getSemaphore("park");
    try {
        park.acquire(2); // 占用两个信号
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "停进2";
}

@GetMapping("/go")
@ResponseBody
public String go() {
    RSemaphore park = redissonClient.getSemaphore("park");
    park.release(2); // 释放两个信号
    return "开走2";
}

闭锁（CountDownLatch）

基于Redisson的Redisson分布式闭锁（CountDownLatch）Java对象RCountDownLatch采用了与java.util.concurrent.CountDownLatch相似的接口和用法。

以下代码只有offLatch()被调用5次后 setLatch()才能继续执行

// CountDownLatch设置解锁的数量，await会阻塞知道解锁数量为0，countDown会减少一次解锁数量
RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch");
latch.trySetCount(1); // 等待一次解锁即可解除阻塞
latch.await(); // 阻塞

// 在其他线程或其他JVM里
RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch");
latch.countDown(); // 一次解锁

缓存数据一致性

双写模式：写完数据库后，同时写缓存

  • 问题：并发时，1先进入，2双写完成，1再写缓存，此时就会有1缓存的脏数据。有暂时的脏数据

  • 在这里插入图片描述

  • 解决：

    1、数据更新加锁

    2、如果对数据的更新不要求实时性，则可以不用管，让缓存自动过期

失效模式：写完数据库后，同时删缓存。

  • 问题：当3号请求去读的时候，2号写请求还没有写完，此时读的是1号写请求修改的数据，当2号写请求完成之后，3号读请求去更新缓存，此时是1号写请求的数据，造成脏数据。

  • 在这里插入图片描述

  • 解决：

    1、缓存设置过期时间，定期更新

    2、写数据写时，加分布式的读写锁。如果经常读，经常写会影响较大

    3、也看实时性

解决方案

  • 如果是用户纬度数据（订单数据、用户数据），这种并发几率非常小，不用考虑这个问题，缓存数据加上过期时间，每隔一段时间触发读的主动更新即可
  • 如果是菜单，商品介绍等基础数据，也可以去使用canal订阅binlog的方式
  • 缓存数据+过期时间也足够解决大部分业务对于缓存的要求。
  • 通过加锁保证并发读写，写写的时候按顺序排好队。读读无所谓。所以适合使用读写锁。（业务不关心脏数据，允许临时脏数据可忽略）；

总结：

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

Canal

  • 直接更新数据库即可，canal相当于mysql的从库，他会根据sql更新，去redis中找到相关的数据进行修改
  • 也可以通过用户的访问记录，来向用户推荐商品。类似大数据

img

SpringCache

每次都那样写缓存太麻烦了，spring从3.1开始定义了Cache、CacheManager接口来统一不同的缓存技术。并支持使用JCache(JSR-107)注解简化我们的开发

Cache接口的实现包括RedisCache、EhCacheCache、ConcurrentMapCache等

在这里插入图片描述

每次调用需要缓存功能的方法时，spring会检查检查指定参数的指定的目标方法是否已经被调用过；如果有就直接从缓存中获取方法调用后的结果，如果没有就调用方法并缓存结果后返回给用户。下次调用直接从缓存中获取。

使用Spring缓存抽象时我们需要关注以下两点：

1、确定方法需要缓存以及他们的缓存策略

2、从缓存中读取之前缓存存储的数据

  • 依赖

  • 配置缓存类型

  • 配置需要的RedisCacheConfiguration

  • 启动类加注解@EnableCaching

  • 方法上加注解使用即可，比如@Cacheable等

    @Cacheable：表示当前方法的结果需要缓存，如果缓存中有，方法不用调用。如果没有，才会调用方法，最后将方法的结果放入缓存

配置

1、配置依赖

<dependency>
    <groupId>org.springframework.b oot</groupId>
    <artifactId>spring-boot-starter-cache</artifactId>
</dependency>

2、指定缓存类型并在主配置类上加上注解@EnableCaching

spring:
  cache:
  	#指定缓存类型为redis
    type: redis
    redis:
      # 指定redis中的过期时间为1h
      time-to-live: 3600000
      cache-null-values=true  # 缓存空值，解决缓存穿透问题

3、Cacheable默认使用jdk进行序列化（可读性差），默认ttl为-1永不过期，自定义序列化方式为json，需要编写配置类

@Configuration
@EnableConfigurationProperties(CacheProperties.class) // 导入CacheProperties，为了修改配置
public class MyCacheConfig {
    @Bean
    public RedisCacheConfiguration redisCacheConfiguration( CacheProperties cacheProperties) {
        // 拿到redis的配置
        CacheProperties.Redis redisProperties = cacheProperties.getRedis();
        
        org.springframework.data.redis.cache.RedisCacheConfiguration config = org.springframework.data.redis.cache.RedisCacheConfiguration
            .defaultCacheConfig(); // defaultCacheConfig默认的缓存配置 
        // 指定缓存序列化方式为json
        config = config.serializeValuesWith(
            RedisSerializationContext.SerializationPair.fromSerializer(new GenericJackson2JsonRedisSerializer())); // json
        // 设置配置文件中的各项配置，如过期时间
        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;
    }
}

缓存自动配置

  • 缓存自动配置源码：CacheAutoConfiguration==>RedisCacheConfiguration==>自动配置了RedisCacheManager==>初始化所有缓存==>每个缓存决定使用什么配置==>如果有RedisCacheConfiguration就用现有的，也就是我们配置的，没有就用默认的==>如果想改缓存的配置，只需要给容器中放一个RedisCacheConfiguration即可==>就会应用到RedisCacheManager管理的所有缓存分区中
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(CacheManager.class)
@ConditionalOnBean(CacheAspectSupport.class)
@ConditionalOnMissingBean(value = CacheManager.class, name = "cacheResolver")
@EnableConfigurationProperties(CacheProperties.class)
@AutoConfigureAfter({ CouchbaseAutoConfiguration.class, HazelcastAutoConfiguration.class,
                     HibernateJpaAutoConfiguration.class, RedisAutoConfiguration.class })
@Import({ CacheConfigurationImportSelector.class, // 看导入什么CacheConfiguration
         CacheManagerEntityManagerFactoryDependsOnPostProcessor.class })
public class CacheAutoConfiguration {

    @Bean
    @ConditionalOnMissingBean
    public CacheManagerCustomizers cacheManagerCustomizers(ObjectProvider<CacheManagerCustomizer<?>> customizers) {
        return new CacheManagerCustomizers(customizers.orderedStream().collect(Collectors.toList()));
    }

    @Bean
    public CacheManagerValidator cacheAutoConfigurationValidator(CacheProperties cacheProperties,
                                                                 ObjectProvider<CacheManager> cacheManager) {
        return new CacheManagerValidator(cacheProperties, cacheManager);
    }

    @ConditionalOnClass(LocalContainerEntityManagerFactoryBean.class)
    @ConditionalOnBean(AbstractEntityManagerFactoryBean.class)
    static class CacheManagerEntityManagerFactoryDependsOnPostProcessor
        extends EntityManagerFactoryDependsOnPostProcessor {

        CacheManagerEntityManagerFactoryDependsOnPostProcessor() {
            super("cacheManager");
        }

    }
    
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(RedisConnectionFactory.class)
@AutoConfigureAfter(RedisAutoConfiguration.class)
@ConditionalOnBean(RedisConnectionFactory.class)
@ConditionalOnMissingBean(CacheManager.class)
@Conditional(CacheCondition.class)
class RedisCacheConfiguration {

    @Bean // 放入缓存管理器
    RedisCacheManager cacheManager(CacheProperties cacheProperties, 
                                   CacheManagerCustomizers cacheManagerCustomizers,
                                   ObjectProvider<org.springframework.data.redis.cache.RedisCacheConfiguration> redisCacheConfiguration,
                                   ObjectProvider<RedisCacheManagerBuilderCustomizer> redisCacheManagerBuilderCustomizers,
                                   RedisConnectionFactory redisConnectionFactory, ResourceLoader resourceLoader) {
        RedisCacheManagerBuilder builder = RedisCacheManager.builder(redisConnectionFactory).cacheDefaults(
            determineConfiguration(cacheProperties, redisCacheConfiguration, resourceLoader.getClassLoader()));
        List<String> cacheNames = cacheProperties.getCacheNames();
        if (!cacheNames.isEmpty()) {
            builder.initialCacheNames(new LinkedHashSet<>(cacheNames));
        }
        redisCacheManagerBuilderCustomizers.orderedStream().forEach((customizer) -> customizer.customize(builder));
        return cacheManagerCustomizers.customize(builder.build());
    }

缓存使用@Cacheable、@CacheEvict

第一个方法存放缓存，第二个方法清空缓存

// 调用该方法时会将结果缓存，缓存的值是采用jdk序列化机制，将序列化后的结果放入redis。缓存名为category，key为方法名
// sync表示该方法的缓存被读取时会加锁
// value等同于cacheNames，表示分区指定缓存的名字，意思就是该缓存放到哪个地方用名字表示
// key如果是普通的字符串，要在里面加单引号，不然会当成动态取值 "'hello'"，#root.methodName是源码中的取值方法，意思就是取方法名作为缓存的key
@Cacheable(value = {"category"},key = "#root.methodName",sync = true)
public Map<String, List<Catalog2Vo>> getCatalogJsonDbWithSpringCache() {
    return getCategoriesDb();
}

// 调用该方法会删除缓存category下的所有cache，如果要删除某个具体，用key="''"
@Override
@CacheEvict(value = {"category"},allEntries = true)
public void updateCascade(CategoryEntity category) {
    this.updateById(category);
    if (!StringUtils.isEmpty(category.getName())) {
        categoryBrandRelationService.updateCategory(category);
    }
}

如果要清空多个缓存，用@Caching(evict={@CacheEvict(value="")})

SpringCache原理与不足

1、读模式

  • 缓存穿透：查询一个null数据。解决方案：缓存空数据，可通过spring.cache.redis.cache-null-values=true
  • 缓存击穿：大量并发进来同时查询一个正好过期的数据。解决方案：加锁 ? 默认是无加锁的;
    • 使用sync = true来解决击穿问题
  • 缓存雪崩：大量的key同时过期。解决：加随机时间。

2、写模式：（缓存与数据库一致）

  • 读写加锁。
  • 引入Canal，感知到MySQL的更新去更新Redis
  • 读多写多，直接去数据库查询就行

3、总结：

  • 常规数据（读多写少，即时性，一致性要求不高的数据，完全可以使用Spring-Cache）：

  • 写模式(只要缓存的数据有过期时间就足够了)

  • 特殊数据：特殊设计

举报

相关推荐

谷粒商城----通过缓存和分布式锁获取数据。

oraclemybatis数据库

分布式- 分布式锁

客户端redis加锁前端开发

谷粒商城分布式基础（一）—— 项目简介 & 分布式基础

服务器微服务分布式系统JavaScript前端开发

【分布式】Redis分布式锁

中间件redis分布式Redis分布式锁Redis锁

【分布式缓存】分布式缓存-缓存技术

缓存分布式java数据redis

【分布式锁】Redission实现分布式锁

sklearnpython机器学习

Redis与分布式-分布式锁

css前端

Java架构师之路四、分布式系统：分布式架构、分布式数据存储、分布式事务、分布式锁、分布式缓存、分布式消息中间件、分布式存储等。

数据结构算法

【分布式锁】Redisson分布式锁底层原理

人工智能大数据物联网自然语言处理神经网络

springboot实现分布式锁，java实现分布式锁，redisson实现分布式锁

springboot分布式锁java实现分布式锁redisson分布式锁集群实现分布式锁集群实现redisson分布式Java编程语言
0 条评论