分布式场景下并发安全问题的引发
前面通过加锁解决了单机状态下一人一单的问题，但是当出现了分布式，前面的加锁形式不再适用 ，每个 jvm 有一个自己的锁监视器，只能被内部线程获取，其他 jvm 无法使用，那么多台 jvm 的锁监视器不共用一个锁监视器，就容易出现分布式场景下并发安全问题。
问题分析
所以我们要使用可以解决分布式场景下的位于 jvm 外的锁，多个 jvm 共同使用该锁，而不是使用每个 jvm 的内部锁。
分布式锁有如下特点：
这里我们就选用 redis 来实现我们的分布式锁！
Redis 锁的 demo
redis 锁要实现如上两个基本操作：获取锁和删除锁，在获取锁的同时为了防止宕机出现死锁，要手动添加过期时间，那么为了防止只加锁没有加过期时间的情况出现，我们要保证加锁和加过期时间的原子性，也就是他俩必须同时进行！
那么上述加锁的命令可以换成如下：
分布式锁初步实现
实现锁接口
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
|
public class SimpleRedisLock implements ILock {
//用户的userid
private String name;
private StringRedisTemplate stringRedisTemplate;
public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {
this.name = name;
this.stringRedisTemplate = stringRedisTemplate;
}
//锁的key值
private static final String KEY_PREFIX = "lock:";
//生成锁的value值
private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";
@Override
public boolean tryLock(long timeoutSec) {
// 获取线程标示
String threadId = ID_PREFIX + Thread.currentThread().getId();
// 获取锁
Boolean success = stringRedisTemplate.opsForValue()
.setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);
//防止Boolean和boolean拆箱出问题，如果success为null，则返回false
return Boolean.TRUE.equals(success);
}
@Override
public void unlock() {
//释放锁
stringRedisTemplate.delete(KEY_PREFIX + name);
}
}
|
使用锁
在VoucherOrderServiceImpl.java 中的seckillVoucher 方法中编写如下代码：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
//获取用户userid
Long id = UserHolder.getUser().getId();
SimpleRedisLock redisLock = new SimpleRedisLock("order:" + id, stringRedisTemplate);
//加锁，1200s是锁的过期时间
boolean tryLock = redisLock.tryLock(1200);
//判断锁是否获取成功
if (!tryLock){
return Result.fail("不允许重复下单");
}
try {
//锁加到这里，事务提交后才释放锁
//获取事务的动态代理对象，需要在启动类加注解暴漏出对象
IVoucherOrderService proxy = (IVoucherOrderService)AopContext.currentProxy();//拿到动态代理对象
// return createVoucherOrder(voucherId, voucher);
return proxy.createVoucherOrder(voucherId, voucher);
//使用动态代理类的对象，事务可以生效
} finally {
//无论如何都要释放锁，防止死锁
redisLock.unlock();
}
|
分布式场景下调试看效果
两个 application，一个是 8081 端口，一个是 8082 端口。
apifox 模拟同一个用户发送请求，authorization 的参数值是同一个用户的，存储在 redis 中。
如下：在 8082 的断点处获取锁失败，在 8081 的断点处获取锁成功，即只有一次成功获取锁。
数据库中优惠券库存 stock-1 而不是-2，优惠券订单产生 1 个，数据库没问题！
redis 中查看锁的 key 值，1010 正是 userid，问题解决，达到我们想要的效果！
Redis 分布式锁误删问题
问题分析
当线程 1 获取锁成功时，如果该业务执行时间长以至于超过了设置的锁过期时间，那么在业务还未完成时，锁便自动释放，此时线程 1 无锁，线程 2 获取到了锁执行业务，当线程 1 业务执行完后，按照业务逻辑仍会释放锁，但此时释放的是线程 2 的锁，这就出现了锁误删的问题。
解决锁误删
对于每个线程，我们获取其线程标识（每个 JVM 内部都维护了线程的 id，这个 id 是自增的，那么多个 jvm 可能出现线程 id 一致的情况，为了避免该情况出现我们用 UUID 生成一个随机字符串作为前缀，以降低线程 id 重复的概率）作为锁的 value
在释放锁时，我们先从 redis 获取对应的 value 值，跟当前线程的 value 做对比，一致则可以删除，否则就不能删除。
修改 trylock 和 unlock 方法
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
|
public class SimpleRedisLock implements ILock {
//用户的userid
private String name;
private StringRedisTemplate stringRedisTemplate;
public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {
this.name = name;
this.stringRedisTemplate = stringRedisTemplate;
}
//锁的key值
private static final String KEY_PREFIX = "lock:";
//线程的前缀，因为分布式下，多个jvm，每个jvm中维护的线程的id都是递增的，那么可能出现多个jvm的线程id一致，所以这里用uuid生成字符串作为前缀
private static final String THREAD_PREFIX = UUID.randomUUID().toString(true)+"-";
@Override
public boolean tryLock(long timeoutSec) {
// 获取线程标示 线程前缀+线程id
String threadId = THREAD_PREFIX + Thread.currentThread().getId();
// 获取锁
Boolean success = stringRedisTemplate.opsForValue()
.setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);
//防止Boolean和boolean拆箱出问题，如果success为null，则返回false
return Boolean.TRUE.equals(success);
}
@Override
public void unlock() {
// 获取线程标示
String threadId =THREAD_PREFIX + Thread.currentThread().getId();
// 获取锁中的标示
String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
// 判断锁标示是否一致，防止锁误删
if(threadId.equals(id)) {
// 释放锁
stringRedisTemplate.delete(KEY_PREFIX + name);
}
}
}
|
分布式锁的原子性
问题分析
JVM 做 Full GC 时会阻塞所有代码，时间过长会出现锁超时自动释放，那么其他线程会趁虚而入获得锁。
那么会出现如下情况：线程 1 获取锁执行业务逻辑后要释放锁，在判断完释放锁的条件为 true 后，即 threadId.equals(id)==true， 正要释放锁时出现 Full GC，所有代码被阻塞，直到锁超时自动释放 （注意此时锁不是正常释放而是锁超时释放的），就在这时 GC 完毕代码恢复，线程 2 趁虚而入获得锁，而线程 1 也恢复了要执行释放锁的代码，**因为 GC 前已经判断过释放条件为 ture，那么此时线程 1 仍然认为锁是自己的，会错误地释放线程 2 的锁，又出现了误删问题。这里我们就要保证****锁的原子性，即 判断锁的标识 和 释放锁 两个动作必须同时发生！**
问题解决（Lua 脚本）
Lua 是一种编程语言，Redis 提供了 Lua 脚本功能，即可以在一个脚本中写多条 redis 指令，确保了多条命令执行的原子性
> 
> 
代码实现
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
|
public class SimpleRedisLock implements ILock {
//用户的userid
private String name;
private StringRedisTemplate stringRedisTemplate;
public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {
this.name = name;
this.stringRedisTemplate = stringRedisTemplate;
}
//锁的key值
private static final String KEY_PREFIX = "lock:";
//线程的前缀，因为分布式下，多个jvm，每个jvm中维护的线程的id都是递增的，那么可能出现多个jvm的线程id一致，所以这里用uuid生成字符串作为前缀
private static final String THREAD_PREFIX = UUID.randomUUID().toString(true)+"-";
private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;
static {
UNLOCK_SCRIPT = new DefaultRedisScript<>();
UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
UNLOCK_SCRIPT.setResultType(Long.class);
}
@Override
public boolean tryLock(long timeoutSec) {
// 获取线程标示 线程前缀+线程id
String threadId = THREAD_PREFIX + Thread.currentThread().getId();
// 获取锁
Boolean success = stringRedisTemplate.opsForValue()
.setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);
//防止Boolean和boolean拆箱出问题，如果success为null，则返回false
return Boolean.TRUE.equals(success);
}
@Override
public void unlock() {
// 调用lua脚本
stringRedisTemplate.execute(
UNLOCK_SCRIPT,
Collections.singletonList(KEY_PREFIX + name),
THREAD_PREFIX + Thread.currentThread().getId());
}
}
|
在 resource 资源文件夹下创建 Lua 脚本内容如下：
1
2
3
4
|
if (redis.call('get',KEYS[1])==ARGV[1]) then
return redis.call('del',KEYS[1])
end
return 0
|