继上一篇Redis缓存中介绍,我们可以很容易解决缓存穿透(空结果缓存),和缓存雪崩问题(加随机值),对于缓存击穿问题,可以采用加锁的方式,但是,这个锁需要用什么样的锁,怎么来加这个锁,也是非常有讲究的,一不小心就可能导致各种问题。
这里我们先测试一下,加本地锁情况下的一些问题。
@Override
public Map getCatalogJson() {
//给缓存中放json字符串,拿出的json字符串,还要逆转为能用的对象类型;【序列化与反序列化】
/**
* 1.空结果缓存:解决缓存穿透
* 2.设置过期时间(加随机值): 解决缓存雪崩
* 3.加锁: 解决缓存击穿
*/
//1. 加入缓存逻辑,缓存中存的数据是json字符串
//JSON好处:跨语言,跨平台兼容
String catalogJSON = redisTemplate.opsForValue().get("catalogJSON");
if (StringUtils.isEmpty(catalogJSON)) {
//2.缓存中没有,查询数据库
System.out.println("缓存不命中...查询数据库...");
Map catalogJsonFromDb = getCatalogJsonFromDb();
//每次调用这个方法,都有查询数据库,得到返回值,效率非常低下,我们需要加入缓存
//3. 查到的数据再放入缓存,将对象转为json放在缓存中
String s = JSON.toJSONString(catalogJsonFromDb);
redisTemplate.opsForValue().set("catalogJSON", s, 1, TimeUnit.DAYS);
return catalogJsonFromDb;
}
System.out.println("缓存命中...直接返回...");
//从缓存中获取的JSON,再转为我们指定的对象返回
Map result = JSON.parseObject(catalogJSON, new TypeReference(){});
return result;
}
//从数据库查询并封装分类数据
public Map getCatalogJsonFromDb() {
//只要是同一把锁,就能锁住需要这个锁的所有线程,使用this,this代表的就是当前对象
//1.synchronized(this):SpringBoot所有的组件在容器中都是单例的
/**
* 100万个请求同时进来,进来以后就先锁住,接下来这100万个请求就来竞争锁,假设有一个竞争上来了,那他就去执行数据库查询,查询完以后返回,释放锁
* 别的请求再一进来,再去查数据库就是不合理的,相当于我们虽然锁住了,相当于再去排队查数据库。
* 所以拿到锁以后,进来要做的第一件事,就是再看一下缓存里面有没有,如果有了说明是上一个人执行完放好的,如果没有你才需要再去查
*/
//TODO 本地锁:synchronized,JUC(Lock),在分布式情况下,想要锁住所有,必须使用分布式锁
synchronized (this) {
//得到锁以后,我们应该再去缓存中确定一次,如果没有才需要继续查询
String catalogJSON = redisTemplate.opsForValue().get("catalogJSON");
if (!StringUtils.isEmpty(catalogJSON)) {
//缓存不为null直接返回
Map result = JSON.parseObject(catalogJSON, new TypeReference(){});
return result;
}
System.out.println("查询了数据库....");
//将数据库的多次查询变为一次
Map parentCid = selectFromDB();
return parentCid;
}
}
这样加锁合适吗?
如果是在单体应用的情况下,也就是我们这个项目只会部署在一个Tomat里面,一台服务器,这样加锁没问题。但是,如果分布式,那就变了
如果分布式,我们常见的情况是把服务放在好多台服务器,假设大并发过来,由于负载均衡机制,现在没有个服务器,都接受1万个并发进来,而我们加锁加的是this,this的意思是指当前实例对象,无论是我们给同步代码块上加的this,还是方法上的this,这都是当前实例作为锁的,当前实例在我们容器中是单实例的,但是呢,我们一个项目一个容器,也就是说我们一个商品服务一个容器,这样呢,我们一个商品服务有八台机器,相当于我们有八台机器,八个容器就相当于我们有八个实例,所以说我们每一个this,只代表当前实例的容器对象。也就是说每一个this都是不同的锁,那相当于我们最终加了八把锁。
最终,导致的现象就是,我们商品服务一个this锁,相当于把10000请求锁住了,只有一个放进来了,然后,2号服务器,也相当于锁住了10000个,只有一个放进来了。最终,在分布式情况下,相当于有几台机器,我们就放了几个线程进来。相当于还是有八个线程同时进来,去查数据库相同的数据。
所以,我们说,本地锁,只能锁住当前进程,而我们现在如果要真正实现大并发百万请求,只留下一个进来查询数据库,我们就需要用分布式锁。分布式锁带来的缺点,就是它的性能比较慢,而我们当前的本地锁,稍微快一点,但是,我们本地锁的缺点,就是在分布式情况下,锁不住我们所有的服务。
但是,如果基于我们这种场景,我们用同步锁也是可以的,我们商品服务,哪怕放上一百台,现在有一百万的并发,我顶多给放一百个进来查数据库,这样呢我们数据库压力也不是很大,而且,我们锁也不用设计的那么重量级。
1.2 本地锁压力测试 1.2.1 问题演示1.先删除redis里面的key
我们想要的效果就是,在控制台一旦打印,缓存不命中,就开始查数据,并且数据库只能查一次,如果查了多次那就是加锁是失败的。
2. 压测
3. 测试结果
4.结果分析
如图,黑色表示上锁部分,1号线程结束释放锁,1号释放锁以后,还没有将结果放入缓存。2号线程获取到锁也进来了,它先要确认缓存中有没有,但是1号线程由于给redis放入数据,是一个网络交互,可能很慢,导致2号数据也没有在缓存中查到数据,然后也差了一遍数据库。所以,这就会导致我们查询了两遍数据库。
1.2.2 问题解决为了解决这个问题,我们可以把结果放入缓存这个操作,不要在释放锁了以后做,我们只要查到了数据库,就把结果放到缓存,这样就不会导致我们锁不住,查询两遍 数据库。
@Override
public Map getCatalogJson() {
//给缓存中放json字符串,拿出的json字符串,还要逆转为能用的对象类型;【序列化与反序列化】
/**
* 1.空结果缓存:解决缓存穿透
* 2.设置过期时间(加随机值): 解决缓存雪崩
* 3.加锁: 解决缓存击穿
*/
//1. 加入缓存逻辑,缓存中存的数据是json字符串
//JSON好处:跨语言,跨平台兼容
String catalogJSON = redisTemplate.opsForValue().get("catalogJSON");
if (StringUtils.isEmpty(catalogJSON)) {
//2.缓存中没有,查询数据库
System.out.println("缓存不命中...查询数据库...");
Map catalogJsonFromDb = getCatalogJsonFromDb();
//每次调用这个方法,都有查询数据库,得到返回值,效率非常低下,我们需要加入缓存
return catalogJsonFromDb;
}
System.out.println("缓存命中...直接返回...");
//从缓存中获取的JSON,再转为我们指定的对象返回
Map result = JSON.parseObject(catalogJSON, new TypeReference(){});
return result;
}
//从数据库查询并封装分类数据
public Map getCatalogJsonFromDb() {
//只要是同一把锁,就能锁住需要这个锁的所有线程,使用this,this代表的就是当前对象
//1.synchronized(this):SpringBoot所有的组件在容器中都是单例的
/**
* 100万个请求同时进来,进来以后就先锁住,接下来这100万个请求就来竞争锁,假设有一个竞争上来了,那他就去执行数据库查询,查询完以后返回,释放锁
* 别的请求再一进来,再去查数据库就是不合理的,相当于我们虽然锁住了,相当于再去排队查数据库。
* 所以拿到锁以后,进来要做的第一件事,就是再看一下缓存里面有没有,如果有了说明是上一个人执行完放好的,如果没有你才需要再去查
*/
//TODO 本地锁:synchronized,JUC(Lock),在分布式情况下,想要锁住所有,必须使用分布式锁
synchronized (this) {
//得到锁以后,我们应该再去缓存中确定一次,如果没有才需要继续查询
String catalogJSON = redisTemplate.opsForValue().get("catalogJSON");
if (!StringUtils.isEmpty(catalogJSON)) {
//缓存不为null直接返回
Map result = JSON.parseObject(catalogJSON, new TypeReference(){});
return result;
}
System.out.println("查询了数据库....");
//将数据库的多次查询变为一次
Map parentCid = selectFromDB();
//3. 查到的数据再放入缓存,将对象转为json放在缓存中
String s = JSON.toJSONString(parentCid);
redisTemplate.opsForValue().set("catalogJSON", s, 1, TimeUnit.DAYS);
return parentCid;
}
}
所以,我们一定要保证,查数据,查完以后放缓存,这是一个原子操作,在同一把锁内进行的,否则就会导致,我们整个释放锁的时序问题,导致我们查了两边数据库。
1.3 本地锁在分布式情况下问题 1.3.1 商品服务创建两个副本1.3.2 Jmeter压测
请求,由nginx来转到gateway,gateway负载均衡到我们的几个商品服务
每一个服务,都查询了一次数据库,以上说明,我们使用本地锁的情况下,不能锁住每一个微服务,我们本地锁的this只能锁住当前服务,为了锁住我们所有服务,就需要加一个分布式锁。
视频教程