缓存与分布式锁—

文章目录

版本 1.0 本地锁
- 原生场景
- 缓存
- 缓存击穿
版本 2.0 分布式锁
- 基本原理
- SETNX
- 分布式锁-阶段 1
- 分布式锁-阶段 2
- 分布式锁-阶段 3
- 分布式锁-阶段 4
- 分布式锁-阶段 5

版本 1.0 本地锁

原生场景

进入一个网站首页，首页向后端发送请求，后端向数据库请求几百个数据的商品分类信息。

向数据库查询分类信息

// 向数据库查询分类信息
String getCatalogJsonFromDB(){String res = null;/*get res from db*/return res;
}

缓存

考虑到网站点击量较高，且商品分类请求方法需要查询数据库，加上数据封装，时间较长，所以要将查询到的数据缓存到 redis 中。忽略数据格式的转化。

String getCatalogJsonFromRedis(){// 尝试向 redis 获取数据String catalogJson = redisTemplate.opsForValue().get("catalogJson");if(StringUtils.isEmpty(catalogJson)){// redis 中没有，向数据库查String catalogJsonFromDB = getCatalogJsonFromDB();// 写入 redisredisTemplate.opsForValue().set("catalogJson", catalogJsonFromDB);return catalogJsonFromDB;}return catalogJson;
}

逻辑也很简单，后端直接向 redis 请求数据，如果不存在的话，再向数据库请求，且把数据库请求得来的数据放到 redis 中。

缓存击穿

相关介绍

作为一个网站的首页，访问量较高，虽设置了缓存，但是缓存有过期时间，过期的时刻，本来由 redis 承担的请求全部打在数据库上，容易造成数据库的宕机。

解决方案：使用互斥锁，只让一个请求去访问数据库。

重写 getCatalogJsonFromDB 方法

String getCatalogJsonFromDB(){synchronized (this){String catalogJson = redisTemplate.opsForValue().get("catalogJson");// 双重校验if(!StringUtils.isEmpty(catalogJson)){return catalogJson;}String res = null;/*get res from db*/return res;}
}

使用 synchronized (this) 的原因是，springboot 生成的对象都是单例模式，可以认为锁到所有访问对象。双重校验，如果前面有线程查询了数据库且向 redis 存放了数据，可以再从 redis 返回，不需要查库。

预想中是一次查库，其余查缓存。

问题： 线程 A、B 发来请求，redis 中数据为空，进入到 getCatalogJsonFromDB 方法。线程 A 争夺到锁，进入代码块，从数据库查询到数据并返回，释放锁之后，向 redis 中存储数据。在这个过程里，线程 B 得到锁， redis 中数据还没存进去，于是继续查库。降低了系统性能

可以将写入缓存这一操作放在释放锁之前

String getCatalogJsonFromDBAndCache(){synchronized (this){String catalogJson = redisTemplate.opsForValue().get("catalogJson");// 双重校验if(!StringUtils.isEmpty(catalogJson)){return catalogJson;}String res = null;/*get res from db*/redisTemplate.opsForValue().set("catalogJson", res);return res;}
}

这样可以实现一次查库，其余查缓存。前提是单体服务。

这里主要利用 synchronized (this) 实现加锁功能，之前说这样可行是因为在一个 springboot 微服务里，每个对象都是单例模式的，所以锁 this 相当于锁所有请求。

但是如果分布式呢？

这里复制 3 个相同的服务，同时启动，使用 JMeter 向网关发送请求。

结果是启动的四个微服务里，有三个查询了数据库，意味着有三个也写了缓存。

这样难免会造成分布式的微服务下数据的不一致性。

版本 2.0 分布式锁

基本原理

虽然服务是分布式的，但是目前来看， redis 只有一个，所以只需要给 redis 加锁，即可实现只有一个微服务下的一个线程向 redis 存储数据。

SETNX

SET key value [EX seconds] [PX milliseconds] [NX|XX]

EX：设置键key的过期时间，单位秒
PX：设置键key的过期时间，单位毫秒
NX：只有键key不存在的时候才会设置key的值
XX：只有键key存在的时候才会设置key的值

NX 这个参数引导，每个微服务下的线程竞争设置一个为 “lock” 的 key，只有一个可以设置成功。

分布式锁-阶段 1

不再使用本地锁，将上述 getCatalogJsonFromDBAndCache去除本地锁。

private String getCatalogJsonFromDbWithRedisLock() {// 去 redis 加锁Boolean successful = redisTemplate.opsForValue().setIfAbsent("lock", "lock");if(successful){// 设置过期时间redisTemplate.expire("lock", 30, TimeUnit.SECONDS);// 加锁成功 执行业务String res = getCatalogJsonFromDbAndCache();// 执行完 删除锁redisTemplate.delete("lock");return res;}else{// 加锁失败 自旋try {Thread.sleep(200);} catch (InterruptedException e) {}return getCatalogJsonFromDbWithRedisLock();}
}

如果不删除锁？其他线程阻塞，一直自旋，导致死锁
如果不设置过期时间？删除锁之前停电/异常，无法删除，导致死锁

分布式锁-阶段 2

加锁的时候同时设置过期时间，必须保证原子性

Boolean successful = redisTemplate.opsForValue().setIfAbsent("lock", "lock", 300, TimeUnit.SECONDS);

删锁的时候锁已经过期的话，其他线程已经获取锁，设置了自己的 lock，此时再删锁其实删的是别人的锁，多个线程会同时访问。

分布式锁-阶段 3

设置 key-value 时创建一个独特的 UUID ，删除时候判断是否是自己的锁。如果是，才删除。

private String getCatalogJsonFromDbWithRedisLock() {// 去 redis 加锁String uuid = UUID.randomUUID().toString();Boolean successful = redisTemplate.opsForValue().setIfAbsent("lock", uuid, 300, TimeUnit.SECONDS);if(successful){// 设置过期时间// 加锁成功 执行业务String res = getCatalogJsonFromDbAndCache();String lock = redisTemplate.opsForValue().get("lock");if(uuid.equals(lock)){// 执行完 删除锁redisTemplate.delete("lock");}return res;}else{// 加锁失败 自旋try {Thread.sleep(200);} catch (InterruptedException e) {}return getCatalogJsonFromDbWithRedisLock();}
}

问题也相当明显，判断锁和删锁也不是原子性的，也会出现误删的情况。判断的时候是自己的锁，然后锁到期，其他线程占锁，删除了别人的。

分布式锁-阶段 4

官方文档建议使用 lua 脚本执行原子性操作

if redis.call("get",KEYS[1]) == ARGV[1] thenreturn redis.call("del",KEYS[1])
elsereturn 0
end

最终版本

private String getCatalogJsonFromDbWithRedisLock() {String uuid = UUID.randomUUID().toString();// 原子加锁Boolean successful = redisTemplate.opsForValue().setIfAbsent("lock", uuid, 300, TimeUnit.SECONDS);if(successful){String res = null;try {// 加锁成功 执行业务res = getCatalogJsonFromDbAndCache();} catch (Exception e) {// lua 脚本String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";// 原子解锁redisTemplate.execute(new DefaultRedisScript<Long>(script),Arrays.asList("lock"),uuid);}return res;}else{// 加锁失败 自旋try {Thread.sleep(200);} catch (InterruptedException e) {}return getCatalogJsonFromDbWithRedisLock();}
}

只有一个微服务中的一次获取分布式锁成功

分布式锁-阶段 5

Redisson 实现

Redisson 相关内容

private String getCatalogJsonFromDbWithRedissonLock(){RLock lock = redissonClient.getLock("catalogJSON-lock");lock.lock();String catalogJsonFromDb;try {catalogJsonFromDb = getCatalogJsonFromDbAndCache();} finally {lock.unlock();}return catalogJsonFromDb;
}