简单理解下基于 Redisson 库的分布式锁机制

2024-11-23 22:15:07

简单理解下基于 Redisson 库的分布式锁机制

这段代码实现了一个基于 Redisson 库的分布式锁机制

Redisson 是一个基于 Redis 的 Java 客户端库，它提供了丰富的分布式数据结构和工具，旨在使 Java 开发者能够轻松地在分布式环境下使用 Redis。

Redisson 是 Redis 的高级封装，除了常规的 Redis 命令支持外，它还提供了很多额外的功能，如分布式锁、分布式集合、分布式队列、分布式缓存等，使得开发者在分布式系统中实现高效、可靠的操作。

比方对一个用户所拥有的金额进行增减操作，肯定需要上锁才能保证一定的安全性。

代码流程：

方法的调用：

一个简单的流程：

LockUtil.riderBalance(riderId, () -> {…}) 是一个加锁操作，是一种使用分布式锁来确保线程安全的机制，确保在修改余额时，避免并发冲突。

简要来说，它的作用是在执行余额操作时，确保只有一个线程能够对指定的骑手账户进行操作，避免多个线程同时修改余额，导致数据不一致的情况。

LockUtil.riderBalance(riderId, …) 通过 riderId 获取一个锁，这样就能确保在同一时刻，只有一个线程可以执行传入的操作逻辑。

riderId 是骑手的唯一标识，通过它来加锁，防止对同一个骑手账户的并发操作。

() -> {…}： 这是一个 Lambda 表达式，表示当获得锁之后要执行的具体操作。这个操作包括：获取骑手的余额信息、执行余额扣除、记录操作日志等步骤。

    /**
     * 扣除余额
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean deductionBalance(Integer riderId, BigDecimal amount, String title, Integer orderId, String orderCode, RiderCashlogType cashlogType, RiderCashlogStatus cashlogStatus) {
        return LockUtil.riderBalance(riderId, () -> {
            // 变更前
            RiderServiceDto riderBefore = this.getRiderById(riderId);

            // 减少余额 -------- ★★★★★★★★★★★★★★★★★★★★★★★★
            boolean flag = xxRiderService.deductionBalance(riderId, amount);

            // 变更后
            RiderServiceDto riderAfter = this.getRiderById(riderId);

            // 增加流水日志
            if (flag) {
                RiderCashlogServiceDto cashlog = new RiderCashlogServiceDto();
                cashlog.setMoneyType(RiderCashlogMoneyType.BALANCE);
                cashlog.setTitle(title);
                ......
				
                return xxRiderCashlogService.addCashlog(cashlog);
            }
            return false;
        }, () -> {
            throw BizException.newInstance(ErrorCode.BUSY);
        });
    }

调用接口方法

    /**
     * 扣除余额
     */
    boolean deductionBalance(Integer riderId, BigDecimal amount);

这里再进行详细的扣除方法，依然是用同个锁

    /**
     * 扣除余额
     */
    @Override
    public boolean deductionBalance(Integer riderId, BigDecimal amount) {
        if (amount.compareTo(BigDecimal.ZERO) == 0) {
            return true;
        }
        // 加锁
        return LockUtil.riderBalance(riderId, () -> {
            if (ObjectUtil.isEmpty(riderId) || amount.compareTo(BigDecimal.ZERO) < 0) {
                log.error("扣除xx余额失败,riderId:{},amount:{}", riderId, amount);
                return false;
            }
            RiderServiceDto rider = this.getRiderById(riderId);
            // 获取当前账号的余额
            BigDecimal valid = rider.getValidMoney();
            // 判断当前余额是否足够
            if (valid.compareTo(amount) < 0) {
                throw BizException.newInstance(ErrorCode.ARGUMENT_ERROR, "扣除余额失败,余额小于" + amount.toPlainString());
            }

            return this.update(new UpdateWrapper<RiderEntity>().lambda()
                    .eq(RiderEntity::getId, riderId)
                    .setSql("valid_money = valid_money - " + amount));
        }, () -> {
            throw BizException.newInstance(ErrorCode.BUSY);
        });

    }

具体锁的实现：

riderBalance 方法：

    /**
     * 锁骑手余额------
     * Integer riderId: 骑手的 ID，作为分布式锁的标识。
     * Supplier<T> supplier: 提供数据或执行操作的函数接口。
     *                       传入的 Supplier<T> 函数接口，用来提供需要执行的业务操作。如果获取锁成功，业务操作会在锁内部执行。
     * InvokeInter lockFail: 如果获取锁失败，执行的操作。
     */
    public static <T> T riderBalance(Integer riderId, Supplier<T> supplier, InvokeInter lockFail) {
        return tryLock(RedisKey.LOCK_RIDER_BALANCE_XXXXX.getKey(riderId), supplier, lockFail);
    }

tryLock 方法（重载）：

    /**
     * 设置分布式锁
     */
    public static <T> T tryLock(String key, Supplier<T> supplier, InvokeInter lockFail) {
    	//这是 tryLock 的一个重载方法，它会在尝试获取锁时默认设置超时时间为 10 秒
    	//通过 key、supplier 和 lockFail，它会调用 tryLock 的另一个重载版本，并使用 10 秒的默认超时时间
        return tryLock(key, supplier, lockFail, 10, TimeUnit.SECONDS);
    }

tryLock 方法（核心实现）：


    /**
     * 设置分布式锁-----------
     * key: 锁的标识符，通常是与某个资源（如骑手余额）相关的唯一标识。
	 * supplier: 需要执行的业务逻辑。
	 * lockFail: 获取锁失败时的回调操作。
	 * amount 和 unit: 锁的持有时间（amount）和时间单位（unit），用于控制获取锁的最长等待时间。
     */
    public static <T> T tryLock(String key, Supplier<T> supplier, InvokeInter lockFail, long amount, TimeUnit unit) {
    	//通过 Redisson 客户端获取一个 RLock 对象，RLock 是 Redisson 提供的分布式锁实现。
        RLock lock = _this.redissonClient.getLock(key);
        try {
        	// lock.isLocked()：判断锁是否已经被其他线程持有。
        	//lock.isHeldByCurrentThread()：判断当前线程是否已经持有该锁。如果当前线程已经持有锁，它不需要重复获取。
            if (lock.isLocked() && !lock.isHeldByCurrentThread()) {
                if (lockFail == null) {
                	// 如果 lockFail 为空，则抛出 BizException 异常，表示当前资源忙（例如余额操作正在进行）
                    throw BizException.newInstance(ErrorCode.BUSY);
                }
                // 如果锁已经被其他线程持有且不是当前线程持有，且 lockFail 不为空，就执行 lockFail.invoke() 来处理锁获取失败的情况。
                lockFail.invoke();
                //lock.tryLock(amount, unit)：尝试在指定的时间内（amount 秒）获取锁。如果成功，执行后续的业务操作。
            }  else if (lock.tryLock(amount, unit)) {
                try {
                	// 如果锁获取成功，调用 supplier.get() 执行传入的业务逻辑。
                    return supplier.get();
                } finally {
                	//无论业务逻辑执行成功与否，都要确保释放锁，lock.unlock() 用于释放锁，防止死锁。
                    lock.unlock();
                }
            } else {
            	//如果在 tryLock 的超时时间内无法获得锁（即返回 false），就执行 lockFail.invoke()，或者抛出 BizException 异常表示资源繁忙
                if (lockFail == null) {
                    throw BizException.newInstance(ErrorCode.BUSY);
                }
                lockFail.invoke();
            }
            //最终返回 null，因为所有锁的操作是围绕着锁获取和释放展开的，所有的业务执行通过 supplier.get() 来完成，返回值由 supplier 提供
            return null;
        } catch (InterruptedException e) {
        	//如果线程在等待锁的过程中被中断，则抛出 BizException 异常，表示当前资源忙（此时无法完成操作）。
            throw BizException.newInstance(ErrorCode.BUSY);
        }
    }

这段代码主要实现了一个分布式锁的功能，通过 Redisson 提供的 RLock 来确保在分布式环境下对资源的访问是互斥的，避免多个线程或进程同时访问同一资源（如xxx余额）。

tryLock 方法用于尝试获取锁，并在成功获取锁时执行指定的业务操作（supplier.get()）。如果获取锁失败，依据 lockFail 参数的值执行相应的失败处理操作。

锁的获取有超时机制，超过超时时间仍无法获取锁时会触发失败处理

码农公寓

简单理解下基于 Redisson 库的分布式锁机制

目录