Redis Streams avec consumer group : traiter les événements

1. Aperçu

Le Pub/Sub de Redis a un gros inconvénient. Si un consumer est arrêté au moment de la publication, le message est perdu. Redis Streams résout ce problème. Les messages sont stockés durablement dans le stream, et chaque consumer group se souvient des messages qui ont été traités.

Cet article vous montre comment construire un pipeline de traitement d’événements de commandes avec Spring Boot 3 et Spring Data Redis. Le pipeline comprend les étapes suivantes : écrire un événement, le lire via un consumer group, l’ACK manuellement, récupérer les messages en attente et transférer les messages d’erreur vers un stream de dead-letter.

2. Comment fonctionnent les Redis Streams et les consumer groups

Avant d’écrire du Java, vous devriez essayer les commandes natives dans 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

Voici ce que chaque commande fait :

• XADD ajoute une entrée au stream orders. L’astérisque * fait que Redis génère automatiquement un ID sous la forme <millis>-<seq>.
• XGROUP CREATE ... 0 crée le group shipping-group et commence à lire depuis le début du stream. Si vous utilisez $, le group lit seulement les nouveaux messages.
• XREADGROUP ... > récupère les messages qui n’ont pas encore été assignés au sein du group. Le message est attribué à worker-1 et enregistré dans la Pending Entries List (PEL) du group.
• XPENDING montre combien de messages ont été attribués mais n’ont pas encore été ACK.
• XACK supprime le message de la PEL. À partir de ce moment, le group considère le message comme traité.

La sortie de XPENDING avant l’ACK ressemble à ceci :

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

La PEL est la clé de la fiabilité. Supposons qu’un consumer plante après avoir reçu un message mais avant de l’ACK. Le message reste dans la PEL, et un autre consumer peut le réclamer (claim) pour le traiter. C’est la sémantique d’at-least-once.

3. Installation

Ajoutez la dépendance Spring Data Redis. Ce starter utilise Lettuce comme client par défaut :

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

Configurez la connexion et le nom du stream :

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

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

Le nom du consumer est tiré de HOSTNAME, donc chaque instance (pod) a un nom unique. C’est important. Si deux instances partagent le même nom, Redis les considère comme un seul consumer, et leur PEL se mélange.

4. Écrire un événement dans le stream

@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(...) crée un MapRecord avec des champs sous forme de chaîne de caractères. La commande opsForStream().add() est équivalente à XADD orders * ... et retourne un RecordId généré par Redis, par exemple 1728640000000-0.

L’utilisation de StringRedisTemplate rend les données dans Redis lisibles via redis-cli. Vous n’avez pas à vous soucier de la sérialisation en byte array du JDK.

5. Créer un consumer group de manière sécurisée

Le group doit exister avant la lecture. L’application peut démarrer lorsqu’aucun message n’existe encore dans le stream, vous devez donc utiliser le flag MKSTREAM. La méthode createGroup() de StreamOperations n’active pas ce flag. Nous devons donc appeler la couche connection directement :

@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;
            }
            // Le group existe déjà : normal au redémarrage de l'application
        }
    }
}

Le dernier paramètre true active le flag MKSTREAM. Redis créera un stream vide s’il n’existe pas. Lors du redémarrage de l’application, Redis retourne l’erreur BUSYGROUP Consumer Group name already exists. Nous ignorons cette erreur spécifique et relançons toute autre erreur.

6. Lire les messages avec StreamMessageListenerContainer

6.1. Listener avec ACK manuel

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

La commande ACK s’exécute seulement si createShipment réussit. Si la méthode lève une exception, le message reste dans la PEL pour être traité ultérieurement. Ne faites pas un ACK avant le traitement. Cela transformerait at-least-once en at-most-once.

ShippingService dans cet article est un service simple. Il garde une trace des orderId pour lesquels un expédition a été créée :

@Service
public class ShippingService {

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

    public void createShipment(String orderId) {
        if (!shippedOrders.add(orderId)) {
            return; // déjà traité, ignorer (idempotent)
        }
        // appeler l'API du transporteur...
    }

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

6.2. Configuration du 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;
    }
}

Les points clés de la configuration :

• ReadOffset.lastConsumed() correspond à > dans XREADGROUP, c’est-à-dire ne lire que les nouveaux messages du group.
• pollTimeout est le timeout de blocage pour chaque lecture. batchSize est le nombre maximal de messages à lire en une seule fois.
• autoAcknowledge(false) désactive l’ACK automatique, car nous faisons nous-mêmes l’ACK dans le listener.
• cancelOnError(t -> false) est la ligne la plus importante. Par défaut, lorsque le listener lève une exception, le container annule la souscription. Le consumer s’arrête silencieusement sans avertissement clair.

Note : N’utilisez pas receiveAutoAck(...) pour les opérations critiques. Ce mode ACK dès la lecture, c’est-à-dire avant que le listener ait terminé. Si l’application plante à mi-chemin, le message est perdu à jamais.

7. Gérer les messages en attente et le dead-letter

Le listener lit seulement les nouveaux messages (>). Un message d’erreur ou appartenant à un consumer mort restera indéfiniment dans la PEL si personne ne le traite. Nous avons besoin d’un job périodique pour cela :

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

Le flux de traitement du job :

1. pending(...) est équivalent à XPENDING orders shipping-group - + 50. Chaque PendingMessage indique quel consumer détient le message, depuis combien de temps il est inactif et combien de fois il a été remis.
2. claim(...) est équivalent à XCLAIM. Cette commande transfère la propriété au consumer actuel. Redis ne transfère que si le message est resté inactif au moins MIN_IDLE, ce qui évite deux instances d’exécuter le job simultanément de réclamer le même message. Chaque claim augmente aussi le delivery count.
3. Un message qui a été remis plus de MAX_DELIVERIES fois est copié vers le stream orders:dlq puis ACK. Un message d’erreur permanent (poison message) n’est donc pas réessayé indéfiniment.

Si createShipment lève une exception, la commande ACK ne s’exécute pas. Le message reste dans la PEL et sera réessayé au cycle suivant. N’oubliez pas d’activer @EnableScheduling sur une classe @Configuration.

Note : À partir de Redis 6.2, il existe la commande XAUTOCLAIM. Cette commande combine les étapes XPENDING et XCLAIM en une, et est plus efficace lorsque la PEL est grande. Vous pouvez l’appeler via redisTemplate.execute(...) si la version de Spring Data Redis que vous utilisez n’a pas encore d’API pratique pour cela.

8. Écrire des tests avec Testcontainers

Ajoutez les dépendances pour les tests :

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

Le test ci-dessous publie un événement. Ensuite, il vérifie que l’expédition a été créée et que la PEL est vide :

@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") laisse Spring Boot automatiquement récupérer le host et le port du container, vous n’avez donc pas besoin de @DynamicPropertySource. await() est nécessaire car le listener s’exécute sur un thread séparé. L’assertion getTotalPendingMessages() == 0 prouve que le message a été ACK et non pas seulement lu.

9. Erreurs courantes et quand ne pas utiliser

Traitement non-idempotent. At-least-once signifie qu’un message peut arriver deux fois. Par exemple : un consumer termine le traitement mais plante avant l’ACK. Par conséquent, la logique métier doit être capable de gérer les répétitions. Une approche courante est de vérifier si l’orderId a déjà été traité, ou d’utiliser une contrainte unique dans la base de données.

Le stream s’agrandit indéfiniment. ACK ne supprime pas l’entrée du stream. Vous devez limiter la taille lors de l’écriture :

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

Cette option est équivalente à XADD orders MAXLEN ~ 100000 * .... Le tilde ~ permet à Redis de tronquer par nœud interne, beaucoup plus rapide qu’une troncature exacte.

Oublier cancelOnError. La première exception arrêtera silencieusement le consumer. Le symptôme est que les messages s’accumulent dans le stream mais sans erreur claire dans les logs.

Consumer « fantôme ». Chaque fois que vous déployez avec un nouveau nom de consumer (par exemple, le nom du pod), l’ancien nom reste dans le group. Nettoyez régulièrement avec XGROUP DELCONSUMER, mais seulement après avoir réclamé tous les messages en attente de ce consumer.

Quand NE PAS utiliser Redis Streams :

• Vous avez besoin de stocker des événements pendant plusieurs jours et de rejouer un gros volume. Redis conserve toutes les données en RAM, donc le coût est très élevé.
• Vous avez besoin de garantir qu’aucune donnée n’est perdue absolument. Avec AOF everysec ou failover de réplicas asynchrones, il y a toujours un risque de perdre quelques messages fraîchement écrits.
• Le débit est très élevé, vous avez besoin de partitioning et de scaling sur des centaines de consumers. Kafka convient mieux dans ce cas.

10. Conclusion

Redis Streams combiné avec les consumer groups vous donne une queue d’événements légère et fiable, fonctionnant directement sur le Redis que vous avez déjà. Vous devez retenir quatre points clés :

• ACK manuel après le traitement.
• Définir cancelOnError(t -> false) pour que le consumer ne s’arrête pas silencieusement.
• Avoir un job qui reclame les messages en attente, avec un stream de dead-letter pour les poison messages.
• Écrire une logique de traitement idempotent.

Ajoutez MAXLEN pour contrôler la mémoire et testez l’intégration avec Testcontainers, vous aurez un pipeline de traitement d’événements at-least-once suffisamment robuste pour la plupart des applications Spring Boot.