Redis StreamsとConsumer Group：信頼性の高いイベント処理

1. 概要

Redisの pub/sub には大きな欠点があります。consumerが切れているときに publish されたメッセージは失われます。Redis Streamsはこの問題を解決します。メッセージはストリームに永続的に保存され、各 consumer group はどのメッセージが処理完了したかを記憶しています。

このガイドでは、Spring Boot 3 と Spring Data Redis を使用して、注文イベント処理パイプラインを構築する方法を説明します。このパイプラインには、イベントの記録、consumer group による読み込み、手動 ACK、保留メッセージの再処理、およびdead-letterストリームへのエラーメッセージ転送が含まれます。

2. Redis Streams と Consumer Group のしくみ

Java を書く前に、redis-cli でネイティブコマンドを試してみるべきです：

XADD orders * orderId 1001 amount 250000
XGROUP CREATE orders shipping-group 0
XREADGROUP GROUP shipping-group worker-1 COUNT 1 STREAMS orders >
XPENDING orders shipping-group
XACK orders shipping-group 1728640000000-0

各コマンドの意味：

XADD はstream orders にentryを追加します。* はRedisに<millis>-<seq>形式のIDを自動生成させます。
XGROUP CREATE ... 0 はgroup shipping-group を作成し、ストリームの最初から読み込みを開始します。$ を使用した場合、groupは新しいメッセージだけを読み込みます。
XREADGROUP ... > はgroup内でまだ誰にも割り当てられていないメッセージを取得します。メッセージはworker-1に割り当てられ、同時にgroupの**Pending Entries List (PEL)**に記録されます。
XPENDING は、割り当てられたが確認されていないメッセージの数を表示します。
XACK はPELからメッセージを削除します。この時点から、groupはメッセージが処理完了したと見なします。

ACK前の XPENDING の出力は次のようになります：

1) (integer) 1
2) "1728640000000-0"
3) "1728640000000-0"
4) 1) 1) "worker-1"
      2) "1"

PELは信頼性の鍵です。例えば、consumerがメッセージを受け取った後、ACKする前にcrashした場合を考えてください。メッセージはPELに残ったままであり、別のconsumerがそれを受け取り(claim)、処理を続けることができます。これがat-least-onceセマンティクスです。

3. セットアップ

Spring Data Redis の dependency を追加します。このstarterはデフォルトのclientとしてLettuceを使用します：

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

接続設定とストリーム名を構成します：

spring:
  data:
    redis:
      host: localhost
      port: 6379

app:
  stream:
    key: orders
    group: shipping-group
    consumer: ${HOSTNAME:worker-local}

consumer名はHOSTNAME から取得されるため、各instance（pod）には一意の名前が付きます。これは重要です。2つのinstanceが同じ名前を使用した場合、Redisはそれらを1つのconsumerと見なし、PELが混在します。

4. ストリームにイベントを書き込む

@Service
public class OrderEventPublisher {

    private final StringRedisTemplate redisTemplate;
    private final String streamKey;

    public OrderEventPublisher(StringRedisTemplate redisTemplate,
                               @Value("${app.stream.key}") String streamKey) {
        this.redisTemplate = redisTemplate;
        this.streamKey = streamKey;
    }

    public RecordId publishOrderCreated(String orderId, long amount) {
        MapRecord<String, String, String> record = StreamRecords.newRecord()
                .in(streamKey)
                .ofMap(Map.of(
                        "type", "ORDER_CREATED",
                        "orderId", orderId,
                        "amount", String.valueOf(amount)));

        return redisTemplate.opsForStream().add(record);
    }
}

StreamRecords.newRecord().in(...).ofMap(...) は、文字列フィールドを持つ MapRecord を作成します。opsForStream().add() コマンドは XADD orders * ... に対応し、Redisが生成した RecordId （例：1728640000000-0）を返します。

StringRedisTemplate を使用すると、Redisのデータが redis-cli で読み取り可能になります。JDKのserializationのバイト文字列を処理する必要はありません。

5. 安全に consumer group を作成する

groupは読み込み前に存在する必要があります。アプリケーションはストリームにメッセージがない状態で起動される可能性があるため、MKSTREAM パラメータが必要です。StreamOperations の createGroup() メソッドはこのパラメータを有効にしていません。したがって、我々は接続層に下ります：

@Component
public class StreamGroupInitializer {

    private final StringRedisTemplate redisTemplate;

    public StreamGroupInitializer(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public void ensureGroup(String streamKey, String group) {
        try {
            redisTemplate.execute((RedisCallback<String>) connection ->
                    connection.streamCommands().xGroupCreate(
                            streamKey.getBytes(StandardCharsets.UTF_8),
                            group,
                            ReadOffset.from("0"),
                            true)); // MKSTREAM
        } catch (DataAccessException e) {
            String msg = NestedExceptionUtils.getMostSpecificCause(e).getMessage();
            if (msg == null || !msg.contains("BUSYGROUP")) {
                throw e;
            }
            // Groupは既に存在しています：アプリケーションの再起動時は正常です
        }
    }
}

最後のパラメータ true は MKSTREAM を有効にします。Redisはストリームがまだない場合、空のストリームを作成します。アプリケーションを再起動すると、Redisは BUSYGROUP Consumer Group name already exists エラーを返します。この特定のエラーは無視し、他のエラーを再スローします。

6. StreamMessageListenerContainer でメッセージを読み込む

6.1. 手動 ACK を使用する Listener

@Component
public class OrderEventListener
        implements StreamListener<String, MapRecord<String, String, String>> {

    private final StringRedisTemplate redisTemplate;
    private final ShippingService shippingService;
    private final String streamKey;
    private final String group;

    public OrderEventListener(StringRedisTemplate redisTemplate,
                              ShippingService shippingService,
                              @Value("${app.stream.key}") String streamKey,
                              @Value("${app.stream.group}") String group) {
        this.redisTemplate = redisTemplate;
        this.shippingService = shippingService;
        this.streamKey = streamKey;
        this.group = group;
    }

    @Override
    public void onMessage(MapRecord<String, String, String> record) {
        Map<String, String> body = record.getValue();
        shippingService.createShipment(body.get("orderId"));

        redisTemplate.opsForStream().acknowledge(streamKey, group, record.getId());
    }
}

ACK コマンドは createShipment が成功した場合にのみ実行されます。method が例外をスローした場合、メッセージはPELに留まり、後で再処理されます。処理前にACKしてはいけません。そうするとat-least-onceがat-most-onceになります。

このエクササイズ内の ShippingService は単純なserviceです。既に配送伝票が作成されている orderId を記録しています：

@Service
public class ShippingService {

    private final Set<String> shippedOrders = ConcurrentHashMap.newKeySet();

    public void createShipment(String orderId) {
        if (!shippedOrders.add(orderId)) {
            return; // すでに処理済み、スキップ（べき等性）
        }
        // 配送業者のAPIを呼び出す...
    }

    public Set<String> shippedOrders() {
        return Set.copyOf(shippedOrders);
    }
}

6.2. Container の設定

@Configuration
public class StreamConfig {

    @Bean(destroyMethod = "stop")
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>>
    orderStreamContainer(RedisConnectionFactory connectionFactory,
                         StreamGroupInitializer initializer,
                         OrderEventListener listener,
                         @Value("${app.stream.key}") String streamKey,
                         @Value("${app.stream.group}") String group,
                         @Value("${app.stream.consumer}") String consumerName) {

        initializer.ensureGroup(streamKey, group);

        var options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofSeconds(2))
                .batchSize(10)
                .build();

        var container = StreamMessageListenerContainer.create(connectionFactory, options);

        var request = StreamMessageListenerContainer.StreamReadRequest
                .builder(StreamOffset.create(streamKey, ReadOffset.lastConsumed()))
                .consumer(Consumer.from(group, consumerName))
                .autoAcknowledge(false)
                .cancelOnError(t -> false)
                .build();

        container.register(request, listener);
        container.start();
        return container;
    }
}

設定の主なポイント：

ReadOffset.lastConsumed() は XREADGROUP の > に対応します。つまり、groupの新しいメッセージのみを取得します。
pollTimeout は各読み込みの block 時間です。batchSize は1回の読み込みで最大のメッセージ数です。
autoAcknowledge(false) は自動ACKをオフにします。listener内で自分自身でACKします。
cancelOnError(t -> false) が最も重要な行です。デフォルトでは、listenerが例外をスローすると、containerはsubscription をキャンセルします。consumerは警告なく暗黙的に読み込みを停止します。

注意： 重要なビジネスロジックに receiveAutoAck(...) を使用しないでください。このモードは読み込み時にlistenerが完了する前にACKします。アプリケーションが途中でcrashした場合、メッセージは永久に失われます。

7. 保留メッセージと dead-letter の処理

Listenerは新しいメッセージ（>）のみを読み込みます。エラーメッセージ、または既に削除されたconsumerに属するメッセージは、処理されなければPELに永遠に残ります。そのために定期的なjobが必要です：

@Component
public class PendingOrderReclaimer {

    private static final Duration MIN_IDLE = Duration.ofMinutes(1);
    private static final long MAX_DELIVERIES = 5;

    private final StringRedisTemplate redisTemplate;
    private final ShippingService shippingService;
    private final String streamKey;
    private final String group;
    private final String consumerName;

    public PendingOrderReclaimer(StringRedisTemplate redisTemplate,
                                 ShippingService shippingService,
                                 @Value("${app.stream.key}") String streamKey,
                                 @Value("${app.stream.group}") String group,
                                 @Value("${app.stream.consumer}") String consumerName) {
        this.redisTemplate = redisTemplate;
        this.shippingService = shippingService;
        this.streamKey = streamKey;
        this.group = group;
        this.consumerName = consumerName;
    }

    @Scheduled(fixedDelay = 30_000)
    public void reclaim() {
        StreamOperations<String, String, String> ops = redisTemplate.opsForStream();
        PendingMessages pending = ops.pending(streamKey, group, Range.unbounded(), 50);

        for (PendingMessage pm : pending) {
            if (pm.getElapsedTimeSinceLastDelivery().compareTo(MIN_IDLE) < 0) {
                continue;
            }
            List<MapRecord<String, String, String>> claimed =
                    ops.claim(streamKey, group, consumerName, MIN_IDLE, pm.getId());

            for (MapRecord<String, String, String> record : claimed) {
                if (pm.getTotalDeliveryCount() >= MAX_DELIVERIES) {
                    moveToDeadLetter(record);
                } else {
                    shippingService.createShipment(record.getValue().get("orderId"));
                }
                ops.acknowledge(streamKey, group, record.getId());
            }
        }
    }

    private void moveToDeadLetter(MapRecord<String, String, String> record) {
        Map<String, String> body = new HashMap<>(record.getValue());
        body.put("originalId", record.getId().getValue());
        redisTemplate.opsForStream().add(
                StreamRecords.newRecord().in(streamKey + ":dlq").ofMap(body));
    }
}

jobの処理フロー：

pending(...) は XPENDING orders shipping-group - + 50 に対応します。各 PendingMessage は、consumerがメッセージを保持しているか、idleの時間と配信回数を示します。
claim(...) は XCLAIM に対応します。このコマンドは所有権を現在のconsumerに転送します。Redisはメッセージが少なくとも MIN_IDLE の間idleのままの場合のみ転送するため、2つのinstanceが同時にjobを実行しても、重複claim されません。claimするたびに配信回数も増加します。
MAX_DELIVERIES を超えて配信されたメッセージは、stream orders:dlq にコピーされた後、ACKされます。永続的に不良なメッセージ（poison message）は、無限に再試行されることはありません。

createShipment が例外をスローした場合、ACK コマンドは実行されません。メッセージはPELに残り、次のサイクルで再試行されます。@Configuration class で @EnableScheduling を有効にすることを忘れないでください。

注意： Redis 6.2以降には XAUTOCLAIM コマンドがあります。このコマンドは XPENDING と XCLAIM のステップを1つに統合し、PELが大きい場合に効率的です。Spring Data Redisの現在のバージョンがまだ便利なAPIを持っていない場合、redisTemplate.execute(...) 経由でこれを呼び出すことができます。

8. Testcontainers でテストを書く

テスト用の 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>
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <scope>test</scope>
</dependency>

以下のテストはイベントを publish します。その後、配送伝票が作成されたこと、およびPELが空であることを確認します：

@SpringBootTest
@Testcontainers
class OrderStreamIntegrationTest {

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

    @Autowired OrderEventPublisher publisher;
    @Autowired ShippingService shippingService;
    @Autowired StringRedisTemplate redisTemplate;

    @Test
    void whenOrderPublished_thenShipmentCreatedAndMessageAcked() {
        publisher.publishOrderCreated("1001", 250_000);

        await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> {
            assertThat(shippingService.shippedOrders()).contains("1001");

            PendingMessagesSummary summary = redisTemplate.opsForStream()
                    .pending("orders", "shipping-group");
            assertThat(summary.getTotalPendingMessages()).isZero();
        });
    }
}

@ServiceConnection(name = "redis") により、Spring Bootはcontainerから自動的にhostとportを取得します。@DynamicPropertySource は必要ありません。await() は必要です。listenerは別のthreadで実行されるためです。assertion getTotalPendingMessages() がゼロであることは、メッセージが読み込まれただけでなくACKされたことを証明しています。

9. よくあるエラーと使用すべきでない場合

べき等性のない処理。 At-least-once は、1つのメッセージが2回到達する可能性があることを意味します。例えば：consumerが処理を完了しましたが、ACKする前にcrashします。したがって、ビジネスロジックは反復に耐えられる必要があります。一般的なアプローチは、orderId が既に処理されているかどうかを確認するか、databaseにunique制約を使用することです。

Stream のサイズが無限に拡大する。 ACKはstream からentryを削除しません。書き込み時にサイズを制限する必要があります：

redisTemplate.opsForStream().add(record, XAddOptions.maxlen(100_000).approximateTrimming(true));

このオプションは XADD orders MAXLEN ~ 100000 * ... に対応しています。~ 記号を使用すると、Redisは内部ノードに従ってトリミングでき、正確なトリミングよりもはるかに高速です。

cancelOnError を忘れる。 最初の例外は consumerを暗黙的に停止させます。症状は、stream にメッセージがたまっているのに、ログに明確なエラーがないというものです。

“ゴースト”コンシューマー。 新しい名前でデプロイするたびに（例：pod名）、古い名前はgroupに残ります。XGROUP DELCONSUMER で定期的にクリーンアップしてください。ただし、そのconsumerのすべての保留メッセージをclaimした後に限ります。

Redis Streams を使用すべきでない場合：

数日間イベントを保存して、大量にreplayする必要があります。RedisはすべてのデータをRAMに保持するため、コストが非常に高くなります。
絶対にデータを失わないことを保証する必要があります。AOF everysec またはasyncフェイルオーバーレプリカでは、書き込まれたばかりのメッセージが失われる可能性があります。
スループットが非常に大きく、数百のconsumerでパーティション化とスケーリングが必要です。Kafkaはこの場合により適しています。

10. まとめ

Redis Streamsとconsumer groupを組み合わせることで、既に持っているRedis上で実行される、軽量で信頼性の高いイベントキューが提供されます。4つの重要なポイントを覚えておく必要があります：

処理完了後に手動でACKします。
cancelOnError(t -> false) を設定してconsumerが暗黙的に停止しないようにします。
保留メッセージを回収するjobを持ち、poison messageのためのdead-letterストリームを含めます。
処理ロジックはべき等であることを書きます。

MAXLEN を追加してメモリを制御し、Testcontainersで統合テストを行うことで、ほとんどのSpring Bootアプリケーションに十分なat-least-once イベント処理パイプラインが完成します。

Redis StreamsとConsumer Group：信頼性の高いイベント処理

1. 概要

2. Redis Streams と Consumer Group のしくみ

3. セットアップ

4. ストリームにイベントを書き込む

5. 安全に consumer group を作成する

6. StreamMessageListenerContainer でメッセージを読み込む

6.1. 手動 ACK を使用する Listener

6.2. Container の設定

7. 保留メッセージと dead-letter の処理

8. Testcontainers でテストを書く

9. よくあるエラーと使用すべきでない場合

10. まとめ

同じトピックの記事

Spring BootのRedisで分散ロックを実装する

Redis の Hot Key と Big Key: 検出と処理

Redis 永続化：RDB、AOF とデータ損失

Redis：キャッシュアサイド、ライトスルー、ライトビハインドの選択

Redisでキャッシュとデータベースの一貫性を保つ