ZH
FrançaisPortuguêsEspañolDeutschTiếng ViệtEnglish简体中文한국어日本語
首页 / 架构与设计 / Redis

Spring Boot 中使用 Redis 实现分布式锁

Cris The Dev · 2026/10/13 · 9 分钟阅读 · 112

1. 概述

当应用在多个实例上运行时，synchronized 或 ReentrantLock 就不再足够了。这些锁只在单个 JVM 内有效。本文将以两种方式指导你构建分布式锁：自己使用 Spring Data Redis 实现，以及使用 Redisson 库。阅读完成后，你将了解如何安全地获取和释放锁、如何为锁编写测试，以及 Redis 锁的安全性限制。

2. 为什么需要分布式锁

假设你有一个 OrderService 运行在 3 个 Pod 上。两个客户同时购买了仓库中最后一个商品。两个请求分别到达两个不同的 Pod。它们都读到 stock = 1，都减少了库存，结果是超卖了。

另一个例子是一个 @Scheduled 的任务在每天早上 2 点生成报告。如果有 3 个实例，该任务就会运行 3 次，客户会收到 3 封完全相同的邮件。

在这两种情况下，你都需要一个所有实例都能看到的锁。Redis 非常适合这个角色，原因有三：

  • Redis 在单个线程上依次处理命令，所以每条命令都是原子的。
  • Key 可以设置 TTL，所以如果持有锁的进程崩溃了，锁会自动过期。
  • 延迟低，而且大多数系统已经有 Redis。

3. 原理：SET NX PX 命令

整个机制基于一条命令：

redis-cli SET lock:product:42 "c1f9-token" NX PX 10000

每个部分的含义：

  • NX：仅当 key 不存在时才写入。这是“获取锁”的步骤。
  • PX 10000：key 在 10 秒后自动过期。这样，如果进程中途崩溃，锁不会永远被卡住。
  • "c1f9-token"：一个随机值，对每次获取锁都是唯一的。这个值用来确定谁持有锁。

连续运行上面的命令两次，你会看到：

OK
(nil)

第一次返回 OK，表示你获得了锁。第二次返回 (nil)，因为 key 已经存在。

注意： 不要先用 SETNX，再用单独的 EXPIRE 命令。如果进程在两条命令之间崩溃，key 将永远不会过期，你会陷入死锁。

4. 使用 Spring Data Redis 自己实现

4.1. 依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

在 application.properties 中配置连接：

spring.data.redis.host=localhost
spring.data.redis.port=6379

4.2. 获取和释放锁

@Component
public class RedisLock {

    private static final DefaultRedisScript<Long> RELEASE_SCRIPT = new DefaultRedisScript<>("""
            if redis.call('get', KEYS[1]) == ARGV[1] then
                return redis.call('del', KEYS[1])
            else
                return 0
            end
            """, Long.class);

    private final StringRedisTemplate redis;

    public RedisLock(StringRedisTemplate redis) {
        this.redis = redis;
    }

    public Optional<String> tryLock(String key, Duration ttl) {
        String token = UUID.randomUUID().toString();
        Boolean acquired = redis.opsForValue().setIfAbsent(key, token, ttl);
        return Boolean.TRUE.equals(acquired) ? Optional.of(token) : Optional.empty();
    }

    public boolean unlock(String key, String token) {
        Long deleted = redis.execute(RELEASE_SCRIPT, List.of(key), token);
        return deleted != null && deleted == 1L;
    }
}

解释：

  • setIfAbsent(key, token, ttl) 生成正确的 SET key token NX PX ttl 命令。返回值 true 表示你已经获得锁。
  • tryLock 在成功时返回 token。你必须保存这个 token，之后才能释放锁。
  • unlock 运行一个 Lua 脚本。脚本检查 key 的值是否确实是你的 token，并且只在匹配时删除 key。

为什么不直接调用 DEL？考虑这种情况。Instance A 用 TTL 10 秒获取锁，但处理花费 12 秒。在第 10 秒，锁过期，instance B 获得锁。在第 12 秒，A 调用 DEL 并删除了 B 的锁。现在 instance C 可以进入，B 和 C 并行运行。

如果你分开调用 GET 然后 DEL，两条命令之间还是有间隙。Lua 脚本由 Redis 原子性执行，所以没有这样的间隙。

4.3. 在 service 中使用

public record Order(long productId, int quantity) {}

@Service
public class OrderService {

    private final RedisLock redisLock;
    private final InventoryRepository inventory;

    public OrderService(RedisLock redisLock, InventoryRepository inventory) {
        this.redisLock = redisLock;
        this.inventory = inventory;
    }

    public void placeOrder(Order order) {
        String key = "lock:product:" + order.productId();
        String token = redisLock.tryLock(key, Duration.ofSeconds(10))
                .orElseThrow(() -> new IllegalStateException("商品正在处理中，请稍后重试"));
        try {
            inventory.decrease(order.productId(), order.quantity());
        } finally {
            redisLock.unlock(key, token);
        }
    }
}

有三点需要注意：

  • 锁的 key 应该针对具体资源，这里是 lock:product:{id}。如果对所有商品使用同一把锁，所有订单将被序列化处理。
  • unlock 总是在 finally 块中，确保即使发生异常也会释放锁。
  • 这个例子在无法获取锁时快速失败。如果要等待，你可以重试几次，使用短的退避。不应该使用无限制的 while(true) 循环。

InventoryRepository 是你项目中的任何仓库，比如 Spring Data JPA 仓库。

5. 使用 Redisson

自己实现可以帮助你理解机制，但还缺少几个功能：当处理花费很长时间时自动延长锁的期限、可重入性，以及使用 pub/sub 而不是轮询来等待锁。Redisson 提供了这些功能。

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.45.1</version>
</dependency>

你应该检查与项目中 Spring Boot 版本兼容的最新版本。这个 starter 会读取 spring.data.redis.* 配置并自动创建 RedissonClient bean。

@Service
public class DailyReportJob {

    private static final Logger log = LoggerFactory.getLogger(DailyReportJob.class);
    private final RedissonClient redisson;

    public DailyReportJob(RedissonClient redisson) {
        this.redisson = redisson;
    }

    @Scheduled(cron = "0 0 2 * * *")
    public void generate() throws InterruptedException {
        RLock lock = redisson.getLock("lock:daily-report");
        if (!lock.tryLock(0, TimeUnit.SECONDS)) {
            log.info("另一个实例正在运行报告，跳过");
            return;
        }
        try {
            log.info("开始生成每日报告");
            // 查询数据、导出文件、发送邮件...
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}

当 job 在 3 个实例上运行时，日志会像这样：

[pod-1] 开始生成每日报告
[pod-2] 另一个实例正在运行报告，跳过
[pod-3] 另一个实例正在运行报告，跳过

重要的几点：

  • tryLock(0, TimeUnit.SECONDS) 不等待。如果锁被持有，该方法立即返回 false。
  • 当你不指定 leaseTime 时，Redisson 启用 watchdog。锁的 TTL 默认为 30 秒（lockWatchdogTimeout），并且大约每 10 秒续一次，只要线程仍然持有锁。如果 JVM 死亡，续期停止，锁自动过期。
  • 当你用具体的 leaseTime 调用 tryLock(wait, lease, unit) 时，watchdog 被禁用。锁会在 lease 时间后过期，不管工作是否完成。
  • RLock 与线程关联，是可重入的。如果从不同的线程调用 unlock()，你会得到 IllegalMonitorStateException。与 CompletableFuture 或响应式代码一起使用时需要特别小心。

注意： 因为 RLock 与线程关联，使用虚拟线程和那些在中途切换线程的框架时需要小心。锁必须在同一线程上获取和释放。

6. 使用 Testcontainers 编写测试

你应该在真实 Redis 上测试锁，不要使用 mock。添加以下测试依赖（版本由 Spring Boot 管理）：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-testcontainers</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>junit-jupiter</artifactId>
    <scope>test</scope>
</dependency>
@DataRedisTest
@Import(RedisLock.class)
@Testcontainers
class RedisLockTest {

    @Container
    @ServiceConnection(name = "redis")
    static GenericContainer<?> redis = new GenericContainer<>("redis:7-alpine").withExposedPorts(6379);

    @Autowired
    RedisLock redisLock;

    @Autowired
    StringRedisTemplate template;

    @Test
    void givenConcurrentRequests_whenTryLock_thenOnlyOneSucceeds() throws Exception {
        List<Callable<Optional<String>>> tasks = IntStream.range(0, 20)
                .<Callable<Optional<String>>>mapToObj(i -> () -> redisLock.tryLock("lock:concurrent", Duration.ofSeconds(5)))
                .toList();

        try (ExecutorService executor = Executors.newFixedThreadPool(20)) {
            long acquired = executor.invokeAll(tasks).stream()
                    .map(Future::resultNow)
                    .filter(Optional::isPresent)
                    .count();

            assertThat(acquired).isEqualTo(1);
        }
    }

    @Test
    void givenWrongToken_whenUnlock_thenLockIsKept() {
        String token = redisLock.tryLock("lock:owner", Duration.ofSeconds(5)).orElseThrow();

        assertThat(redisLock.unlock("lock:owner", "not-my-token")).isFalse();
        assertThat(template.hasKey("lock:owner")).isTrue();

        assertThat(redisLock.unlock("lock:owner", token)).isTrue();
        assertThat(template.hasKey("lock:owner")).isFalse();
    }
}

解释：

  • @DataRedisTest 仅加载 Redis 配置部分，所以测试运行很快。@Import(RedisLock.class) 将被测试的 bean 引入上下文。
  • @ServiceConnection(name = "redis") 自动将 StringRedisTemplate 指向刚启动的容器。你不需要手动配置主机和端口。
  • 第一个测试让 20 个线程竞争一把锁，验证只有 1 个线程获胜。
  • 第二个测试验证 Lua 脚本工作正确：错误的 token 无法删除锁，正确的 token 可以。

7. 常见错误和限制

7.1. TTL 短于处理时间

这是最常见的错误。锁在工作仍在进行时过期，另一个实例会进入。问题不仅是代码慢。GC 暂停、网络延迟，或等待连接池也可能导致你超过 TTL。

有两种处理方法。第一种是使用 Redisson 的 watchdog。第二种是设置充足的 TTL，比如 p99 处理时间的几倍。

7.2. 锁不提供绝对保证

Redis 复制是异步的。假设 master 接收到锁但在复制到 replica 前崩溃了。Replica 被提升为新 master，上面没有这个 key。此时另一个客户端可以获得相同的锁。

Redlock 算法（在 N 个独立节点中获取多数派的锁）降低了这种风险。然而，Redlock 仍然有争议，因为它依赖于时钟和延迟的假设。在 Redisson 中，RedissonRedLock 现已被标记为过时。

如果数据错误造成严重后果，比如重复扣款，不要仅依赖锁。结合使用 fencing token：每次获取锁时，你会收到一个递增的数字（比如来自 INCR lock:product:42:fence）。存储层会拒绝携带旧 token 的写入：

UPDATE inventory
SET stock = stock - :qty, last_fence = :fence
WHERE product_id = :id AND last_fence < :fence;

如果 UPDATE 影响 0 行，说明一个带更新 token 的进程已经先写入了。旧的请求会被拒绝。

7.3. 其他错误

  • 对所有资源使用同一个 key，导致吞吐量严重下降。
  • 忘记 finally，所以锁只在 TTL 过期时才释放。
  • 当不再持有锁时调用 Redisson 的 unlock()（因为已过期）。在调用前检查 isHeldByCurrentThread()。
  • 连续重试获取锁，没有退避，导致 Redis 被请求淹没。

7.4. 何时不应使用

  • 应用只有一个实例：ReentrantLock 足够且简单得多。
  • 数据在关系数据库中：乐观锁（@Version）、SELECT ... FOR UPDATE，或 unique 约束通常更安全。它们与数据在同一事务中。
  • 需要绝对保证：考虑使用具有共识的系统，如 ZooKeeper 或 etcd，结合 fencing token。
  • 只需要消息恰好处理一次：幂等性 key 或消息 broker 的 consumer group 通常比锁更合适。

8. 总结

Redis 中的分布式锁基于三个原则：使用原子 SET NX PX 命令获取锁、为每次获取锁绑定唯一 token，以及使用带 token 检查的 Lua 脚本释放锁。你可以用 StringRedisTemplate 自己实现（只需几十行代码），或使用 Redisson 以获得 watchdog、可重入性和高效的等待机制。

记住，Redis 锁在大多数情况下有助于避免重复处理，但不是绝对保证。对于关键操作，结合锁和 fencing token 或数据库约束。此外，使用 Testcontainers 在真实 Redis 上测试你的锁。

redisdistributed-lockspring-bootredissonconcurrency
作者的应用Glowmi · Lantern NightA cozy Mid-Autumn match-3 puzzleiOS · Android

同主题文章