Redis Streams mit Consumer Groups: zuverlässige Ereignisverarbeitung

1. Übersicht

Das Pub/Sub von Redis hat einen großen Nachteil. Wenn ein Verbraucher ausgeschaltet ist, wenn eine Nachricht veröffentlicht wird, geht die Nachricht verloren. Redis Streams löst dieses Problem. Nachrichten werden dauerhaft im Stream gespeichert, und jede Verbrauchergruppe merkt sich, welche Nachrichten bereits verarbeitet wurden.

Dieser Artikel führt Sie durch den Aufbau eines Ereignisverarbeitungsablaufs für Bestellungen mit Spring Boot 3 und Spring Data Redis. Der Ablauf umfasst folgende Schritte: Ereignisse schreiben, über eine Verbrauchergruppe lesen, manuelle ACK, Wiederholung von hängenden Nachrichten und Verschieben von Fehlernachrichten in einen Dead-Letter-Stream.

2. Wie Redis Streams und Verbrauchergruppen funktionieren

Bevor Sie Java schreiben, sollten Sie die nativen Befehle in redis-cli ausprobieren:

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

Bedeutung jedes Befehls:

• XADD fügt einen Eintrag zum Stream orders hinzu. Das * lässt Redis die ID automatisch im Format <millis>-<seq> generieren.
• XGROUP CREATE ... 0 erstellt die Gruppe shipping-group und beginnt mit dem Lesen vom Anfang des Streams. Bei Verwendung von $ liest die Gruppe nur neue Nachrichten.
• XREADGROUP ... > ruft Nachrichten ab, die noch niemandem zugewiesen wurden. Die Nachricht wird worker-1 zugewiesen und gleichzeitig in die Pending Entries List (PEL) der Gruppe geschrieben.
• XPENDING zeigt an, wie viele Nachrichten zugewiesen, aber noch nicht bestätigt wurden.
• XACK entfernt die Nachricht aus der PEL. Ab diesem Zeitpunkt betrachtet die Gruppe die Nachricht als verarbeitet.

Die Ausgabe von XPENDING vor ACK sieht ungefähr so aus:

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

Die PEL ist der Schlüssel zur Zuverlässigkeit. Angenommen, ein Verbraucher stürzt ab, nachdem er eine Nachricht empfangen hat, aber bevor er sie bestätigt. Die Nachricht bleibt in der PEL, und ein anderer Verbraucher kann sie in Anspruch nehmen zur weiteren Verarbeitung. Dies ist die at-least-once-Semantik.

3. Setup

Fügen Sie die Abhängigkeit Spring Data Redis hinzu. Dieses Starter verwendet standardmäßig Lettuce als Client:

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

Konfigurieren Sie die Verbindung und den Stream-Namen:

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

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

Der Verbrauchername wird aus HOSTNAME bezogen, sodass jede Instanz (Pod) einen eindeutigen Namen hat. Das ist wichtig. Wenn zwei Instanzen denselben Namen verwenden, betrachtet Redis sie als einen Verbraucher, und ihre PELs vermischen sich.

4. Ereignisse in den Stream schreiben

@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(...) erstellt einen MapRecord mit Feldern als Strings. Der Befehl opsForStream().add() entspricht XADD orders * ... und gibt die von Redis generierte RecordId zurück, z.B. 1728640000000-0.

Die Verwendung von StringRedisTemplate macht Daten in Redis mit redis-cli lesbar. Sie müssen sich nicht mit Serialisierungs-Byte-Strings herumschlagen.

5. Sicher eine Verbrauchergruppe erstellen

Die Gruppe muss existieren, bevor Sie Nachrichten lesen. Die Anwendung kann starten, wenn der Stream noch keine Nachrichten enthält. Daher benötigen Sie den Parameter MKSTREAM. Die createGroup()-Methode von StreamOperations aktiviert diesen Parameter nicht. Deshalb rufen wir die Connection-Ebene auf:

@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;
            }
            // Gruppe existiert bereits: normal beim Neustart der Anwendung
        }
    }
}

Der letzte Parameter true aktiviert MKSTREAM. Redis erstellt einen leeren Stream, wenn dieser noch nicht existiert. Wenn Sie die Anwendung neu starten, gibt Redis den Fehler BUSYGROUP Consumer Group name already exists zurück. Wir ignorieren diesen bestimmten Fehler und werfen alle anderen Fehler erneut.

6. Nachrichten mit StreamMessageListenerContainer lesen

6.1. Listener mit manuellem ACK

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

Der ACK-Befehl wird nur ausgeführt, wenn createShipment erfolgreich ist. Wenn die Methode eine Exception auslöst, bleibt die Nachricht in der PEL zur späteren Verarbeitung. Bestätigen Sie nicht, bevor Sie verarbeiten. Dies würde at-least-once in at-most-once verwandeln.

Der ShippingService in diesem Artikel ist ein einfacher Service. Er zeichnet orderId auf, für die Versandetiketten erstellt wurden:

@Service
public class ShippingService {

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

    public void createShipment(String orderId) {
        if (!shippedOrders.add(orderId)) {
            return; // bereits verarbeitet, überspringen (idempotent)
        }
        // Kurier-API aufrufen...
    }

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

6.2. Container-Konfiguration

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

Die wichtigsten Punkte in der Konfiguration:

• ReadOffset.lastConsumed() entspricht > in XREADGROUP, d.h. nur neue Nachrichten der Gruppe abrufen.
• pollTimeout ist die Blockierungszeit pro Lesevorgang. batchSize ist die maximale Anzahl von Nachrichten pro Lesevorgang.
• autoAcknowledge(false) deaktiviert das automatische ACK, da wir im Listener selbst ACK durchführen.
• cancelOnError(t -> false) ist die wichtigste Zeile. Standardmäßig bricht der Container die Subscription ab, wenn der Listener eine Exception auslöst. Der Verbraucher wird stillschweigend beendet, ohne dass es eine klare Warnung gibt.

Hinweis: Verwenden Sie receiveAutoAck(...) nicht für kritische Geschäfte. Dieser Modus bestätigt sofort beim Lesen, also bevor der Listener fertig ist. Wenn die Anwendung abstürzt, geht die Nachricht für immer verloren.

7. Hängende Nachrichten und Dead-Letter verarbeiten

Der Listener liest nur neue Nachrichten (>). Eine fehlgeschlagene oder verwaiste Nachricht bleibt auf ewig in der PEL, wenn sich niemand darum kümmert. Wir brauchen einen regelmäßigen Job dafür:

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

Der Verarbeitungsablauf des Jobs:

1. pending(...) entspricht XPENDING orders shipping-group - + 50. Jede PendingMessage zeigt an, welcher Verbraucher die Nachricht hält, wie lange sie untätig ist und wie oft sie zugewiesen wurde.
2. claim(...) entspricht XCLAIM. Dieser Befehl überträgt das Eigentum an den aktuellen Verbraucher. Redis überträgt nur, wenn die Nachricht mindestens MIN_IDLE untätig war. Dadurch greifen zwei Instanzen nicht gleichzeitig zu. Jedes Claim erhöht auch den Zustellungszähler.
3. Nachrichten, die mehr als MAX_DELIVERIES Mal zugestellt wurden, werden in den Stream orders:dlq kopiert und dann bestätigt. Poison Messages werden daher nicht unbegrenzt wiederholt.

Wenn createShipment eine Exception auslöst, wird der ACK-Befehl nicht ausgeführt. Die Nachricht bleibt in der PEL und wird im nächsten Zyklus erneut versucht. Vergessen Sie nicht, @EnableScheduling auf einer @Configuration-Klasse zu aktivieren.

Hinweis: Ab Redis 6.2 gibt es den Befehl XAUTOCLAIM. Dieser Befehl kombiniert XPENDING und XCLAIM in einen und ist effizienter, wenn die PEL groß ist. Sie können ihn über redisTemplate.execute(...) aufrufen, wenn Ihre Version von Spring Data Redis noch keine praktische API hat.

8. Mit Testcontainers testen

Fügen Sie die Test-Abhängigkeiten hinzu:

<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>

Der Test unten veröffentlicht ein Ereignis. Anschließend prüft er, dass die Versandetikette erstellt wurde und die PEL leer ist:

@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") lässt Spring Boot automatisch Host und Port vom Container auslesen, sodass Sie @DynamicPropertySource nicht benötigen. await() ist notwendig, weil der Listener auf einem separaten Thread läuft. Die Assertion getTotalPendingMessages() gleich 0 beweist, dass die Nachricht bestätigt wurde, nicht nur gelesen.

9. Häufige Fehler und wann man Redis Streams nicht verwenden sollte

Nicht-idempotente Verarbeitung. At-least-once bedeutet, dass eine Nachricht zweimal ankommen kann. Beispiel: Der Verbraucher verarbeitet erfolgreich, stürzt aber vor dem ACK ab. Daher muss die Geschäftslogik mit Wiederholungen umgehen können. Der häufige Weg ist, zu prüfen, ob orderId bereits verarbeitet wurde, oder eine Unique Constraint in der Datenbank zu verwenden.

Stream wächst unbegrenzt. ACK löscht Einträge nicht aus dem Stream. Sie müssen die Größe beim Schreiben begrenzen:

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

Diese Option entspricht XADD orders MAXLEN ~ 100000 * .... Die Tilde ~ erlaubt Redis, nach internen Knoten zu trimmen, was viel schneller ist als exaktes Trimmen.

cancelOnError vergessen. Die erste Exception stoppt den Verbraucher stillschweigend. Das Symptom ist, dass Nachrichten sich im Stream stauen, aber die Logs zeigen keinen klaren Fehler.

Geister-Verbraucher. Jedes Mal, wenn Sie mit einem neuen Verbrauchernamen bereitstellen (z.B. neuer Pod-Name), bleibt der alte Name in der Gruppe. Bereinigen Sie regelmäßig mit XGROUP DELCONSUMER, aber nur nachdem Sie alle ausstehenden Nachrichten des Verbrauchers beansprucht haben.

Wann man Redis Streams NICHT verwenden sollte:

• Sie müssen Ereignisse Tage lang speichern und große Datenmengen wiedergeben. Redis speichert alles im RAM, daher sind die Kosten sehr hoch.
• Sie müssen garantieren, dass keine Daten verloren gehen. Mit AOF everysec oder asynchronem Failover kann immer noch etwas verloren gehen.
• Der Durchsatz ist sehr hoch und Sie benötigen hunderte Verbraucher mit Partitionierung. Kafka ist besser geeignet in diesem Fall.

10. Fazit

Redis Streams kombiniert mit Verbrauchergruppen geben Ihnen eine leichte und zuverlässige Ereigniswarteschlange, die auf dem Redis läuft, das Sie bereits haben. Sie müssen vier Hauptpunkte im Kopf behalten:

• Manuelles ACK nach erfolgreichem Verarbeiten.
• Setzen Sie cancelOnError(t -> false), damit der Verbraucher nicht stillschweigend stoppt.
• Führen Sie einen Job durch, der hängende Nachrichten zurückfordert, mit Dead-Letter-Stream für Poison Messages.
• Schreiben Sie die Verarbeitungslogik idempotent.

Mit MAXLEN zur Speicherkontrolle und Integrationstests mit Testcontainers haben Sie einen zuverlässigen at-least-once-Ereignisverarbeitungsablauf für die meisten Spring Boot Anwendungen.