redis分布式锁java实现解决缓存雪崩

128 阅读 0 评论 85 点赞

我是靠谱客的博主诚心烤鸡，这篇文章主要介绍redis分布式锁java实现解决缓存雪崩，现在分享给大家，希望可以做个参考。

缓存雪崩：因为缓存失效（key生存时间到期）导致所有请求都去查找数据库，导致数据库CPU和内存负载过高导致宕机。

缓存雪崩原因及解决方案：

使用缓存主要解决数据同步，并减少对数据库访问次数。因此，通常解决方案往往是使用互斥锁，让一个线程访问数据库，并将数据更新到缓存中，其他线程访问缓存中数据。如果是基于jvm锁机制的话，只能解决单机问题，也就是只让本机一个线程访问缓存，但是分布式条件下是不能使用的。所以，要基于缓存的分布式锁来实现。

以redis为例解释下实现分布式锁的原理:

获取锁：

所有线程操作一个共同的key比如:lock，如果redis中不存在key为lock的值，那么当前线程获取锁并为lock设置一个随机值。

如果lock已经存在了，说明已经有线程获取锁，该线程不能再获取了。

释放锁：

获取锁的线程操作执行完毕后，清除lock的值，这样锁就释放了。所以，对锁的操作就是通过对同一个key值的添加和删除操作。

代码：

复制代码

@Service
public class RedisLock implements Lock {
@Autowired
private JedisConnectionFactory factory;
private static final String LOCK="lock";
private ThreadLocal<String> local=new ThreadLocal<String>();
//获取锁
@Override
public boolean tryLock() {
//获取Jedis的原始数据连接
Jedis jedis = (Jedis)factory.getConnection().getNativeConnection();
String uuid = UUID.randomUUID().toString();
/** 获取锁：设置一个随机值，超期时间1s
String key, String value, String nxxx, String expx, int time)
nxxx: NX:key不存在时设值
XX:key存在时设值
expx: EX|PX, expire time units: EX = seconds; PX = milliseconds
*/
String ret = jedis.set(LOCK, uuid, "NX", "PX", 1000);
if(!StringUtils.isEmpty(ret)&&ret.equals("OK")){
local.set(uuid);
return true;
}
return false;
}
/**
* 解锁
*/
@Override
public void unlock() {
String script=null;
try {
script=FileCopyUtils.copyToString(new FileReader(ResourceUtils.getFile("classpath:cn/rjx/spring/cache/unlock.c")));
} catch (IOException e) {
e.printStackTrace();
}
Jedis jedis = (Jedis)factory.getConnection().getNativeConnection();
List<String> keys=new ArrayList<String>();
keys.add(LOCK);
List<String> args=new ArrayList<String>();
args.add(local.get());
//如果redis中的 lock值和当前线程的uuid值相等，删除Key值
jedis.eval(script, keys, args);
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@Service
public class RedisLock implements Lock {
@Autowired
private JedisConnectionFactory factory;
private static final String LOCK="lock";
private ThreadLocal<String> local=new ThreadLocal<String>();
//获取锁
@Override
public boolean tryLock() {
//获取Jedis的原始数据连接
Jedis jedis = (Jedis)factory.getConnection().getNativeConnection();
String uuid = UUID.randomUUID().toString();
/** 获取锁：设置一个随机值，超期时间1s
String key, String value, String nxxx, String expx, int time)
nxxx: NX:key不存在时设值
XX:key存在时设值
expx: EX|PX, expire time units: EX = seconds; PX = milliseconds
*/
String ret = jedis.set(LOCK, uuid, "NX", "PX", 1000);
if(!StringUtils.isEmpty(ret)&&ret.equals("OK")){
local.set(uuid);
return true;
}
return false;
}
/**
* 解锁
*/
@Override
public void unlock() {
String script=null;
try {
script=FileCopyUtils.copyToString(new FileReader(ResourceUtils.getFile("classpath:cn/rjx/spring/cache/unlock.c")));
} catch (IOException e) {
e.printStackTrace();
}
Jedis jedis = (Jedis)factory.getConnection().getNativeConnection();
List<String> keys=new ArrayList<String>();
keys.add(LOCK);
List<String> args=new ArrayList<String>();
args.add(local.get());
//如果redis中的 lock值和当前线程的uuid值相等，删除Key值
jedis.eval(script, keys, args);
}
}

删除键值是执行的脚本unlock.c：

复制代码

1
2
3
4
5
if redis.call("get",KEYS[1])==ARGV[1] then
return redis.call("del",KEYS[1])
else
return 0
end

操作缓存的具体流程：

1.当线程查询某一值时先查看缓存是否存在该值。

2.如果存在直接返回主缓存中的数据。

3.1如果不存在，只有一个线程获取锁并去数据库读取数据，读取后更新主缓存和备份缓存。

3.2 其他线程取备份缓存中的数据。.

代码实现：初始时，主缓存和备份缓存为空，此时可能会有线程获取的值为空，但是并不影响用户体验，用户可以再刷新一次。在要求比较高的场景里面，可以考虑先把数据写入缓存中，可以搭配定时刷新缓存的机制。

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public List<Integer> queryCountByLeiMu() {
List<Integer> cacheResult = cacheService.cacheResult("101", "leimu");
if(cacheResult!=null){
logger.info("================get cache=======================");
return cacheResult;
}
if(lock.tryLock()){
logger.info("================get db=======================");
List<Integer> list=empDao.queryCountByLeiMu();
cacheService.cachePut("101", list, "leimu");//主缓存
cacheService.cachePut("beifen101", list, "beifenleimu");//备份缓存
lock.unlock();
return list;
}else{
logger.info("================get BEIFEN=======================");
//备份中拿
return cacheService.cacheResult("beifen101", "beifenleimu");
}
}

数据同步问题：主缓存中key的过期时间比较短，这样保证尽可能获取新数据。

bean.xml中缓存失效时间设置：

复制代码

1
2
3
4
5
6
7
8
9
10
11
12

<!-- 开启缓存注解扫描 -->
<cache:annotation-driven />
<bean id="cacheManager" class="org.springframework.data.redis.cache.RedisCacheManager">
<constructor-arg index="0" ref="redisTemplate"></constructor-arg>
<property name="expires">
<map>
<entry key="leimu" value="5"></entry>
<entry key="beifenleimu" value="100"></entry>
</map>
</property>
</bean>

测试方法模拟高并发情景：

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@Autowired
LeiMuService leiMuService;
private static final int threadNum=13;
//倒计数器（发令枪）
用于制造线程并发执行
private static CountDownLatch cdl=new CountDownLatch(threadNum);
/**
* 模拟高并发条件下，数据库查询耗时比较长
* @throws InterruptedException
*/
@Test
public void test04() throws InterruptedException{
for(int i=0;i<threadNum;i++){
new Thread(new UserRequest()).start();
cdl.countDown();//threadNum每次减1，到零时同时执行cdl.await();后边代码
}
//主线程挂起，等子线程执行完以后
Thread.currentThread().join();
}
private class UserRequest implements Runnable{
@Override
public void run() {
//所有子线程在这里等待，当所有线程实例化后，同时停止等待
try {
cdl.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
//N个子线程同时调用获取类目
List<Integer> leimu = leiMuService.queryCountByLeiMu();
logger.info(Thread.currentThread().getName()+"==========================================>"+leimu.size());
}
}
}