Redis sorted set: リーダーボード、優先度キュー、スライディングウィンドウ

1. 概要

Sorted set (ZSET) は Redis で最も柔軟なデータ構造です。各要素（member）はスコア（score）に紐付けられており、Redis は常にそれらをスコア順に保持します。本記事では ZSET を使用して 3 つの一般的な問題を解決します：リーダーボード、優先度キュー/遅延キュー、そしてスライディングウィンドウ型のrate limiterです。すべて Java 21 と Spring Boot 3.x で記述されています。読み終わったら、各ユースケースに適切なスコアの選択方法と、よくあるrace conditionの回避方法を理解できるでしょう。

2. Sorted setの仕組み

ZSET の主な特性は以下の通りです：

• memberは一意です。既存のmemberに対してZADDを実行すると、スコアのみが更新されます。
• スコアはdouble型の実数です。同じスコアを持つ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の 2 つの依存関係は、セクション 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 と任意の選手のランクを取得する必要があります。ここでスコアはポイント、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は初期スコアを 0 と見なします。古いスコアを読み取って加算して書き戻す必要がないため、複数のリクエストが同時に同じプレイヤーのポイントを加算する場合でもrace conditionが発生しません。
• reverseRangeWithScores(KEY, 0, n - 1)はZREVRANGE ... WITHSCORESに対応しており、終了インデックスは inclusive です。したがって、トップ 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. スコアが同じ場合の処理

2 人が同じスコアの場合、Redisはmember名でソートします。ZREVRANGEでは辞書順が逆になるため、zoeはannaの前に来ます。これはビジネスロジックとしては望ましいことはほぼありません。スコアに達した時刻が早いほど上位にしたい場合は、スコアの小数部分に時刻を符号化できます：

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)で実際のスコアを取得します。

注意： スコアはdoubleで、2^53（約 9×10^15）までの整数のみを正確に表現できます。複数の値をスコアに結合する場合は、精度の損失を避けるために事前に大きさを計算する必要があります。上記の方法では、pointsは数百万以下である必要があります。さらに、この方法はincrementScoreでは使用できません。毎回加算するたびに小数部分も累積されるためです。

4. 優先度キューと遅延キュー

4.1. 優先度キュー

今回はスコアが優先度です：数値が小さいほど優先度が高いです。同じ優先度の job が FIFO 順で出るようにするには、スコアにタイムスタンプを結合します：

@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 の閾値を大きく下回ったままです。優先度 0 の job は、後から追加されたとしても、常に優先度 1 の job より先に出ます。
• popMinはZPOPMINに対応しており、このコマンドは要素を取得して削除するのを 1 つのアトミックステップで行います。2 つの worker が同じ job を受け取ることはありません。
• popMin(key, timeout, unit)はBZPOPMINに対応しており、job が到着するか、タイムアウトするまで connection をブロックします。これにより、CPU を消費する継続的なポーリングを避けることができます。

注意： BZPOPMINは待機中 connection を保有します。デフォルトでは Lettuce が共有 connection を使用するため、ブロッキングコマンドは他のコマンドもブロックします。connection pool を有効にすべきです（commons-pool2を追加してspring.data.redis.lettuce.pool.*を設定するか、またはworker に専用の connection factory を使用してください。

4.2. 遅延キュー

30 分後に支払いリマインダーメールを送信したいですか？スコアを実行時刻（epoch millis）とします。worker は定期的に、スコアが現在時刻以下の job を取得します。

以下の方法は間違っています：rangeByScoreを呼び出してからremoveします。2 つの worker が削除する前に同じ job を読み取る可能性があります。2 つのステップを Lua script にまとめて、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);
    }
}

script は最大 1 つの期限切れ job を取得し、それを削除します。Lua ではreturn falseは nil に変換されるため、Java 側はnullを受け取ります。worker は@Scheduled(fixedDelay = 500)でpollDueをループで呼び出して、空になるまで続行できます。

注意： pop 後、job は Redis から削除されます。worker が途中で死ぬと、job は失われます。at-least-once 保証が必要な場合は、job を “processing” ZSET に移動し（スコアは deadline）、処理完了時に削除してください。さらに複雑な要件がある場合は、Redis Streams または RabbitMQ や Kafka などの実際の message broker を検討してください。

5. スライディングウィンドウ型のrate limiter

Fixed window（毎分リクエスト数をカウント）は境界に弱点があります。例えば、00:59 に 100 request と 01:00 に 100 request はどちらも通過し、2 秒間で 200 request になります。スライディングウィンドウは ZSET に各 request のタイムスタンプを保存することで、これを修正します：

1. now - windowより古い request を削除します（ZREMRANGEBYSCORE）。
2. 残りの request の数をカウントします（ZCARD）。
3. 制限を超えていない場合は、現在の request をスコアnowで追加します。

3 つのステップはアトミックに実行する必要があるため、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 として使用すると、同じミリ秒内の 2 つの request が相互に上書きされ、1 つとしてカウントされます。
• PEXPIREにより、Redis はもはやアクティブでない顧客の key を自動的にクリーンアップできます。
• Clock bean を宣言する必要があります（例：Clock.systemUTC()）。Clockを inject するとテストが簡単になります。ただし、複数のアプリケーション instance がある場合、それらの clock は同期する必要があるか、script でredis.call('TIME')を使用して時刻を取得してください。

注意： 各 request は ZSET の 1 つの要素を消費します。制限が大きい場合、例えば顧客あたり時間 100,000 request の場合、メモリが急速に増加します。その場合、スライディングウィンドウカウンター（2 つの fixed window から推定）または token bucket を使用するべきです。

cluster mode では、script は同じ hash slot の key のみを処理できます。この記事の script はいずれも 1 つの key のみを使用しているため、問題はありません。

6. Testcontainers を使用したテスト

以下のテストは Docker 内の実際の Redis で実行されます。@ServiceConnectionは Spring に対して host と port を自動的に設定します。

@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();
    }
}

最後のテストは 2 つのことを確認します：c-42の 4 番目の request が拒否されていることと、制限が顧客ごとに個別に計算されていることです。ウィンドウの “スライド” をテストするには、調整可能なClock（例えば、独自に作成したMutableClockクラス）を inject し、時間を 10 秒進めてからtryAcquireを再度呼び出します。

7. Sorted set を使用すべきでない場合

• メッセージキューのような永続性が必要： ZSET には acknowledgement、consumer group、またはretry 機能がありません。Redis Streams、RabbitMQ、または Kafka のほうが適しています。
• ZSET が大きすぎる場合： 1 つの key に数千万の要素があると “big key” が発生します。DEL、slot migration、replication はすべて遅くなります。シーズン、地域、または日付別に key を分割する必要があります（例：leaderboard:2026-10）。
• スコアが絶対に正確である必要がある場合： 例えば、多くの小数点以下の桁数を持つ通貨。doubleには誤差があるため、最小単位（セント、円など）で整数として保存してください。
• 概算カウントで十分な場合： 「大約何人のユニークユーザーか」を知るだけの場合、HyperLogLog はメモリを大幅に少なくします。

8. まとめ

Sorted set は適切なスコアを選択することで、多くの問題を解決できます。リーダーボードの場合、スコアはポイントで、ZINCRBY、ZREVRANGE、ZREVRANKを使用します。優先度キューの場合、スコアはタイムスタンプと結合された優先度で、ZPOPMIN/BZPOPMINと組み合わせます。遅延キューの場合、スコアは実行予定時刻です。rate limiter の場合、スコアは各 request の時刻です。

2 つのことを覚えておく必要があります。まず、“読み取ってから書き込む” 操作のシーケンスは、すべてアトミックコマンドまたは Lua script でラップする必要があります。次に、複数の値をスコアに結合する場合、常にdoubleの精度の制限をチェックしてください。これら 2 つのことを理解することで、production システムで ZSET を自信を持って使用できます。