背景

随着业务的发展，IT项目逐渐演进为微服务架构，这也带来了一些挑战，例如在锁的使用方面。在传统的单体应用中，锁通常在整个应用程序中共享，然而在微服务架构中，每个服务都有独立的数据库和缓存，这意味着锁需要在服务之间进行协调。

以下通过两张图来阐明本地锁和分布式锁的区别：

image

image

为了应对这一问题，分布式锁应运而生，而其中最常采用的技术之一是基于 Redis 实现的分布式锁。

基本实现思路

通过 Redis 的 SET NX 命令，我们可以实现一种原子操作：只有在指定的 key 不存在时，写入才会成功；若 key 已存在，则写入会失败。

public synchronized boolean tryLock() {
    if (this.locking) {
        log.warn("【Redis锁异常】key=[{}] 重复请求锁：不支持重入，请检查代码", this.key);
        return false;
    }


    //尝试拿锁，如果拿不到就等待并重试，最多等待this.maxWaitSeconds
    boolean success = tryWaitForLock();

    if (success) {
        //已获取到锁
        this.locking = true;
        //注册到manager，以进行续期管理
        manager.registerLock(this);
        lastRenewalTime = LocalDateTime.now();
        return true;
    } else {
        //未获取到锁
        log.warn("【Redis锁获取失败】key=[{}] 取锁失败，且等待时间超出最多等待[{}]秒 value=[{}]", this.key, this.maxWaitSeconds, this.value);
        return false;
    }

}

public boolean tryWaitForLock(String lockKey, String lockValue, long expireTime, long maxWaitSeconds) {
        final int sleepMills = 100;
        final long maxWaitMills = maxWaitSeconds * 1000;
        final long maxLoop = maxWaitMills / sleepMills;
        final Random random = new Random();
        for (int idx = 0; idx <= maxLoop; idx++) {
            try {
                String result = redisCache.set(lockKey, lockValue, NX, EX, expireTime);
                if (LOCK_SUCCESS.equalsIgnoreCase(result)) {
                    return true;
                }
                if (idx < maxLoop) {
                    // 20ms上下浮动，避免波峰
                    Thread.sleep(sleepMills + (20 - random.nextInt(40)));
                    log.debug("【等待Redis锁】key=[{}] 已等待[{}]毫秒 maxWaitMills=[{}]毫秒 value=[{}]", lockKey,
                            (idx + 1) * sleepMills, maxWaitMills, lockValue);
                }
            } catch (Exception e) {
                log.debug("【等待Redis锁】key=[{}] 等待时出现异常 已等待[{}]毫秒 maxWaitMills=[{}]毫秒  value=[{}]", lockKey,
                        (idx + 1) * sleepMills, maxWaitMills, lockValue, e);
            }
        }
        return false;
    }

在获取锁的过程中，如果第一次尝试失败，会进行多次尝试，若依然无法获取锁，则返回失败。

然而，我们仍需处理一种情况：业务执行时间较长，但锁已过期。为应对这种情况，客户端可以在成功设置锁后，启动定时任务，在锁即将超时之前更新锁的超时时间，以确保业务完成的同时保持锁的有效性。

private RedisLockRenewalManager() {
    ScheduledExecutorService executorService = Executors.newScheduledThreadPool(
            3,
            runnable -> new Thread(runnable, "redis-lock-renewal")
    );
    executorService.scheduleAtFixedRate(this::heartbeat, 500, 500, TimeUnit.MILLISECONDS);
}

private void heartbeat() {
    for (RedisLock lock : locks) {
        if (lock.isLocking()) {
            //正在锁定中的，检查是否需要续期，需要的自动执行续期
            lock.renewal();
        } else {
            //移除未使用的锁，避免内存泄露
            locks.remove(lock);
        }
    }
}

public void renewal() {
    LocalDateTime nextRenewalTime = lastRenewalTime.plusSeconds(this.expireSeconds / 2);
    if (nextRenewalTime.isAfter(LocalDateTime.now())) {
        log.trace("【redis锁续期】key=[{}] 过期时间[{}s]未过半，暂不需要刷新，本次跳过", this.key, this.expireSeconds);
        return;
    }
    Object result = redisCache.eval(RENEWAL_LUA_SCRIPT, 1, this.key, this.value, String.valueOf(this.expireSeconds));
    if (Objects.nonNull(result) && Objects.equals(result, 1L)) {
        this.lastRenewalTime = LocalDateTime.now();
        log.debug("【redis锁续期成功】key=[{}]", this.key);
    } else {
        String redisValue = redisCache.get(this.key);
        log.warn("【redis锁续期失败】key=[{}] this.value=[{}] redis.value=[{}]", this.key, this.value, redisValue);
    }
}

于是加锁的整个过程如图：

image

加锁环节几个问题解决了，锁释放应如何实现呢？
可以使用redis的del命令对锁进行释放，这里释放的时候需要判断当前的锁对象是不是自己的，避免误释放了。因此也采用Redis脚本命令的方式：

/**
 * 解锁lua脚本
 */
public static final String UNLOCK_LUA_SCRIPT = "if (redis.call('GET', KEYS[1]) == ARGV[1]) then return redis.call('DEL', KEYS[1]) else return 0 end";

/**
 * 续期lua脚本
 */
public static final String RENEWAL_LUA_SCRIPT = "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('EXPIRE', KEYS[1], ARGV[2]) else return 0 end";

public boolean unlock() {
    if (!this.locking) {
        log.warn("【Redis锁异常】key=[{}] 重复解锁，请检查代码", this.key);
        return false;
    }
    try {
        Object result = redisCache.eval(UNLOCK_LUA_SCRIPT, 1, this.key, this.value);
        if (Objects.isNull(result) || !Objects.equals(1L, result)) {
            String redisValue = redisCache.get(this.key);
            log.warn("【释放redis锁失败】key=[{}] this.value=[{}] redis.value=[{}]", this.key, this.value, redisValue);
        }
    }catch (Exception exception){
        log.warn("【释放Redis锁异常】key=[{}],msg=[{}]",this.key,exception.getMessage(),exception);
    }finally {
        // 无论释放锁实际是否成功，均返回成功。
        // 如果释放锁失败，则由redis自动过期清除该锁，需要自动禁止续期
        this.locking = false;
        manager.unregisterLock(this);
    }
    return true;
}

这里对key的定义是这样的：

private String generateValue() {
    Thread thread = Thread.currentThread();
    String hostname = System.getProperty("HOSTNAME");
    return UUID.randomUUID() + "#34; + thread.getName() + "#" + thread.getId() + "@" + hostname;
}

在分布式环境下，需要将机器名也作为key的一部分，避免UUID在多机器上出现重复的问题（虽然是小概率）。

最终使用的代码如下：

public void demo() {

    RedisCache redisCache = createRedisCache();
    RedisLockFactory lockFactory = new RedisLockFactory(redisCache);

    //分布式锁使用参考模板
    String lockKey = "test-lock-key";
    RedisLock redisLock = lockFactory.create(lockKey, 10, 10);
    try {
        boolean success = redisLock.tryLock();
        if (!success) {
            log.warn("【业务流程名】Redis锁[{}]申请失败", lockKey);
            return;
        }
        //region 业务处理代码
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //endregion
    } finally {
        if (redisLock.isLocking() && !redisLock.unlock()) {
            log.warn("【业务流程名】Redis锁[{}]释放失败", lockKey);
        }
    }
}

总结

本文通过几个代码示例详细介绍了 Redis 分布式锁的几个重要特性：

• 互斥性：在任意时刻，只允许一个客户端持有锁，确保了锁的独占性。
• 无死锁：即使在某个客户端持有锁的期间发生崩溃，未主动解锁的情况下，也能确保后续其他客户端能够正常加锁，避免了死锁问题。
• 自持自解：加锁和解锁必须由同一客户端（线程）完成，禁止客户端解除其他客户端持有的锁，确保了锁的所有权。

除了这些优点，我们也需要注意以下问题：

• 主从切换问题：在单实例环境中，该分布式锁方案是可行的。然而，在 Redis 集群环境中，尤其是在主从切换时，可能会出现问题。例如，当主节点挂掉，从节点升级为主节点，但数据尚未完全同步时，新的主节点上的锁信息可能会丢失，导致后续请求获取了无效的锁。为解决这一问题，可考虑使用 Redisson 框架的 Redlock 算法。
• 不支持重入：该分布式锁不支持重入，对于某些场景需要自己调用自己的递归调用可能会出现问题。为解决这一限制，可以参考 AQS 实现，对锁进行计数，每进入一次加1，每释放一次减1，数量为0时释放锁，实现了对锁的可重入性。

号外号外

总结了很多年的Java面试宝典，相关的高频面试点，全是大厂真题，并免费提供面试问题咨询、一对一简历优化、模拟面试~

地址：https://github.com/xbox1994/Java-Interview