2023.12.10

集群模式下的并发安全问题及解决

        随着现在分布式系统越来越普及，一个应用往往会部署在多台机器上（多节点），通过加锁可以解决在单机情况下的一人一单安全问题，但是在集群模式下就不行了。见下图：

        多台服务器会对应多个jvm， synchronized锁可以锁住单台服务器的多线程，多台服务器就锁不住了，所以我们需要有一个多服务器共享的锁监视器，这里就需要使用到分布式锁了，这里我们使用redis的SETNX这个方法来实现。  流程图如下：

        首先定义一个锁的接口，并实现它：

public interface ILock {

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

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

public class SimpleRedisLock implements ILock{
    private String name;
    private StringRedisTemplate stringRedisTemplate;

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

    private static final String KEY_PREFIX = "lock:";
    @Override
    public boolean tryLock(long timeoutSec) {
        //获取线程标识
        long threadId = Thread.currentThread().getId();
        //获取锁
        Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name,threadId + "",timeoutSec, TimeUnit.SECONDS);
        return Boolean.TRUE.equals(success);//防止拆箱的时候出现空指针异常
    }

    @Override
    public void unlock() {
        //释放锁
        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();
        }

        再使用jmeter+多台服务器进行测试，集群模式下的并发安全问题得到解决。

redis分布式锁误删问题及解决

        考虑一种情况：假设线程1获取了分布式锁，然后业务阻塞了，阻塞的时间超过了redis中锁的超时时间，redis将锁释放了。这时线程2就顺利获取了该锁，并执行它的业务。此时线程1苏醒了并执行完自己的业务，于是释放锁，此时释放的锁是线程2刚刚获取的锁，意味着此时其他线程也可以获取锁进来了，这就又出现了并发安全问题了。 核心原因就在于：线程1在释放锁之前没有判断一下这把锁是不是自己之前获取的锁，导致误删了其他线程的锁。

        解决办法就是：在获取锁的时候存入线程标识（用UUID标识，在一个JVM中，ThreadId一般不会重复，但是我们现在是集群模式，有多个JVM，多个JVM之间可能会出现ThreadId重复的情况），在释放锁的时候先获取锁的线程标识，判断是否与当前线程标识一致：如果一致则允许释放。

        流程图改为：

        需要修改SimpleRedisLock.java代码：

public class SimpleRedisLock implements ILock{
    private String name;
    private StringRedisTemplate stringRedisTemplate;

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

    private static final String KEY_PREFIX = "lock:";
    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);
        return Boolean.TRUE.equals(success);//防止拆箱的时候出现空指针异常
    }

    @Override
    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);
        }
    }
}

分布式锁的原子性问题及解决

        然而上述解决方案在极端情况下还有问题：当线程1在判断完锁的标示之后，准备释放锁之前如果出现阻塞的话(由于jvm的垃圾回收机制等原因)，redis的超时时间到了，将锁释放掉，其他线程又可以获取锁了，则又会出现和上述一样的情况：线程1会将其他线程的锁误释放掉。 产生此问题的核心原因就在于：判断锁标示和释放锁这两个操作不具有原子性。 导致在这期间又有可能出现并发安全问题。

        这里我们使用Lua脚本解决多条命令原子性问题。Redis提供了Lua脚本功能，在一个脚本中编写多条Redis命令，确保多条命令执行时的原子性。

        编写lua脚本：

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

        调用lua脚本：

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

        这样判断和释放操作就具有原子性了。