Redission的基本使用

比如现在有一个电商网站，由于商城的访问人数越来越多，最高访问人数达到了每秒1万请求，导致用户在商城首页中等待的时间越来越长，使用jemter 测试的吞吐量很低，我们需要考虑优化首页。

假设我们现在已经使用nginx进行了动静分离，给数据库添加了索引，调大了jvm的内存，现在需要考虑是否缓存首页数据，因为我们首页的分类菜单很少改变，却请求很多，可以把数据库缓存到redis中。

使用Redis

基本使用

引用依赖

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

配置参数

讯享网spring redis: port: 6379 host: 127.0.0.1 password: xxxx 

springBoot帮我们封装了redisTemplate，本次使用的是StringRedisTemplate，以下是初步使用缓存保存数据

讯享网

 private static final String INDEX_DATA="categoryjson"; // 从数据库中获取数据 private Map<String, List<IndexCategoryVo>> getCategoryFromDb(){ ValueOperations<String, String> opsForValue = stringRedisTemplate.opsForValue(); String dataStr = opsForValue.get(CategoryServiceImpl.INDEX_DATA); if(StringUtils.isEmpty(dataStr)){ // 从数据库中查询 System.out.println("查询了数据库"); Map<String, List<IndexCategoryVo>> collect = categoryOneLevel(); // 放入缓存 String jsonString = JSON.toJSONString(collect); opsForValue.set(CategoryServiceImpl.INDEX_DATA,jsonString); return collect; } return JSON.parseObject(dataStr, new TypeReference<Map<String, List<IndexCategoryVo>>>() {}); } 

以上代码使用jemter进行测试会出现大量的请求错误，原因springBoot在2.0以后RedisTemplate底层默认使用的是Lettuce客户端操作Redis的,而Lettuce是使用netty进行网络通信的，netty如果没有指定堆外内存，默认使用Options中的-Mmx300m,Lettuce的bug会导致内存不够(没有及时的释放内存)，导致堆外内存溢出，就算调大了内存早晚还是会产生该问题

解决办法:使用Jedis客户端，但是缺点就是性能相对来说较低
依赖排除,引入Jedis依赖(无需引入版本号，springBoot帮我们控制了版本号)

讯享网 <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> 

使用Redis会产生的问题

1. 缓存穿透：比如大量的请求访问一个缓存和数据库中都没有数据，到缓存中查询没有，请求就会全部到达数据库，导致数据库压力过大而崩溃
   解决办法：缓存空对象并设置失效时间，布隆过滤器
2. 缓存雪崩：比如程序中缓存数据库设置相同的过期时间，在某一刻数据全部过期，导致请求全部到数据库中查询，一瞬间导致数据库压力过大而崩溃
   解决办法：缓存数据的过期时间设置随机，防止同一时间大量数据过期现象发生。如果缓存数据库是分布式部署，将热点数据均匀分布在不同缓存数据库中。设置热点数据永远不过期。
3. 缓存击穿：比如有10万请求请求同一数据，就在请求到达的前一瞬间缓存数据过期了，请求就会全部到达数据库，导致数据库压力过大而崩溃，理想状态：只有一个请求到数据库查询数据，其他请求到缓存中取数据
   解决办法：使用锁，请求到达后在缓存中查询一次，有就直接返回，没有就获取锁，没有获取到锁睡100ms后再次尝试获取锁，获取到锁之后再查询一次缓存中有没有数据，有就直接返回，没有就真正查询数据库，查询后把数据放入缓存中再释放锁，其他请求进来就会在缓存中获取数据直接返回

 private static final String INDEX_DATA="categoryjson"; // 从数据库中获取数据 private Map<String, List<IndexCategoryVo>> getIndexCategoryMapLocalLock(){ ValueOperations<String, String> opsForValue = stringRedisTemplate.opsForValue(); String dataStr = opsForValue.get(CategoryServiceImpl.INDEX_DATA); if(StringUtils.isEmpty(dataStr)){ // 加锁 synchronized (this){ return this.getCategoryFromDb(); } }else{ this.getIndexCategoryMapLocalLock(); } } // 从数据库中获取数据 private Map<String, List<IndexCategoryVo>> getCategoryFromDb(){ ValueOperations<String, String> opsForValue = stringRedisTemplate.opsForValue(); String dataStr = opsForValue.get(CategoryServiceImpl.INDEX_DATA); if(StringUtils.isEmpty(dataStr)){ // 从数据库中查询 System.out.println("查询了数据库"); Map<String, List<IndexCategoryVo>> collect = categoryOneLevel(); // 放入缓存 String jsonString = JSON.toJSONString(collect); opsForValue.set(CategoryServiceImpl.INDEX_DATA,jsonString); return collect; } return JSON.parseObject(dataStr, new TypeReference<Map<String, List<IndexCategoryVo>>>() {}); } 

讯享网 public Map<String, List<IndexCategoryVo>> getIndexList() { ValueOperations<String, String> opsForValue = stringRedisTemplate.opsForValue(); String dataStr = opsForValue.get(CategoryServiceImpl.INDEX_DATA); if(StringUtils.isEmpty(dataStr)){ Map<String, List<IndexCategoryVo>> indexCategoryMap = this.getIndexCategoryMapDispersedLock(); String jsonString = JSON.toJSONString(indexCategoryMap); opsForValue.set(CategoryServiceImpl.INDEX_DATA,jsonString); } return JSON.parseObject(dataStr, new TypeReference<Map<String, List<IndexCategoryVo>>>() {}); } // 分布式锁 public Map<String, List<IndexCategoryVo>> getIndexCategoryMapDispersedLock() { ValueOperations<String, String> ops = stringRedisTemplate.opsForValue(); // 占锁，谁占到谁就查询 Boolean lock = ops.setIfAbsent("lock","xxx"); if(lock){ // 执行业务，业务中再次查询缓存数据并缓存数据 Map<String, List<IndexCategoryVo>> data = this.getCategoryFromDb(); // 解锁 ops.del("lock"); return data; }else { try { Thread.sleep(300); } catch (InterruptedException e) { e.printStackTrace(); } this.getIndexCategoryMapLocalLock(); } return null; } // 从数据库中获取数据 private Map<String, List<IndexCategoryVo>> getCategoryFromDb(){ ValueOperations<String, String> opsForValue = stringRedisTemplate.opsForValue(); String dataStr = opsForValue.get(CategoryServiceImpl.INDEX_DATA); if(StringUtils.isEmpty(dataStr)){ // 从数据库中查询 System.out.println("查询了数据库"); Map<String, List<IndexCategoryVo>> collect = categoryOneLevel(); // 放入缓存 String jsonString = JSON.toJSONString(collect); opsForValue.set(CategoryServiceImpl.INDEX_DATA,jsonString); return collect; } return JSON.parseObject(dataStr, new TypeReference<Map<String, List<IndexCategoryVo>>>() {}); } 

以上代码还是有问题，我们就对getIndexCategoryMapDispersedLock方法进行改造
比如这种情况：
程序执行到占锁之后，解锁之前的时候，突然断电或者程序异常导致锁不能释放成为了死锁，为了避免出现这种情况，我们需要给锁加上过期时间,到一定时间自动释放避免死锁，并且得保持原子性

public Map<String, List<IndexCategoryVo>> getIndexCategoryMapDispersedLock() { // 占锁，谁占到谁就查询 ValueOperations<String, String> ops = stringRedisTemplate.opsForValue(); // 为了避免占锁后突然程序断电或停止等导致死锁，需要给锁加上过期时间，占锁和加锁必须保持原子性 Boolean lock = ops.setIfAbsent("lock", "xxx",60, TimeUnit.SECONDS); if(lock){ // 执行业务，业务中再次查询缓存数据并缓存数据 Map<String, List<IndexCategoryVo>> data = this.getCategoryFromDb(); String value = ops.get("lock"); // 解锁 ops.del("lock"); return data; }else { try { Thread.sleep(300); } catch (InterruptedException e) { e.printStackTrace(); } this.getIndexCategoryMapLocalLock(); } return null; } 

讯享网 public Map<String, List<IndexCategoryVo>> getIndexCategoryMapDispersedLock() { // 占锁，谁占到谁就查询 ValueOperations<String, String> ops = stringRedisTemplate.opsForValue(); // 为了避免占锁后突然程序断电或停止等导致死锁，需要给锁加上过期时间，占锁和加锁必须保持原子性 String uuid = UUID.randomUUID().toString(); Boolean lock = ops.setIfAbsent("lock", uuid,60, TimeUnit.SECONDS); if(lock){ // 执行业务，业务中再次查询缓存数据并缓存数据 Map<String, List<IndexCategoryVo>> data = this.getCategoryFromDb(); String value = ops.get("lock"); // 解锁 // 为了避免业务执行时间太长，锁自动过期 // 其他线程拿到锁线程B正在执行业务而线程A业务才执行结束解锁时解的是别人的锁 // 导致没有锁住，所以解锁时需要判断占锁时的value值是否一样，一样再解锁，解锁和判断值必须保证原子性 // 还存在问题过期时间小于业务执行时间，需要续期过期时间，这个技术由分布式redission解决 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"),value); return data; }else { try { Thread.sleep(300); } catch (InterruptedException e) { e.printStackTrace(); } this.getIndexCategoryMapLocalLock(); } return null; } 

业务没执行完就释放锁的问题可以把过期时间设置长一点，或者使用Redission看门狗的原理，使用一条线程每10秒执行一次，如果锁还没有释放则自动续期锁的过期时间

以上代码写的太多过于繁琐，我们可以使用更专业的Redission分布式锁来解决

使用Redsiion

引入依赖

 <dependency> <groupId>org.redisson</groupId> <artifactId>redisson</artifactId> <version>3.13.4</version> </dependency> 

程序配置

讯享网 // redission通过redissonClient对象使用 // 如果是多个redis集群，可以配置 @Bean(destroyMethod = "shutdown") public RedissonClient redisson(){ Config config = new Config(); // 创建单例模式的配置 config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("xudaze"); return Redisson.create(config); } 

开始使用，改造上面的加锁代码

 @Autowired private RedissonClient redissonClient; public Map<String, List<IndexCategoryVo>> getIndexCategoryMapRedissionLock() { // 占锁 没有拿到锁的会自动阻塞 // 锁自动加了默认30秒过期 // 如果业务代码耗时长，锁也会自动续期 RLock lock = redissonClient.getLock(CategoryServiceImpl.LOCK); lock.lock(); //lock.lock(10,TimeUnit.SECONDS); 这种是10秒后锁自动过期，不会有自动续期的机制 //boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS); 尝试加锁，最多等待100秒，上锁以后10秒自动解锁 try { // 模拟业务执行了40s,测试是否有自动续期 Thread.sleep(40000); Map<String, List<IndexCategoryVo>> data = this.getCategoryFromDb(); return data; } catch (InterruptedException e) { e.printStackTrace(); } finally { lock.unlock(); } return null; } 

读写锁

基于Redis的Redisson分布式可重入读写锁RReadWriteLock Java对象实现了java.util.concurrent.locks.ReadWriteLock接口。其中读锁和写锁都继承了RLock接口。

讯享网 / * 测试读写锁 * @return */ @ResponseBody @GetMapping("index/read") public String testRead() throws InterruptedException { RReadWriteLock lock = redisson.getReadWriteLock("test-wr-lock"); // 拿到读锁 RLock rLock = lock.readLock(); rLock.lock(); // 模拟业务数据执行了20s //Thread.sleep(20000); rLock.unlock(); return "读完毕"; } / * 测试读写锁 * @return */ @ResponseBody @GetMapping("index/write") public String testWrite() throws InterruptedException { RReadWriteLock lock = redisson.getReadWriteLock("test-wr-lock"); // 拿到写锁 RLock rLock = lock.writeLock(); rLock.lock(); // 模拟业务数据执行了20s Thread.sleep(20000); rLock.unlock(); return "写完毕"; } 

测试得出:
读+读 没有任何影响
写+读 读请求在写请求没有执行完毕的情况下会处于阻塞状态，等写完毕之后才会拿到最新请求
写+写 会依次排队，没拿到锁的请求后阻塞
读+写 会等读请求读完毕之后，写请求才会执行

信号量

信号量为存储在redis中的一个数字，当这个数字大于0时，即可以调用acquire()方法增加数量，也可以调用release()方法减少数量，但是当调用release()之后小于0的话方法就会阻塞，直到数字大于0
模拟停车场
/
* 测试信号量
* 初始化停车场
* @return
*/
@ResponseBody
@GetMapping(“index/initSemaphore”)
public String initSemaphore() throws InterruptedException {
RSemaphore semaphore = redisson.getSemaphore(“semaphore”);
semaphore.release(10);
return “初始化成功”;
}

/ * 测试信号量 * 模拟进库 * @return */ @ResponseBody @GetMapping("index/inSemaphore") public String inSemaphore() throws InterruptedException { // 进库车位就 -1 RSemaphore semaphore = redisson.getSemaphore("semaphore"); if(semaphore.tryAcquire(2,TimeUnit.SECONDS)){ semaphore.acquire(1); return "进库"; } return "车位已满"; } / * 测试信号量 * 模拟出库 * @return */ @ResponseBody @GetMapping("index/outSemaphore") public String outSemaphore() throws InterruptedException { // 出库车位 +1 RSemaphore semaphore = redisson.getSemaphore("semaphore"); semaphore.release(1); return "出库"; } 

闭锁

讯享网 / * 模拟学校锁门 * 保安锁门 * @return */ @ResponseBody @GetMapping("index/studentlockroom") public String studentlockroom() throws InterruptedException { RCountDownLatch room = redisson.getCountDownLatch("room"); // 设置几个班,必须要等5个班的同学都走完之后再锁门 room.trySetCount(5); room.await(); return "锁门"; } / * 模拟学校锁门 * 班级放假 * @return */ @ResponseBody @GetMapping("index/gogogo/{id}") public String gogogo(@PathVariable("id")String id) throws InterruptedException { RCountDownLatch room = redisson.getCountDownLatch("room"); room.countDown(); return id+"班放假了"; } 

数据一致性

我们使用缓存后如果需要保持数据一致性怎么办
解决方案：
双写模式： 写数据库的同时写缓存，但是可能会出现以下情况

如果对某些数据的实时性要求不是很高的话可以设置过期时间，到时间会自动更新缓存，但如果要改动了之后需要立马更新到缓存中的解决办法就是加锁或加读写锁即可解决上面的问题
失效模式：数据库写完之后删除缓存，由查询数据那里更新缓存，但是有可能也会出现以下这种情况

假设现在有请求ABC同时到达，但由于各种原因导致它们处理数据的速度不一致，请求A到达后处理的最快，已经把数据库更新了并且把缓存删除了。请求B正在写数据库但是还没有写完，这是请求C读取数据的时候发现缓存中没有，就去数据库查询到了请求A修改的数据库，这时候请求B才把数据库写完然后删除缓存，然后请求C更新请求A修改的值到缓存中，导致数据库的值是请求B写的，缓存中的值是请求A写的，数据不一致问题，但这些也只是暂时性的脏读，等到过期时间到了即可一致。如果非要保证数据一致性高的话可用读写锁解决问题。