当前位置：首页 > news >正文

【分布式锁通关指南 12】源码剖析redisson如何利用Redis数据结构实现Semaphore和CountDownLatch

news 2025/11/4 9:15:51

引言

本期我们继续探索 Redisson 的分布式同步组件。本篇聚焦在 Semaphore（信号量）和 CountDownLatch（倒数闩），看看它们在单机环境中常用的同步思想是如何“穿越”到分布式场景，并依赖 Redis 来保证一致性的。

在普通的 Java 线程编程中，Semaphore 常用于控制对特定资源的同时访问数，而 CountDownLatch 则常用于等待其他线程把某些工作执行完毕后再继续。但若应用部署在分布式环境，单机版本的对象自然约束在某一 JVM 中，无法跨节点协同控制。此时我们可以借助 Redisson 提供的 RSemaphore 和 RCountDownLatch，让这两种同步模式在分布式下也能“有法可依”。

介绍

RSemaphore

使用场景：

可以限制同时访问共享资源（如数据库连接、缓存访问、某段关键逻辑等）的并发数量。

关键特性：

1.可动态调整可用许可数量；

2.在分布式集群环境下，任何一个节点都能正确感知许可的剩余数量。

RCountDownLatch

使用场景：

类似“倒计时”，需要等待某些操作完成后才能继续。例如：只有当分布式系统中多个子任务都执行完成后，主流程才能继续执行。

关键特性：

1.可设置初始倒计数值，不同节点根据执行进度减少计数。

2.当计数归零时，就会唤醒等待该闩的节点。

接下来，让我们直击它们在源码中的实现方式。

Semaphore 的实现思路

Redisson 提供的分布式信号量接口 RSemaphore 在底层主要依赖 Redis 的 Lua 脚本 和 Redis key 自增/自减 这类能力：

获取许可 (acquire) ：

Lua 脚本会检查是否存在足够的许可值（通常是 Redis 中的某个 key 记录当前可用许可数）；

如果足够，则进行减操作；否则阻塞或返回失败。
释放许可 (release) ：

将 Redis 中存储的许可数 +1，并通知阻塞挂起中的线程（如果有）。
可重入机制（可选场景）：
一些分布式信号量场景还会考虑同一个线程可否重复获取许可，但多数情况下信号量都是面向线程独立的计数控制，所以可能不涉及可重入。

核心结构

RedissonSemaphore 源码示意，整体只展示关键思路：

public class RedissonSemaphore extends RedissonBaseDistributedObject implements RSemaphore {public RedissonSemaphore(CommandAsyncExecutor commandExecutor, String name) {super(commandExecutor, name);}@Overridepublic void acquire() throws InterruptedException {tryAcquire(-1, TimeUnit.MILLISECONDS); }@Overridepublic boolean tryAcquire(long waitTime, TimeUnit unit) throws InterruptedException {return get(tryAcquireAsync(waitTime, unit, 1));}@Overridepublic void release() {get(releaseAsync(1));}// ...
}

RedissonSemaphore 依托一些异步方法，例如 tryAcquireAsync、releaseAsync 等，这些方法内部最终会组装 Lua 脚本并发送到 Redis，来做自增自减和阻塞通知。

获取许可

在 tryAcquireAsync 中，一般会：

先检查 Redis 中的可用许可数（key 的数值）是否足够；
若足够，做 key = key - 1；
若不足则阻塞或直接返回 false（根据 tryAcquire 的超时时间和模式）。

private RFuture<Boolean> tryAcquireAsync(long waitTime, TimeUnit unit, int permits) {// Lua脚本核心：判断key是否 >= permits，否则阻塞/返回// 执行成功后减去相应数量return commandExecutor.evalWriteAsync(getName(), RedisCommands.EVAL_BOOLEAN, "local currVal = redis.call('get', KEYS[1]); " + "if (currVal ~= false and tonumber(currVal) >= tonumber(ARGV[1])) then " +"redis.call('decrby', KEYS[1], ARGV[1]); " + "return 1; " +"end; " +"return 0;", Collections.<Object>singletonList(getName()),permits);
}

此处脚本只是示意，真实的 Redisson 源码会更全面，比如要处理阻塞和等待通知的逻辑。

释放许可

与获取许可对立，release 时关键就是将 Redis 中存储的许可数 +1，或一次性释放多个许可：

private RFuture<Long> releaseAsync(int permits) {// Lua脚本：对key执行incr操作return commandExecutor.evalWriteAsync(getName(),RedisCommands.EVAL_LONG,"return redis.call('incrby', KEYS[1], ARGV[1]);",Collections.<Object>singletonList(getName()),permits);
}

当可用许可数从 0 变为正数，会通知等待获取许可的线程。

CountDownLatch 的实现思路

Redisson 的 RCountDownLatch 和 Semaphore 类似，同样通过 Redis 来维护一个“倒计数值”。当这个值大于 0 时，需要阻塞调用 await() 的线程；当计数归零后，所有阻塞线程会被唤醒。

核心结构

源码示意：

public class RedissonCountDownLatch extends RedissonBaseDistributedObject implements RCountDownLatch {public RedissonCountDownLatch(CommandAsyncExecutor commandExecutor, String name) {super(commandExecutor, name);}@Overridepublic void await() throws InterruptedException {get(awaitAsync());}@Overridepublic void countDown() {get(countDownAsync());}@Overridepublic boolean trySetCount(long count) {return get(trySetCountAsync(count));}// ...
}

我们可以看到几个核心方法：

trySetCount：初始化或重置倒计数值。
countDown：使计数值 -1。
await：阻塞等待，直到计数值清零。

初始化计数

trySetCount 负责在 Redis 中设置 key 的初始计数值：

private RFuture<Boolean> trySetCountAsync(long count) {// 如果key不存在，则设置它为 count// 如果已经存在，则返回false 表示无法再次设置return commandExecutor.evalWriteAsync(getName(),RedisCommands.EVAL_BOOLEAN,"if redis.call('exists', KEYS[1]) == 0 then " +"redis.call('set', KEYS[1], ARGV[1]); " +"return 1; " +"end; " +"return 0;",Collections.singletonList(getName()),count);
}

若 key 已存在，这次设置就会失败，说明已经有倒计数在进行中。

递减计数

countDown 的 Lua 脚本则做 decrement 操作。如果计数归零，就发布一个消息或事件，唤醒所有 await() 中阻塞的线程（它们往往在 Redis 中通过订阅频道或等待队列的方式实现阻塞唤醒）。

private RFuture<Long> countDownAsync() {return commandExecutor.evalWriteAsync(getName(),RedisCommands.EVAL_LONG,"local val = redis.call('decr', KEYS[1]); " + // 如果归零，发布消息到特定 channel，唤醒等待方"if val <= 0 then " +"redis.call('publish', KEYS[2], '0'); " +"end;" +"return val;",Arrays.asList(getName(), getChannelName(getName())));
}