基于 Redis 实现分布式锁，分析解决锁误删情况 及 利用Lua脚本解决原子性问题并改造锁

Posted 2022-09-20 Perceus

(目录)

---

上一篇博文部分：

1. 优惠卷秒杀

---

分布式锁

1 、基本原理和实现方式对比

分布式锁：

分布式锁的核心思想就是：

---

分布式锁他应该满足一些什么样的条件呢？

---

常见的分布式锁有三种：

---

2 、Redis分布式锁的实现核心思路

实现分布式锁时需要实现的两个基本方法：

---

核心思路：

---

3、实现分布式锁版本一

锁的基本接口

public interface ILock 

    /**
     *  尝试获取锁
     * @param timeoutSec 锁持有的超时时间 ， 过期自动释放
     * @return true -> 获取锁成功 , false -> 获取锁失败
     */
    boolean tryLock(long timeoutSec);

    /**
     *  释放锁
     */
    void unlock();

SimpleRedisLock类

    private static final String KEY_PREFIX="lock:";
    private String name;
    private StringRedisTemplate stringRedisTemplate;

    public SimpleRedisLock(StringRedisTemplate stringRedisTemplate, String name) 
        this.stringRedisTemplate = stringRedisTemplate;
        this.name = name;
    


    @Override
    public boolean tryLock(long timeoutSec) 

        // 获取线程标示
        long threadId = Thread.currentThread().getId();
        // 获取锁
        Boolean success = stringRedisTemplate.opsForValue()
                .setIfAbsent(KEY_PREFIX + name, threadId+"" ,timeoutSec, TimeUnit.SECONDS);
        // 是否获取锁成功 不要直接返回success 自动拆箱会有空指针的可能
        return Boolean.TRUE.equals(success);
    

---

SimpleRedisLock

释放锁，防止删除别人的锁

public void unlock() 
    //通过del删除锁
    stringRedisTemplate.delete(KEY_PREFIX + name);

---

  @Override
    public Result seckillVoucher(Long voucherId) 
        // 1.查询优惠券
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        // 2.判断秒杀是否开始
        if (voucher.getBeginTime().isAfter(LocalDateTime.now())) 
            // 尚未开始
            return Result.fail("秒杀尚未开始！");
        
        // 3.判断秒杀是否已经结束
        if (voucher.getEndTime().isBefore(LocalDateTime.now())) 
            // 尚未开始
            return Result.fail("秒杀已经结束！");
        
        // 4.判断库存是否充足
        if (voucher.getStock() < 1) 
            // 库存不足
            return Result.fail("库存不足！");
        
        Long userId = UserHolder.getUser().getId();
        //创建锁对象(新增代码)
        SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
        //获取锁对象
        boolean isLock = lock.tryLock(1200);
		//加锁失败
        if (!isLock) 
            return Result.fail("不允许重复下单");
        
        try 
            //获取代理对象(事务)
            IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
            return proxy.createVoucherOrder(voucherId);
         finally 
            //释放锁
            lock.unlock();
        
    

效果：在分布式下只有一个线程拿到了锁

---

4、Redis分布式锁误删情况说明

逻辑说明：

解决方案：

---

5、 解决Redis分布式锁误删问题

需求：

---

核心逻辑：

---

具体代码如下：

    private static final String KEY_PREFIX="lock:";
    private String name;
	private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";
	private StringRedisTemplate stringRedisTemplate;

    @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);
        // 是否获取锁成功 不要直接返回success 自动拆箱会有空指针的可能
        return Boolean.TRUE.equals(success);
    

public void unlock() 
    // 获取线程标示
    String threadId = ID_PREFIX + Thread.currentThread().getId();
    // 获取锁中的标示
    String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
    // 判断标示是否一致
    if(threadId.equals(id)) 
        // 释放锁
        stringRedisTemplate.delete(KEY_PREFIX + name);
    

代码实操说明：

---

6、 分布式锁的原子性问题

更为极端的误删逻辑说明：

---

7、 Lua脚本解决多条命令原子性问题

Redis提供了Lua脚本功能，在一个脚本中编写多条Redis命令，确保多条命令执行时的原子性。

这里重点介绍Redis提供的调用函数，语法如下：

redis.call(命令名称, key, 其它参数, ...)

例如，我们要执行set name jack，则脚本是这样：

# 执行 set name jack
redis.call(set, name, jack)

例如，我们要先执行set name Rose，再执行get name，则脚本如下：

# 先执行 set name jack
redis.call(set, name, Rose)
# 再执行 get name
local name = redis.call(get, name)
# 返回
return name

写好脚本以后，需要用Redis命令来调用脚本，调用脚本的常见命令如下：

---

例如，我们要执行

redis.call(set, name, jack)

这个脚本，语法如下：

EVAL "redis.call(set, name, jack)" 0

如果脚本中的key、value不想写死，可以作为参数传递。key类型参数会放入KEYS数组，其它参数会放入ARGV数组，在脚本中可以从KEYS和ARGV数组获取这些参数：

接下来我们来回一下我们释放锁的逻辑：

释放锁的业务流程是这样的

如果用Lua脚本来表示则是这样的：

-- 这里的 KEYS[1] 就是锁的key，这里的ARGV[1] 就是当前线程标示
-- 获取锁中的标示，判断是否与当前线程标示一致
if (redis.call(GET, KEYS[1]) == ARGV[1]) then
  -- 一致，则删除锁
  return redis.call(DEL, KEYS[1])
end
-- 不一致，则直接返回
return 0

---

8、 利用Java代码调用Lua脚本改造分布式锁

---

定义一个lua脚本

Java代码

    // 提前读取脚本 静态代码块
    static 
        UNLOCK_SCRIPT = new DefaultRedisScript<>();
        UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua")); // 读取脚本
        UNLOCK_SCRIPT.setResultType(Long.class); // 返回值
    

    public void unlock() 
        // 调用lua脚本
        stringRedisTemplate.execute(
                UNLOCK_SCRIPT,
                Collections.singletonList(KEY_PREFIX + name),
                ID_PREFIX + Thread.currentThread().getId());
    

---

小总结：

基于Redis的分布式锁实现思路：

特性：

---

总结：

但是目前还剩下一个问题锁不住，什么是锁不住呢？

测试逻辑：

---