Pub/Sub und Redis Streams: Unterschiede und Verwendung mit Spring Boot

1. Überblick

Redis bietet zwei unterschiedliche Mechanismen zur Nachrichtenübertragung: Pub/Sub und Streams. Viele neue Redis-Benutzer verwechseln diese beiden, wählen das falsche Tool und verlieren Daten bei einem Systemstart.

Dieser Artikel erklärt die grundlegenden Unterschiede zwischen Pub/Sub und Streams, gibt dir Java-Code für beide Mechanismen, und zeigt dir klar, welches Modell du wann verwenden solltest. Nach dem Lesen wirst du mit Zuversicht das richtige Modell für dein praktisches Messaging-Problem auswählen können.

2. Was ist Redis Pub/Sub?

Pub/Sub (Publish/Subscribe) arbeitet nach dem Fire-and-Forget-Modell. Ein Publisher sendet eine Nachricht an einen Channel, Redis leitet sie sofort an alle Subscriber weiter, die diesen Channel abhören. Falls kein Subscriber online ist, ist die Nachricht für immer verloren.

2.1. Maven-Abhängigkeit

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

2.2. Publisher

@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 sendet die Nachricht an den Channel order-events. Falls noch niemand den Channel abonniert hat, wird der Befehl trotzdem erfolgreich ausgeführt, aber die Nachricht wird nirgends gespeichert.

2.3. Subscriber

@Component
public class OrderEventListener implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String body = new String(message.getBody());
        System.out.println("Event empfangen: " + 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 verwaltet eine dedizierte Blocking-Verbindung, um den Channel order-events abzuhören. Wenn der Publisher eine Nachricht sendet, wird onMessage sofort aufgerufen.

Ausgabe beim Veröffentlichen eines Events:

Event empfangen: orderId=ORD-001;status=CONFIRMED

Hinweis: Falls OrderEventListener nicht rechtzeitig startet (Container nicht gebunden), sind alle Nachrichten, die davor veröffentlicht werden, vollständig verloren, es gibt keine Möglichkeit, sie wiederzugeben.

3. Was sind Redis Streams?

Ein Stream ist eine geordnete, append-only Log-Datenstruktur, wobei jede Nachricht eine eindeutige ID in der Form <timestamp>-<sequence> hat. Im Gegensatz zu Pub/Sub speichert Streams Nachrichten in Redis, bis du sie aktiv löschst (XDEL) oder kürzt (XTRIM). Consumer können von Anfang an lesen, in Gruppen (Consumer Groups) lesen und haben einen Kafka-ähnlichen Acknowledge-Mechanismus.

3.1. Nachricht in einen Stream schreiben

@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 schreibt einen Record in den Stream order-stream und gibt eine RecordId zurück — ähnlich einem autogenerierten Primärschlüssel für diesen Record. Dieser Record existiert in Redis, bis er gekürzt oder gelöscht wird, unabhängig davon, ob jemand ihn gelesen hat.

3.2. Stream mit Consumer Group lesen

@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) {
            // Gruppe existiert bereits, ignorieren
        }
    }

    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("Verarbeite Record " + record.getId() + ": " + record.getValue());
            redisTemplate.opsForStream().acknowledge("order-stream", "order-processors", record.getId());
        }
    }
}

• createGroup erstellt die Consumer Group order-processors, ähnlich wie eine Consumer Group in Kafka.
• Consumer.from("order-processors", "worker-1") deklariert einen Consumer namens worker-1 in dieser Gruppe. Mehrere Worker in der gleichen Gruppe teilen sich die Records auf, es gibt keine Duplikate.
• ReadOffset.lastConsumed() bedeutet, dass nur neue Records gelesen werden, die dieser Consumer-Gruppe noch nie zugewiesen wurden.
• acknowledge bestätigt die erfolgreiche Verarbeitung, Redis entfernt den Record aus der “pending list” der Gruppe.

Ausgabe:

Verarbeite Record 1696500000000-0: {orderId=ORD-001, status=CONFIRMED}

Hinweis: Wenn du vergisst, acknowledge aufzurufen, bleibt der Record in der Pending-Liste. Ein anderer Consumer oder dieser Consumer selbst (nach Neustart) können den Record mit XCLAIM/readPending abrufen und erneut verarbeiten — das ist der “At-Least-Once Delivery”-Mechanismus, den Pub/Sub nicht bietet.

4. Direkter Vergleich: Pub/Sub vs. Streams

Kriterium	Pub/Sub	Streams
Nachrichtenspeicherung	Nein	Ja, bis Trim/Löschen
Alte Nachrichten erneut abspielen	Nein	Ja (XRANGE, anpassbarer Offset)
Consumer offline verliert	Alle während Offline gesendeten Nachrichten	Nichts, lesbar nach Onlinegehen
Load-Balancing für mehrere Consumer	Jeder Subscriber erhält eine Kopie (Broadcast)	Consumer Group teilt auf, jeder Record 1 Consumer verarbeitet
Acknowledge / Retry	Nicht vorhanden	Ja (XACK, XCLAIM)
Komplexität	Sehr einfach	Komplexer, ähnlich Kafka
Verwendungsfälle	Echtzeit-Benachrichtigungen, keine Garantie erforderlich	Event Sourcing, Job Queue, vollständige Verarbeitung erforderlich

4.1. Wann du Pub/Sub verwenden solltest

Verwende Pub/Sub, wenn du sofortige Broadcasts brauchst und bereit bist, Nachrichten zu verlieren, die niemand abhört. Typische Beispiele: Tippstatus-Updates im Chat, Cache-Synchronisierung zwischen mehreren Anwendungsinstanzen (Cache-Invalidierung) oder Echtzeit-Benachrichtigungen, die nicht kritisch sind, wenn ein paar verloren gehen.

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

Jede Anwendungsinstanz abonniert den Channel cache-invalidation und erhält den Schlüssel, der aus dem lokalen Cache gelöscht werden soll. Wenn eine Instanz während des Neustarts ein Event übersieht, verwendet sie nur für kurze Zeit den alten Cache — das ist kein schwerwiegender Fehler.

4.2. Wann du Streams verwenden solltest

Verwende Streams, wenn Nachrichten wichtige Geschäftsdaten sind, die vollständig verarbeitet werden müssen: Bestellungen, Zahlungstransaktionen, Audit-Logs. Du musst sicherstellen, dass ein Consumer, der mittendrin abstürzt, den Record nach dem Neustart erneut verarbeiten kann.

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

Jedes Payment-Event wird in payment-stream geschrieben. Auch wenn PaymentWorker zum Bereitstellen heruntergefahren ist, wartet dieses Event immer noch in Redis, und der Worker wird nach dem Neustart weitermachen.

5. Tests für Stream Consumer schreiben

Da Streams persistent sind, solltest du den Publish-Consume-Acknowledge-Ablauf sorgfältig testen. Verwende Testcontainers, um echtes Redis auszuführen und Mocks zu vermeiden, die sich falsch verhalten.

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

Dieser Test veröffentlicht einen Record, ruft pollOnce() auf, um ihn zu verbrauchen und zu bestätigen, und überprüft dann, ob die Pending-Liste leer ist — das bedeutet, dass der Record vollständig verarbeitet und bestätigt wurde und nicht mehr in der Gruppe hängt.

6. Häufige Fehler

• Pub/Sub für wichtige Daten verwenden: Viele Anfänger verwenden Pub/Sub, um Order-Events zu senden, und sind dann überrascht, wenn Nachrichten nach einem Consumer-Neustart verloren gehen. Pub/Sub ist keine Message Queue.
• Stream nicht kürzen: Streams löschen alte Daten nicht automatisch. Wenn du kontinuierlich veröffentlichst, ohne XTRIM aufzurufen oder MAXLEN beim add zu setzen, wird der Redis-Speicher mit der Zeit aufgebläht.

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

• Consumer Group vor dem Lesen nicht erstellen: Der XREADGROUP-Befehl schlägt fehl, wenn die Gruppe nicht existiert. Rufe immer createGroup mit try-catch auf und fange den Fehler “BUSYGROUP” auf (Gruppe existiert bereits).
• Keine Connection Timeout für Pub/Sub setzen: Subscriber verwenden eine eigene Blocking-Verbindung; wenn das Netzwerk unterbrochen wird und es keinen Reconnect-Mechanismus gibt, verlierst du alle Nachrichten von da an, bis sie erkannt und erneut verbunden werden.

7. Fazit

Pub/Sub und Streams lösen zwei unterschiedliche Probleme, obwohl beide in Redis enthalten sind. Pub/Sub ist einfach und schnell, bietet aber keine Garantien — geeignet für Echtzeit-Benachrichtigungen, die übersehen werden können. Streams sind komplexer, aber robust, haben Consumer Groups und Acknowledge-Mechanismen — geeignet für Geschäftsdaten, die vollständig verarbeitet werden müssen, ähnlich einer leichten Message Queue.

Wenn dein System eine noch höhere Zuverlässigkeit benötigt (Exactly-Once, große Partitionierung, viele Consumer Groups parallel in großem Maßstab), solltest du Kafka oder RabbitMQ in Betracht ziehen, anstatt Redis Streams zu überfordern.