【Redis】Redis实现分布式锁

在实现分布式锁之前，首先我们要明白什么是分布式锁？为什么要实现分布式锁？

1. 分布式锁概念

分布式锁：满足分布式系统或集群模式下多进程可见并且互斥的锁。

---

2. 为什么要实现分布式锁

设定一个场景：美团外卖的特价优惠卷一个用户只能购买一张，不能够重复购买。

在系统单机部署的情况下，通过加锁就能够解决就可以解决一人一单的安全问题，但是当系统的用户量越来越多，并发越来越大，单机部署的项目就越来越感到“捉襟见肘”，所以我们需要在多台服务器上集群部署系统，以满足用户的需求。但是当我们集群部署系统之后，问题也随之而来了，之前的锁无法满足集群部署下系统的并发问题，容易出现一人多单的问题。基于此，我们就需要实现 分布式锁 去解决系统集群部署下的并发问题。

---

2.1 并发安全问题

首先明白一点，我们单机部署下系统是通过 synchronized (userId.toString().intern()) 对用户id进行加锁。因为在单个jvm中可以确保每次从常量池中得到的用户id都是同一个。当一个线程拿到锁之后，其他线程就无法重复拿到锁了。这样就实现了一人一单。

当系统集群部署之后，每个单独部署的系统都有了各自的jvm，如果还是像之前通过 synchronized (userId.toString().intern()) 对用户id加锁，那么只能做到在各台单独部署的系统上实现并发安全，但是在整个集群上是并发不安全的。jvm1的第一个线程拿到了锁，jvm2的第一个线程也能拿到锁，因为它们不是拿的不是同一个线程池上的用户id。这样就会出现一人多单的问题。

所以我们不能使用之前的方案了，我们需要一个外部共享的工具来替代原来的锁监视器。如下图所示：

由此，就引出来分布式锁。

---

3. 分布式锁的实现方案

分布式锁的核心是实现多进程之间互斥，满足这一点的方式有很多，常见的方案有以下三种：

• MYSQL
• Redis
• Zookeeper

	MSQL	Redis	Zookeeper
互斥	利用mysql本身的互斥锁机制	利用setnx这样的互斥命令	利用节点的唯一性和有序性实现互斥
高可用	好	好	好
高性能	一般	好	一般
安全性	断开连接，自动释放锁	利用锁超时时间，到期释放	临时节点，断开连接自动释放

综合考虑，我们使用Redis实现分布式锁。

---

3.1 Redis实现分布式锁

3.1.1 定义分布锁接口和类

首先我们定义分布式锁的接口：

public interface ILock {

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

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

然后定义一个类来实现这个接口：

public class SimpleRedisLock implements ILock {

    private StringRedisTemplate stringRedisTemplate;
    private String name;

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

    private static final String KEY_PREFIX = "lock:";
    private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";
    private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;

    static {
        UNLOCK_SCRIPT = new DefaultRedisScript<>();
        UNLOCK_SCRIPT.setResultType(Long.class);
        UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
    }

    //获取锁
    @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);
        return BooleanUtil.isTrue(success);
    }
    
    //版本一 释放锁
    //@Override
    //public void unlock() {
    //    //获取缓存线程标识
    //    String cacheThreadId = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
    //    //获取当前线程标识
    //    String threadId = ID_PREFIX + Thread.currentThread().getId();
    //    if (threadId.equals(cacheThreadId)) {
    //        //判断线程标示是否一致，防止误删
    //        stringRedisTemplate.delete(KEY_PREFIX + name);
    //    }
    //}

    //版本二 释放锁
    public void unlock() {
        //调用lua脚本
        stringRedisTemplate.execute(
                UNLOCK_SCRIPT,
                Collections.singletonList(KEY_PREFIX + name),
                ID_PREFIX + Thread.currentThread().getId()
        );
    }
}

解释：

• 虽然线程id在每个jvm内都是唯一的，但是不保证在其它jvm中不会出现相同的线程id。所以我们在线程id前面拼接一个uuid，避免线程标识相同而误删锁。
• 释放锁的过程中，为防止误删锁需要保证校验线程标识是否相同操作和释放锁操作的原子性，这里我们使用lua脚本。

3.1.2 编写lua脚本

在lua脚本中我们要实现两个功能，一个是判断线程标识是否一致，一个是释放锁。

-- 比较线程标识与锁中的标识是否一致
if (redis.call('get', KEYS[1]) == ARGV[1]) then
    -- 释放锁 del key
    return redis.call('del', KEYS[1])
end
return 0

---

3.1.3 使用线程锁

@Override
public Result seckillVoucher(Long voucherId) {
    //1.查询优惠券id
    SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
    //2.判断秒杀是否开始
    LocalDateTime beginTime = voucher.getBeginTime();
    LocalDateTime endTime = voucher.getEndTime();
    if (LocalDateTime.now().isBefore(beginTime)) {
        //2.1.秒杀未开始
        return Result.fail("秒杀未开始");
    }
    if (LocalDateTime.now().isAfter(endTime)) {
        //2.2.秒杀已结束
        return Result.fail("秒杀已结束");
    }
    //2.3.秒杀正在进行
    int stock = voucher.getStock();
    //3.判断库存是否充足
    if (stock <= 0) {
        //3.1库存不足
        return Result.fail("库存不足");
    }
    //3.2.库存充足

    Long userId = UserHolder.getUser().getId();
    //确保每个userId的锁是同一把
    SimpleRedisLock lock = new SimpleRedisLock(stringRedisTemplate, "order:" + userId);
    if (!lock.tryLock(1200)) {
        //获取锁失败
        return Result.fail("不允许重复下单");
    }
    try {
        //获取代理对象（事务）
        IVoucherOrderService proxy = (IVoucherOrderService) AopContext.currentProxy();
        return proxy.createVoucherOrder(voucherId);
    } finally {
        //释放锁
        lock.unlock();
    }
}

@Transactional
public Result createVoucherOrder(Long voucherId) {
    //4.一人一单
    Long userId = UserHolder.getUser().getId();

    //4.1.查询是否已经有订单
    int count = lambdaQuery().eq(VoucherOrder::getUserId, userId).eq(VoucherOrder::getVoucherId, voucherId).count();
    if (count > 0) {
        //已有订单
        return Result.fail("不能重复购买");
    }

    //5.扣减库存
    //5.1.写法一
    LambdaUpdateWrapper<SeckillVoucher> updateWrapper = new LambdaUpdateWrapper<>();
    updateWrapper.setSql("stock=stock-1").eq(SeckillVoucher::getVoucherId, voucherId).gt(SeckillVoucher::getStock, 0);
    boolean success = seckillVoucherService.update(updateWrapper);
    //5.2.写法二
    //boolean success = seckillVoucherService.update()
    //        .setSql("stock = stock - 1")
    //        .eq("voucher_id", voucherId).gt("stock", 0).update();

    if (!success) {
        //扣减失败
        return Result.fail("库存不足！");
    }

    //6.创建订单
    VoucherOrder voucherOrder = new VoucherOrder();
    //6.1.订单id
    long orderId = redisIdWorker.nextId("order");
    voucherOrder.setId(orderId);
    //6.2.用户id
    voucherOrder.setUserId(UserHolder.getUser().getId());
    //6.3.代金券id
    voucherOrder.setVoucherId(voucherId);
    //6.4.写入数据库
    save(voucherOrder);

    //7.返回订单id
    return Result.ok(voucherOrder.getId());
}

解释

在方法中调用另一个加了事务注解的方法，会导致事务注解失效。我们需要让代理对象来调用加了事务注解的方法，这样就不会失效了。

---

3.1.4 总结

至此，Redis实现的分布式锁已经完成了，但是还有优化的余地。我们也可以使用成熟的分布式锁框架 Redisson 。