1.本地锁实现
如下图所示，共有8个商品服务，每个商品服务都有10000个并发去查询数据库，用本地锁去实现，只能锁住单个服务，在同一时间，还是会有8个线程去查询数据库。
本地锁
test1服务：核心代码
@Service
public class HelloServiceImpl implements HelloService {
@Resource
private StringRedisTemplate stringRedisTemplatel;
@Override
public String hello() {
String testString = stringRedisTemplatel.opsForValue().get("testString");
// 如果有数据直接返回
if (!StringUtils.isEmpty(testString)) {
System.out.println("返回缓存中的数据！");
return testString;
}
// 模拟调用查询数据库方法，获取数据
String dateFromDb = getDateFromDb();
// 如果在此处放入缓存，可能会导致同一服务多次去查询数据库
// stringRedisTemplatel.opsForValue().set("testString", dateFromDb);
return dateFromDb;
}
/**
* 只需要给查数据库方法加锁，保证在当前服务中只有一个线程去查数据库
* @return
*/
private synchronized String getDateFromDb() {
//再次确认缓存中没有数据
String testString = stringRedisTemplatel.opsForValue().get("testString");
// 如果有数据直接返回
if (!StringUtils.isEmpty(testString)) {
System.out.println("返回缓存中的数据！");
return testString;
}
System.out.println("查询数据库");
// 模拟查到的数据结果
String date = "testString";
/**
* 查询到结果后，立即放入缓存中，防止下一个线程重复去数据库查询。
*
* 如果不在这放入缓存，该方法返回后，下一个得到锁的线程，去缓存中查看时，
* 有可能因为其他因素，上一个线程没有来得及将结果放入缓存，因此被当前线程判定缓存中没有数据
* 致使该线程又重新去查一遍数据库。
*
*/
stringRedisTemplatel.opsForValue().set("testString", date);
return date;
}
}
test2服务：用openfeign去远程调用test1
// 指定微服务名称
@FeignClient("test1")
public interface HelloServiceFeign {
// 调用test1
@GetMapping("/hi")
String hello();
}
先把redis缓存清空，再用Jmeter去压测，20个线程循环2次
image.png
查看每个微服务控制台，可以看到都只有一个
查询数据库输出，说明在每个服务中，都已经实现锁，但还是有三个线程去查了数据库
test1 8081端口
test1 8082端口
test1 8083端口
分布式锁
从上面例子可以看出，本地锁无法满足需求，接下来，可以尝试，让各个服务都去相同的一个地方获得锁，如果获取到锁，就执行业务逻辑，否则就等待别人释放锁。
去相同的一个地方获得锁，可以是redis,可以是数据库，可以去任何大家都能访问的地方。
等待可以用自旋的方式。
分布式锁原理
用redis实现
redis有一个set命令，
SET key value [EX seconds] [PX milliseconds] [NX|XX],使用的时候带后面的NX参数，如果要set的key已经存在，则会失败返回，只有key不存在时，才可以set成功，因此就可以用这个表示获取锁的成功与失败。
分布式锁阶段一
分布式锁阶段一
例子
@Service
public class HelloServiceImpl implements HelloService {
@Resource
private StringRedisTemplate stringRedisTemplatel;
@Resource
private RedisTemplate redisTemplate;
@Override
public String hello() {
String testString = stringRedisTemplatel.opsForValue().get("testString");
// 如果有数据直接返回
if (!StringUtils.isEmpty(testString)) {
System.out.println("返回缓存中的数据！");
return testString;
}
// 模拟调用查询数据库方法，获取数据
String dateFromDb = null;
try {
dateFromDb = getDateFromDbWithRedisLock();
} catch (InterruptedException e) {
e.printStackTrace();
}
// 如果在此处放入缓存，可能会导致同一服务多次去查询数据库
// stringRedisTemplatel.opsForValue().set("testString", dateFromDb);
return dateFromDb;
}
/**
* 只需要给查数据库方法加锁，保证在当前服务中只有一个线程去查数据库
*
* @return
*/
private synchronized String getDataFromDbWithLocalLock() {
return getDataFromDB();
}
private String getDateFromDbWithRedisLock() throws InterruptedException {
// 1.获取分布式锁
Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "lock");
if (lock) {
//加锁成功
String dataFromDB = getDataFromDB();
//删除锁
redisTemplate.delete("lock");
return dataFromDB;
} else {
//加锁失败，重试(自旋)
TimeUnit.MILLISECONDS.sleep(1000);
return getDateFromDbWithRedisLock();
}
}
private synchronized String getDataFromDB() {
//再次确认缓存中没有数据
String testString = stringRedisTemplatel.opsForValue().get("testString");
// 如果有数据直接返回
if (!StringUtils.isEmpty(testString)) {
System.out.println("返回缓存中的数据！");
return testString;
}
System.out.println("查询数据库");
// 模拟查到的数据结果
String date = "testString";
/**
* 查询到结果后，立即放入缓存中，防止下一个线程重复去数据库查询。
*
* 如果不在这放入缓存，该方法返回后，下一个得到锁的线程，去缓存中查看时，
* 有可能因为其他因素，上一个线程没有来得及将结果放入缓存，因此被当前线程判定缓存中没有数据
* 致使该线程又重新去查一遍数据库。
*
*/
stringRedisTemplatel.opsForValue().set("testString", date);
return date;
}
}
test1 8081端口没有查询数据库
test1 8081端口
test1 8082端口没有查询数据库
test1 8082端口
test1 8083端口查询了数据库
test1 8083端口
从上面结果可以看出，我们已经简单实现了分布式锁，但是这其中存在着很多问题。如：
问题： setnx获取到锁，但此时业务代码异常或者程序在过程中宕机。没有执行删除锁逻辑，就会造成死锁
解决：设置锁的自动过期，即使没有删除，也会自动删除
分布式锁阶段二
分布式锁阶段二
修改代码，添加过期时间
private String getDateFromDbWithRedisLock() throws InterruptedException {
// 1.获取分布式锁
Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "lock");
if (lock) {
//加锁成功
//设置过期时间,30s过期
redisTemplate.expire("lock",30,TimeUnit.SECONDS);
String dataFromDB = getDataFromDB();
//删除锁
redisTemplate.delete("lock");
return dataFromDB;
} else {
//加锁失败，重试(自旋)
TimeUnit.MILLISECONDS.sleep(1000);
return getDateFromDbWithRedisLock();
}
}
此时仍然会存在问题：
问题：setnx设置好，正要去设置过期时间，然后正好宕机，又造成死锁了。
解决：设置过期时间和占位必须是原子操作，redis支持使用setnx ex命令。
分布式锁阶段三
分布式锁阶段三
修改代码，设置锁和过期时间为原子操作
private String getDateFromDbWithRedisLock() throws InterruptedException {
// 1.获取分布式锁,并设置过期时间,30s过期
Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "lock",30,TimeUnit.SECONDS);
if (lock) {
//加锁成功
String dataFromDB = getDataFromDB();
//删除锁
redisTemplate.delete("lock");
return dataFromDB;
} else {
//加锁失败，重试(自旋)
TimeUnit.MILLISECONDS.sleep(1000);
return getDateFromDbWithRedisLock();
}
}
此时仍然会存在问题：
问题：如果业务时间过长，锁自己已经过期了，当完成业务删除锁的时候，有可能是删除别人持有的锁。
解决：获取锁的时候，值指定为一个uuid，当要删除锁的时候，需要匹配为自己持有的锁才删除。
分布式锁阶段四
分布式锁阶段四
修改代码，设置删除逻辑
private String getDateFromDbWithRedisLock() throws InterruptedException {
String token = UUID.randomUUID().toString();
// 1.获取分布式锁,并设置过期时间,30s过期
Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", token,30,TimeUnit.SECONDS);
if (lock) {
//加锁成功
String dataFromDB = getDataFromDB();
//删除锁时，先获取这个锁是否为自己持有
String lockVal = redisTemplate.opsForValue().get("lock").toString();
if(token.equals(lockVal)){
//删除我自己获得的锁
redisTemplate.delete("lock");
}
return dataFromDB;
} else {
//加锁失败，重试(自旋)
TimeUnit.MILLISECONDS.sleep(1000);
return getDateFromDbWithRedisLock();
}
}
问题：由于数据是在网络中传输，如果在锁没有过期前，当前线程去获取值，redis返回我自己保存的token数据后，redis里面的数据正好过期，锁被别人获取了，但是当前线程依然判断锁是自己所获取的，也会造成删除别人获得的锁。
原因：当前删除锁的逻辑分为两步，先获取值，然后判断相同再删。
解决：删除锁也应该设置成原子操作，可以用redis+Lua脚本。
分布式锁阶段五
分布式锁阶段五
修改代码，设置lua脚本删除逻辑
private String getDateFromDbWithRedisLock() throws InterruptedException {
String token = UUID.randomUUID().toString();
// 1.获取分布式锁,并设置过期时间,30s过期
Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", token, 300, TimeUnit.SECONDS);
if (lock) {
//加锁成功,执行业务
String dataFromDB = getDataFromDB();
//获取值对比和对比成功后删除为原子操作
// 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(script, Integer.class), Arrays.asList("lock"), token);
return dataFromDB;
} else {
//加锁失败，重试(自旋)
TimeUnit.MILLISECONDS.sleep(1000);
return getDateFromDbWithRedisLock();
}
}
