Java 代码实现如下

public Map<String, List<Catalog2Vo>> getCatalogJsonDbWithRedisLock() {
        String uuid = UUID.randomUUID().toString();
        ValueOperations<String, String> ops = stringRedisTemplate.opsForValue();
        Boolean lock = ops.setIfAbsent("lock", uuid,5, TimeUnit.SECONDS);
        if (lock) {
            Map<String, List<Catalog2Vo>> categoriesDb = getCategoryMap();
            String lockValue = ops.get("lock");
            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();
            }
            return getCatalogJsonDbWithRedisLock();
        }
    }

总结而言，Redis 单节点实现分布式锁最重要的问题是 实现加锁和删除锁的原子性 。

基于 Redis 单节点的分布式锁基本完成了，但是这并不是一个完美的方案，只是相对完全一点，因为它并没有完全解决当前线程执行超时锁被提前释放后，其它线程乘虚而入的问题。

Redisson

Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。Redisson的宗旨是促进使用者对Redis的关注分离（Separation of Concern），从而让使用者能够将精力更集中地放在处理业务逻辑上。
Github：redisson/redisson

在本节中主要关注 Redisson 在分布式锁上的应用。

导入依赖：

<!--pom.xml-->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.13.4</version>
</dependency>

在容器中注入 Redisson 客户端：

@Configuration
public class RedissonConfig {
    @Bean
    public RedissonClient redissonClient(){
        Config config = new Config();
        //redis://127.0.0.1:7181
        //可以用"rediss://"来启用 SSL 连接
        config.useSingleServer().setAddress("redis://115.159.148.114:6380");
        RedissonClient redisson = Redisson.create(config);
        return redisson;
    }
}

可重入锁（Reentrant Lock）

什么是可重入锁？什么是不可重入锁？

假设存在两个方法，方法A，方法B，存在两把锁，锁1，锁2。

• 可重入锁：A需要锁1，B方法需要锁2。这两个方法不会发生死锁。
• 不可重入锁：A需要锁1，B也需要锁1。这两个方法可能发生死锁。

public Map<String, List<Catalog2Vo>> getCatalogJsonDbWithRedisson() {
        Map<String, List<Catalog2Vo>> categoryMap=null;
        // 只要名字相同就是同一把锁
        RLock lock = redissonClient.getLock("CatalogJson-Lock");
        // 上锁
        lock.lock();
        try {
            Thread.sleep(30000);
            categoryMap = getCategoryMap();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
            return categoryMap;
        }
    }

大家都知道，如果负责储存这个分布式锁的Redisson节点宕机以后，而且这个锁正好处于锁住的状态时，这个锁会出现锁死的状态。为了避免这种情况的发生，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期，只要占锁成功就会启动一个定时任务，重新给锁设置过期时间。默认情况下，看门狗的检查锁的超时时间是30秒钟，也可以通过修改Config.lockWatchdogTimeout来另行指定。每隔10秒钟都会重新对锁进行续期。

另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。不会自动续期！

// 加锁以后10秒钟自动解锁
// 无需调用unlock方法手动解锁,如果要手动解锁一定要确保业务执行时间小于锁的失效时间
lock.lock(10, TimeUnit.SECONDS);

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

在实际开发中一般是给锁限定一个过期时间，而不是采用默认的续期方式。

读写锁（ReadWriteLock）

与 经典进程同步问题 中的读写者模型类似。

1. 可以多个读者同时读
2. 但是不允许多个写者同时写, 只允许一个写者写
3. 不允许在写的时候读, 也不能有新的写者来写
4. 写者执行写之前, 不应该有读者或写者在操作

当然，Redisson 为我们封装了读写操作，我们不需要去关心线程的互斥问题。

@GetMapping("/read")
    @ResponseBody
    public String read() {
        RReadWriteLock lock = redissonClient.getReadWriteLock("ReadWrite-Lock");
        RLock rLock = lock.readLock();
        String s = "";
        try {
            rLock.lock();
            System.out.println("读锁加锁"+Thread.currentThread().getId());
            Thread.sleep(5000);
            s= redisTemplate.opsForValue().get("lock-value");
        }finally {
            rLock.unlock();
            return "读取完成:"+s;
        }
    }

    @GetMapping("/write")
    @ResponseBody
    public String write() {
        RReadWriteLock lock = redissonClient.getReadWriteLock("ReadWrite-Lock");
        RLock wLock = lock.writeLock();
        String s = UUID.randomUUID().toString();
        try {
            wLock.lock();
            System.out.println("写锁加锁"+Thread.currentThread().getId());
            Thread.sleep(10000);
            redisTemplate.opsForValue().set("lock-value",s);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            wLock.unlock();
            return "写入完成:"+s;
        }
    }

信号量（Semaphore）

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

常常用于限定流量的服务，当服务器无法承受高并发量的服务时可以限定流量。

@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）

相当于教室门，只有当学生走完了，教室门才可以关闭。

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

@GetMapping("/setLatch")
    @ResponseBody
    public String setLatch() {
        RCountDownLatch latch = redissonClient.getCountDownLatch("CountDownLatch");
        try {
            latch.trySetCount(5);
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "门栓被放开";
    }

    @GetMapping("/offLatch")
    @ResponseBody
    public String offLatch() {
        RCountDownLatch latch = redissonClient.getCountDownLatch("CountDownLatch");
        latch.countDown();
        return "门栓被放开1";
    }

缓存数据的一致性

缓存的数据如何保证与数据库中的一致。

双写模式

当数据更新时，更新数据库时同时更新缓存

存在的问题： 由于卡顿等原因，导致写缓存2在最前，写缓存1在后面就出现了不一致

这是暂时性的脏数据问题，但是在数据稳定，缓存过期以后，又能得到最新的正确数据。

失效模式

数据库更新时将缓存删除

当两个请求同时修改数据库，一个请求已经更新成功并删除缓存时又有读数据的请求进来，这时候发现缓存中无数据就去数据库中查询并放入缓存，在放入缓存前第二个更新数据库的请求成功，这时候留在缓存中的数据依然是第一次数据更新的数据

解决方法：

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

解决方案

无论是双写模式还是失效模式，都会导致缓存的不一致问题。即多个实例同时更新会出事。怎么办？

• 如果是用户纬度数据（订单数据、用户数据），这种并发几率非常小，不用考虑这个问题，缓存数据加上过期时间，每隔一段时间触发读的主动更新即可

• 如果是菜单，商品介绍等基础数据，也可以去使用canal订阅binlog的方式。

  

• 缓存数据+过期时间也足够解决大部分业务对于缓存的要求。

• 通过加锁保证并发读写，写写的时候按顺序排好队。读读无所谓。所以适合使用读写锁。（业务不关心脏数据，允许临时脏数据可忽略）；

总结：

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

SpringCache

简介

• Spring 从 3.1 开始定义了 org.springframework.cache.Cache 和 org.springframework.cache.CacheManager 接口来统一不同的缓存技术； 并支持使用 JCache（JSR-107）注解简化我们开发；

  

• Cache 接口为缓存的组件规范定义，包含缓存的各种操作集合； Cache 接口下 Spring 提供了各种 xxxCache 的实现；如 RedisCache ， EhCacheCache , ConcurrentMapCache 等；

  

  

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

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

  1. 确定方法需要被缓存以及他们的缓存策略
  2. 从缓存中读取之前缓存存储的数据

一个 CacheManager 管理着多种缓存 Cache。CacheManager 定义了一些规则，造出的缓存组件，这些缓存组件才是真正给缓存 CRUD 数据的。 RedissonSpringCacheManager 实现了 CacheManager，可以对存放在 Redis 的缓存进行管理。

简单上手

1. 导入依赖

<!--pom.xml-->
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-cache</artifactId>
</dependency>
<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>

1. 添加配置

spring:
  redis:
    host: 192.168.206.131
  cache:
    type:redis

1. 主类添加 @EnableCaching 注解开启缓存

示例代码

@SpringBootTest
@EnableCaching
class DemoApplicationTests {

	@Autowired
	StringRedisTemplate stringRedisTemplate;
	@Autowired
	ProductService productService;

	@Test
	public void test(){
		Product product = productService.readFromDb();
		System.out.println(product);
	}
}
@Service
class ProductService{

	@Cacheable(cacheNames = "product")
	public Product readFromDb(){
		System.out.println("在数据库中查询");
		return new DB().read();
	}
}
@Data
class Product implements Serializable {
	private String name;
	public Product(String name) {
		this.name = name;
	}
}
class DB{
	public Product read(){
		return new Product("MI");
	}
}

第一次执行查询

在 Redis 中新增了一个名为 product 的键值对

第二次执行，可见并没有执行 readFromDb() 方法，是直接从 Redis 中读取的缓存。

更多

SpringCache 提供的注解：

每一个需要缓存的数据需要我们来指定要放到哪个名字的缓存，缓存分区按照业务类型来划分。

@Cacheable(cacheNames = “product”) 代表当前方法的结果需要缓存，如果缓存中有方法就不调用。如果缓存中没有会调用方法，最后将方法的结果放入缓存。

默认行为：

1. 如果缓存中存在该方法的缓存，则方法不调用
2. key 默认生成，缓存的名字::SimpleKey[] 自动生成的 key 值
3. 缓存的 value 的值，默认使用 jdk 序列化机制，将序列化后的数据存放到 redis
4. 默认 ttl 的时间为 -1

自定义：

1. 指定生成的缓存使用的 key。key 属性指定，接收一个 SpEL：

   

2. 指定缓存数据的 ttl： 配置文件中修改 ttl

3. 将数据保存为 json 格式

前置原理：

CacheAutoConfiguration → RedisCacheConfiguration → 自动配置了 RedisCacheManager → 初始化所有的缓存 → 每个缓存决定使用什么配置 → 如果 redisCacheConfiguration 有就用已有的，没有就用默认配置单 →想要修改缓存的配置，只需要给容器中放一个 RedisCacheConfiguration 即可 →就会应用到当前 RedisCacheManager 管理的所有缓存分区中。

注入 RedisCacheConfiguration 的代码

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

测试结果如下

项目中如何使用

//调用该方法时会将结果缓存，缓存名为category，key为方法名
	//表示该方法的缓存被读取时会加锁
	@Cacheable(value = {"category"},key = "#root.methodName",sync = true)
    public Map<String, List<Catalog2Vo>> getCatalogJsonDbWithSpringCache() {
        return getCategoriesDb();
    }

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

如何级联更新所有关联的数据

@CacheEvict: 失效模式

1. 同时进行多种缓存操作
2. 指定删除分区下的所有数据 @CacheEvict( value = “catagory”, allEntries = true)
3. 存储同一类型的数据都可以指定成同一个分区。分区名默认是缓存的前缀。

SpringCache 的不足之处

读模式：

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

写模式（缓存与数据库的一致性问题）：

1. 读写加锁
2. 引入Canal,感知到MySQL的更新去更新Redis
3. 读多写多，直接去数据库查询就行

总结

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

写模式，只要缓存的数据有过期时间就足够了。

参考资料

• NiceSeason/gulimall-learning
• 浅析 Redis 分布式锁解决方案-InfoQ
• https://github.com/redisson/redisson/wiki/8.-%E5%88%86%E5%B8%83%E5%BC%8F%E9%94%81%E5%92%8C%E5%90%8C%E6%AD%A5%E5%99%A8
• StringRedisTemplate常用API_黑旋风李鬼的博客-CSDN博客
• 30. Cache Abstraction