Pub/SubとRedis Streams：違いと使い分け方

1. 概要

Redisには2つの異なるメッセージ配信メカニズムがあります：Pub/SubとStreams。Redisを初めて使う多くの開発者は、これら2つを同じものと勘違いすることが多く、その結果、間違ったツールを選んでしまい、システム再起動時にデータを失ってしまうことがあります。

このガイドでは、Pub/SubとStreamsの本質的な違いを説明し、これら2つを使用するJavaコード例を示します。さらに、どの状況でどちらを使用すべきかを明確に説明します。読み終わったら、あなたの実装で適切なモデルを自信を持って選択できるようになります。

2. Redis Pub/Subとは

Pub/Sub (Publish/Subscribe) は「ファイアアンドフォーゲット」モデルで動作します。パブリッシャーはメッセージをチャネルに送信し、Redisはそのチャネルをリッスンしている全てのサブスクライバーに即座に転送します。オンラインのサブスクライバーがない場合、メッセージは永遠に消えてしまいます。

2.1. Maven依存関係

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

2.2. パブリッシャー

@Service
public class OrderEventPublisher {

    private final StringRedisTemplate redisTemplate;

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

    public void publish(String orderId, String status) {
        String message = "orderId=%s;status=%s".formatted(orderId, status);
        redisTemplate.convertAndSend("order-events", message);
    }
}

convertAndSendはorder-eventsチャネルにメッセージを送信します。まだサブスクライバーがいない場合でも、このコマンドは成功しますが、メッセージはどこにも保存されません。

2.3. サブスクライバー

@Component
public class OrderEventListener implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String body = new String(message.getBody());
        System.out.println("イベント受信: " + body);
    }
}

@Configuration
public class RedisPubSubConfig {

    @Bean
    public RedisMessageListenerContainer container(
            RedisConnectionFactory connectionFactory,
            OrderEventListener listener) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(listener, new ChannelTopic("order-events"));
        return container;
    }
}

RedisMessageListenerContainerは、order-eventsチャネルをリッスンするための専用接続（ブロッキング接続）を管理します。パブリッシャーがメッセージを送信すると、すぐにonMessageが呼び出されます。

イベント発行時の出力：

イベント受信: orderId=ORD-001;status=CONFIRMED

注意： OrderEventListenerがまだ起動していない場合（containerがまだバインドしていない場合）、その前に発行されたメッセージは完全に失われ、リプレイする方法はありません。

3. Redis Streamsとは

Streamは順序が保証されたログ構造のデータで、追加のみ可能で、各メッセージは<timestamp>-<sequence>形式の一意のIDを持ちます。Pub/Subと異なり、StreamsはメッセージをRedisに保存し、あなたが明示的に削除（XDEL）またはトリム（XTRIM）するまで保持します。コンシューマーはメッセージを読み直したり、Kafkaのようなコンシューマーグループで確認応答機能を使って読むことができます。

3.1. Streamにメッセージを追加

@Service
public class OrderStreamPublisher {

    private final StringRedisTemplate redisTemplate;

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

    public RecordId publish(String orderId, String status) {
        Map<String, String> payload = Map.of(
                "orderId", orderId,
                "status", status
        );
        return redisTemplate.opsForStream()
                .add("order-stream", payload);
    }
}

opsForStream().addはorder-streamストリームにレコードを書き込み、RecordIdを返します。これはそのレコードの自動生成される主キーのようなものです。このレコードはトリムまたは削除されるまでRedisに存在し、誰も読んでいなくても保持されます。

3.2. コンシューマーグループを使ったStreamの読み取り

@Component
public class OrderStreamConsumer {

    private final StringRedisTemplate redisTemplate;

    public OrderStreamConsumer(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        createGroupIfAbsent();
    }

    private void createGroupIfAbsent() {
        try {
            redisTemplate.opsForStream()
                    .createGroup("order-stream", "order-processors");
        } catch (Exception e) {
            // グループは既に存在するため、無視
        }
    }

    public void pollOnce() {
        List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream()
                .read(Consumer.from("order-processors", "worker-1"),
                        StreamReadOptions.empty().count(10),
                        StreamOffset.create("order-stream", ReadOffset.lastConsumed()));

        for (MapRecord<String, Object, Object> record : records) {
            System.out.println("レコード " + record.getId() + " を処理中: " + record.getValue());
            redisTemplate.opsForStream().acknowledge("order-stream", "order-processors", record.getId());
        }
    }
}

createGroupはコンシューマーグループorder-processorsを作成します。これはKafkaのコンシューマーグループに似ています。
Consumer.from("order-processors", "worker-1")は、このグループに属するコンシューマーworker-1を宣言します。同じグループの複数のワーカーはレコードを共有し、重複して読まないようにします。
ReadOffset.lastConsumed()は、そのコンシューマーグループ内でまだ配信されたことがない新しいレコードを読むことを意味します。
acknowledgeは処理の完了を確認し、Redisはレコードをグループの「保留リスト」から削除します。

出力：

レコード 1696500000000-0 を処理中: {orderId=ORD-001, status=CONFIRMED}

注意： acknowledgeを呼び出し忘れた場合、レコードは保留リストに残ります。同じグループの別のコンシューマーまたは再起動後に同じコンシューマーがXCLAIM/readPendingを使ってそのレコードを再度取得し、処理することができます。これはPub/Subにはない「少なくとも1回配信」メカニズムです。

4. Pub/SubとStreamsの直接比較

項目	Pub/Sub	Streams
メッセージ保存	なし	あり（トリム/削除まで）
古いメッセージの再読み込み	不可	可能（`XRANGE`、オフセット指定）
オフラインコンシューマーが失うもの	すべてのメッセージ	なし（オンライン時に読み直し可）
複数コンシューマーのロードバランシング	各サブスクライバーが独立したコピーを受信（ブロードキャスト）	コンシューマーグループが分担、各レコードは1つのコンシューマーで処理
確認応答/リトライ	なし	あり（`XACK`、`XCLAIM`）
複雑性	非常にシンプル	より複雑、Kafkaに似ている
適切なユースケース	リアルタイム通知、配信保証不要	イベントソーシング、ジョブキュー、配信保証必須

4.1. Pub/Subを使うべき時

メッセージを即座にブロードキャストする必要があり、リスナーがいない場合のメッセージ喪失を許容できる場合はPub/Subを使用します。典型的な例としては、チャットアプリケーションで「入力中…」ステータスの更新、複数のアプリケーションインスタンス間でのキャッシュ無効化、または見逃しても重大ではない重要度が低いリアルタイム通知が挙げられます。

@Service
public class CacheInvalidationPublisher {

    private final StringRedisTemplate redisTemplate;

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

    public void invalidate(String cacheKey) {
        redisTemplate.convertAndSend("cache-invalidation", cacheKey);
    }
}

各アプリケーションインスタンスがcache-invalidationチャネルを購読し、削除すべきキャッシュキーを受信します。あるインスタンスが再起動中にイベントを見逃したとしても、その間は古いキャッシュが使用されるだけで、重大なエラーではありません。

4.2. Streamsを使うべき時

メッセージが重要なビジネスデータである場合は、Streamsを使用します。例えば、注文、決済トランザクション、監査ログなどです。コンシューマーが処理途中でクラッシュしても、再起動後にそのレコードを再度処理できることを保証する必要があります。

@Service
public class PaymentEventPublisher {

    private final StringRedisTemplate redisTemplate;

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

    public RecordId publish(String paymentId, BigDecimal amount) {
        Map<String, String> payload = Map.of(
                "paymentId", paymentId,
                "amount", amount.toString()
        );
        return redisTemplate.opsForStream().add("payment-stream", payload);
    }
}

各決済イベントはpayment-streamに記録されます。PaymentWorkerがデプロイのために停止していても、このイベントはRedisに待機中の状態で残り、ワーカーが再起動すると処理を継続します。

5. Stream Consumerのテストを書く

Streamsは永続的な状態を持つため、publish-consume-acknowledgeのフロー全体をテストする必要があります。実際のRedisを実行するTestcontainersを使用して、挙動の誤ったモック化を回避します。

<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>junit-jupiter</artifactId>
    <scope>test</scope>
</dependency>

@Testcontainers
class OrderStreamConsumerTest {

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

    StringRedisTemplate redisTemplate;

    @BeforeEach
    void setUp() {
        RedisStandaloneConfiguration config = new RedisStandaloneConfiguration(
                redis.getHost(), redis.getFirstMappedPort());
        LettuceConnectionFactory factory = new LettuceConnectionFactory(config);
        factory.afterPropertiesSet();
        redisTemplate = new StringRedisTemplate(factory);
    }

    @Test
    void shouldConsumeAndAcknowledgeRecord() {
        OrderStreamPublisher publisher = new OrderStreamPublisher(redisTemplate);
        OrderStreamConsumer consumer = new OrderStreamConsumer(redisTemplate);

        publisher.publish("ORD-002", "SHIPPED");
        consumer.pollOnce();

        PendingMessages pending = redisTemplate.opsForStream()
                .pending("order-stream", Consumer.from("order-processors", "worker-1"),
                        Range.unbounded());

        assertThat(pending.getTotalPendingMessages()).isZero();
    }
}

このテストは1つのレコードを発行し、pollOnce()を呼び出してコンシュームして確認応答し、その後保留リストが空であることを確認します。これはレコードが処理完了として確認されたことを意味し、グループ内に留まっていません。

6. よくあるエラー

重要なデータにPub/Subを使う：Pub/Subを使って注文イベントを送信したのに、コンシューマー再起動時にイベントが失われて驚く人が多いです。Pub/Subはメッセージキューではありません。
Streamを定期的にトリムしない：Streamは古いデータを自動で削除しません。継続的にデータを発行しながらXTRIMを呼び出さない、またはadd時にMAXLENを設定しないと、Redisのメモリはどんどん増えていきます。

redisTemplate.opsForStream().add(
        StreamRecords.newRecord()
                .in("order-stream")
                .ofMap(payload),
        RedisStreamCommands.XAddOptions.maxlen(10_000)
);

コンシューマーグループを作成してから読む：XREADGROUPコマンドはグループが存在しない場合エラーになります。常にcreateGroupをtry-catchで呼び出して「BUSYGROUP」エラー（グループが既に存在）を捕捉してください。
Pub/Subのコネクションタイムアウトを設定しない：サブスクライバーは専用のブロッキング接続を使用します。ネットワークが切断されても再接続メカニズムがない場合、その時点からイベント検出と再接続まで全てのメッセージを失います。

7. まとめ

Pub/SubとStreamsはRedis内にあり、どちらもメッセージ配信に使われますが、2つの全く異なる問題を解決します。Pub/Subはシンプルで高速ですが、何も保証しません。リアルタイム通知でメッセージ喪失を許容できる場合に適しています。Streamsはより複雑ですが堅牢で、コンシューマーグループと確認応答機能を備えており、Kafkaのような軽量メッセージキューで、ビジネスデータが完全に処理されることを保証する必要がある場合に適しています。

もしシステムがさらに高い信頼性を必要とする場合（exactly-once配信、大規模パーティション分割、大量の並列コンシューマーグループなど）は、Redis Streamsを無理に使い続けるのではなく、KafkaやRabbitMQの採用を検討してください。