Redis 有序集合：排行榜、优先级队列和滑动窗口

1. 概述

Sorted set (ZSET) 是 Redis 中最灵活的数据结构。每个元素（member）都关联一个分数（score），Redis 始终按 score 的顺序保持它们。本文使用 ZSET 解决三个常见问题：排行榜（leaderboard）、优先级队列/延迟队列和滑动窗口 rate limiter。全部使用 Java 21 和 Spring Boot 3.x 编写。读完后，你将了解如何为每个问题选择 score，并避免常见的 race condition 错误。

2. Sorted set 如何工作

一个 ZSET 具有以下特性：

Member 是唯一的。ZADD 已存在的 member 只会更新它的 score。
Score 是浮点数（double）。具有相同 score 的 member 按 member 的字典序（lexicographic order）排序。
添加、删除、更新和获取排名的时间复杂度都是 O(log N)。获取一个范围的元素需要 O(log N + M)，其中 M 是返回的元素数。

你可以用 redis-cli 快速尝试：

ZADD scores 120 alice 95 bob 150 carol
ZREVRANGE scores 0 1 WITHSCORES
ZREVRANK scores bob

1) "carol"
2) "150"
3) "alice"
4) "120"
(integer) 2

ZREVRANGE 按分数降序返回元素。ZREVRANK 在该顺序中显示 bob 的位置（从 0 开始计数）。

2.1. 项目配置

添加 Spring Data Redis 依赖。两个 Testcontainers 依赖用于第 6 节的测试：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<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>

配置连接：

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

Spring Boot 自动创建 StringRedisTemplate bean。所有 ZSET 操作通过 opsForZSet()。

3. 排行榜

问题：游戏为玩家加分，需要获取前 N 名和任意一个人的排名。这里 score 是分数，member 是玩家 ID。

public record PlayerScore(String player, double score) {}

@Service
public class LeaderboardService {

    private static final String KEY = "leaderboard:game:2048";
    private final ZSetOperations<String, String> zSet;

    public LeaderboardService(StringRedisTemplate redis) {
        this.zSet = redis.opsForZSet();
    }

    public double addPoints(String player, double points) {
        return zSet.incrementScore(KEY, player, points);
    }

    public List<PlayerScore> top(int n) {
        Set<ZSetOperations.TypedTuple<String>> tuples =
                zSet.reverseRangeWithScores(KEY, 0, n - 1);
        if (tuples == null) {
            return List.of();
        }
        return tuples.stream()
                .map(t -> new PlayerScore(t.getValue(), t.getScore()))
                .toList();
    }

    public Optional<Long> rankOf(String player) {
        Long rank = zSet.reverseRank(KEY, player);
        return Optional.ofNullable(rank).map(r -> r + 1);
    }
}

逐部分解释：

incrementScore 对应于 ZINCRBY 命令。这个命令是原子的。如果 member 不存在，Redis 将初始 score 视为 0。你不需要读取旧分数、加上新分数再写回，所以当多个请求同时给一个人加分时不会有 race condition。
reverseRangeWithScores(KEY, 0, n - 1) 对应于 ZREVRANGE ... WITHSCORES。结束索引是包含的，所以前 10 名是 0..9。
reverseRank 如果玩家不在表中返回 null。我们加 1 使排名从 1 开始，更友好。

试试调用：

leaderboard.addPoints("alice", 120);
leaderboard.addPoints("bob", 95);
leaderboard.addPoints("carol", 150);
leaderboard.addPoints("bob", 60);

System.out.println(leaderboard.top(3));
System.out.println(leaderboard.rankOf("alice"));

[PlayerScore[player=bob, score=155.0], PlayerScore[player=carol, score=150.0], PlayerScore[player=alice, score=120.0]]
Optional[3]

3.1. 处理分数相同的情况

当两个人分数相同时，Redis 按 member 名称排序。使用 ZREVRANGE，字典序被反转，所以 zoe 排在 anna 之前。这个结果很少符合业务意图。如果你想先达到分数的人排在前面，可以将时间编码到 score 的小数部分：

public double addPointsWithTieBreak(String player, long points) {
    long now = System.currentTimeMillis();
    double tieBreak = 1.0 - (now / 1e13);
    double score = points + tieBreak;
    zSet.add(KEY, player, score);
    return score;
}

更早达到分数的人有较小的 now，所以 tieBreak 更大，排名更靠前。显示时，你可以使用 Math.floor(score) 得到实际分数。

注意： score 是 double，只能精确表示到 2^53 的整数（约 9×10^15）。当将多个值合并到一个 score 时，你需要预先计算大小以避免失去精度。用上面的方法，points 应该在几百万以下。此外，这种方法不能与 incrementScore 一起使用，因为每次增加都会累加小数部分。

4. 优先级队列和延迟队列

4.1. 优先级队列

这次 score 是优先级：数字越小优先级越高。要使相同优先级的 job 仍按 FIFO 顺序处理，我们将时间戳附加到 score 中：

@Service
public class JobQueue {

    private static final String KEY = "queue:email-jobs";
    private final ZSetOperations<String, String> zSet;

    public JobQueue(StringRedisTemplate redis) {
        this.zSet = redis.opsForZSet();
    }

    public void enqueue(String jobId, int priority) {
        if (priority < 0 || priority > 9) {
            throw new IllegalArgumentException("priority must be 0..9");
        }
        double score = priority * 1e13 + System.currentTimeMillis();
        zSet.add(KEY, jobId, score);
    }

    public Optional<String> poll() {
        var tuple = zSet.popMin(KEY);
        return Optional.ofNullable(tuple).map(ZSetOperations.TypedTuple::getValue);
    }

    public Optional<String> take(Duration timeout) {
        var tuple = zSet.popMin(KEY, timeout.toSeconds(), TimeUnit.SECONDS);
        return Optional.ofNullable(tuple).map(ZSetOperations.TypedTuple::getValue);
    }
}

要点：

priority * 1e13 + millis 的最大值约 9.2×10^13，仍远低于 2^53 的阈值。Priority 0 的 job 总是在 priority 1 的 job 之前处理，即使后添加。
popMin 对应于 ZPOPMIN。这个命令在一个原子步骤中获取并删除元素。两个 worker 永远不会收到相同的 job。
popMin(key, timeout, unit) 对应于 BZPOPMIN。这个命令会阻塞连接直到有 job 或超时，所以你可以避免持续轮询浪费 CPU。

注意： BZPOPMIN 在整个等待期间持有一个连接。使用默认的 Lettuce shared connection，blocking 命令会阻塞其他命令。你应该启用连接池（添加 commons-pool2 并配置 spring.data.redis.lettuce.pool.*）或为 worker 使用单独的连接工厂。

4.2. 延迟队列

想在 30 分钟后发送支付提醒邮件？使用 score 作为需要运行的时间点（epoch 毫秒数）。Worker 定期获取 score 小于或等于当前时间的 job。

下面的写法错误：先调用 rangeByScore 再 remove。两个 worker 可能在删除前读到同一个 job。我们将两个步骤放入一个 Lua 脚本，让 Redis 原子执行：

@Service
public class DelayedJobQueue {

    private static final String KEY = "queue:delayed";
    private static final RedisScript<String> POP_DUE = new DefaultRedisScript<>("""
            local jobs = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', ARGV[1], 'LIMIT', 0, 1)
            if #jobs == 0 then return false end
            redis.call('ZREM', KEYS[1], jobs[1])
            return jobs[1]
            """, String.class);

    private final StringRedisTemplate redis;

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

    public void schedule(String jobId, Instant runAt) {
        redis.opsForZSet().add(KEY, jobId, runAt.toEpochMilli());
    }

    public Optional<String> pollDue(Instant now) {
        String job = redis.execute(POP_DUE, List.of(KEY), String.valueOf(now.toEpochMilli()));
        return Optional.ofNullable(job);
    }
}

脚本最多获取一个已到期的 job 并删除它。在 Lua 中，return false 被转换为 nil，所以 Java 端收到 null。Worker 可以是一个 @Scheduled(fixedDelay = 500) 在循环中调用 pollDue 直到为空。

注意： pop 后，job 已从 Redis 中移除。如果 worker 中途崩溃，job 就丢失了。要确保 at-least-once，你可以将 job 移到一个 ZSET “processing”，score 为截止期限，处理完成后再删除。如果需求更复杂，考虑使用 Redis Streams 或真正的消息代理。

5. 滑动窗口 rate limiter

Fixed window（按分钟计数请求）在边界处有弱点。例如：在 00:59 有 100 个请求，01:00 有 100 个请求都通过，即 2 秒内有 200 个请求。滑动窗口通过将每个请求的时间戳存储在 ZSET 中来解决：

删除早于 now - window 的请求（ZREMRANGEBYSCORE）。
计数剩余请求数（ZCARD）。
如果未超过限制，添加当前请求，score 为 now。

这三个步骤必须原子执行，所以我们再次使用 Lua：

@Service
public class SlidingWindowRateLimiter {

    private static final RedisScript<Long> SCRIPT = new DefaultRedisScript<>("""
            local key    = KEYS[1]
            local now    = tonumber(ARGV[1])
            local window = tonumber(ARGV[2])
            local limit  = tonumber(ARGV[3])
            redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
            if redis.call('ZCARD', key) < limit then
                redis.call('ZADD', key, now, ARGV[4])
                redis.call('PEXPIRE', key, window)
                return 1
            end
            return 0
            """, Long.class);

    private final StringRedisTemplate redis;
    private final Clock clock;

    public SlidingWindowRateLimiter(StringRedisTemplate redis, Clock clock) {
        this.redis = redis;
        this.clock = clock;
    }

    public boolean tryAcquire(String customerId, int limit, Duration window) {
        Long allowed = redis.execute(SCRIPT,
                List.of("rate:" + customerId),
                String.valueOf(clock.millis()),
                String.valueOf(window.toMillis()),
                String.valueOf(limit),
                UUID.randomUUID().toString());
        return Long.valueOf(1L).equals(allowed);
    }
}

解释：

Member 是一个随机 UUID。如果使用时间戳作为 member，两个在同一毫秒内的请求会互相覆盖并被计数为一个。
PEXPIRE 帮助 Redis 自动清理不再活跃的客户的 key。
你需要声明一个 Clock bean（例如 Clock.systemUTC()）。注入 Clock 有利于测试。但是，如果有多个应用实例，它们的时钟应该同步，或者你可以在脚本中使用 redis.call('TIME') 获取时间。

注意： 每个请求占用 ZSET 中的一个元素。对于大限制，例如每小时 100,000 个请求，内存增长很快。此时你应该使用 sliding window counter（从两个 fixed window 估计）或 token bucket。

在集群模式中，脚本只能访问同一哈希槽的 key。本文的脚本每次只使用一个 key，所以没有问题。

6. 使用 Testcontainers 测试

下面的测试使用 Docker 中的真实 Redis 运行。@ServiceConnection 自动为 Spring 配置主机和端口。

@SpringBootTest
@Testcontainers
class SortedSetFeaturesTest {

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

    @Autowired LeaderboardService leaderboard;
    @Autowired JobQueue jobQueue;
    @Autowired SlidingWindowRateLimiter limiter;
    @Autowired StringRedisTemplate template;

    @BeforeEach
    void flush() {
        template.getConnectionFactory().getConnection().serverCommands().flushAll();
    }

    @Test
    void leaderboardOrdersByScoreDescending() {
        leaderboard.addPoints("alice", 120);
        leaderboard.addPoints("bob", 95);
        leaderboard.addPoints("bob", 60);

        assertThat(leaderboard.top(2))
                .extracting(PlayerScore::player)
                .containsExactly("bob", "alice");
        assertThat(leaderboard.rankOf("alice")).contains(2L);
        assertThat(leaderboard.rankOf("nobody")).isEmpty();
    }

    @Test
    void higherPriorityJobComesFirst() {
        jobQueue.enqueue("newsletter", 5);
        jobQueue.enqueue("password-reset", 0);

        assertThat(jobQueue.poll()).contains("password-reset");
        assertThat(jobQueue.poll()).contains("newsletter");
        assertThat(jobQueue.poll()).isEmpty();
    }

    @Test
    void rateLimiterRejectsAfterLimit() {
        Duration window = Duration.ofSeconds(10);

        assertThat(limiter.tryAcquire("c-42", 3, window)).isTrue();
        assertThat(limiter.tryAcquire("c-42", 3, window)).isTrue();
        assertThat(limiter.tryAcquire("c-42", 3, window)).isTrue();
        assertThat(limiter.tryAcquire("c-42", 3, window)).isFalse();
        assertThat(limiter.tryAcquire("c-99", 3, window)).isTrue();
    }
}

最后一个测试验证两件事：c-42 的第四个请求被拒绝，并且每个客户都有独立的限制。要测试“滑动窗口”的行为，你可以注入一个可调整的 Clock（例如一个自定义 MutableClock 类），提前 10 秒再调用 tryAcquire。

7. 何时不应使用 sorted set

需要消息队列的持久性： ZSET 没有确认、消费者组或重试。Redis Streams、RabbitMQ 或 Kafka 更合适。
单个 ZSET 过大： 一个 key 中有数千万个元素会导致“big key”问题。DEL、迁移 slot 或复制都会很慢。你应该按赛季、区域或日期分割 key（leaderboard:2026-10）。
Score 需要绝对精确： 例如有多个小数位的货币。double 有舍入误差，所以你应该以最小单位（分、美分）以整数形式存储。
近似计数足够： 如果只需要知道“大约有多少唯一用户”，HyperLogLog 内存消耗少得多。

8. 总结

Sorted set 通过选择合适的 score 可以解决许多问题。对于排行榜，score 是分数，你使用 ZINCRBY、ZREVRANGE、ZREVRANK。对于优先级队列，score 是优先级加上时间戳，结合 ZPOPMIN/BZPOPMIN。对于延迟队列，score 是需要运行的时间点。对于 rate limiter，score 是每个请求的时间戳。

需要记住两件事。首先，所有“读后写”的操作序列必须用原子命令或 Lua 脚本包装。其次，在将多个值合并到一个 score 时，始终检查 double 的精度限制。掌握这两点，你就可以自信地在生产系统中使用 ZSET。

Redis 有序集合：排行榜、优先级队列和滑动窗口

1. 概述

2. Sorted set 如何工作

2.1. 项目配置

3. 排行榜

3.1. 处理分数相同的情况

4. 优先级队列和延迟队列

4.1. 优先级队列

4.2. 延迟队列

5. 滑动窗口 rate limiter

6. 使用 Testcontainers 测试

7. 何时不应使用 sorted set

8. 总结

同主题文章

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

Redis Bloom 过滤器：Spring Boot 缓存穿透防护

Redis Streams 与消费者组：可靠的事件处理

Redis 中的 hot key 和 big key：检测与处理

Redis 持久化：RDB、AOF 和何时会丢失数据