Pub/Sub et Redis Streams : Différences et Utilisation avec Spring Boot

1. Vue d’ensemble

Redis fournit deux mécanismes de transmission de messages différents : Pub/Sub et Streams. Beaucoup de développeurs débutants avec Redis confondent ces deux systèmes, ce qui les conduit à choisir le mauvais outil et à perdre des données lors du redémarrage du système.

Cet article explique les différences essentielles entre Pub/Sub et Streams, vous fournit du code Java fonctionnant avec les deux, et vous dit clairement quand utiliser chacun. Après cette lecture, vous saurez avec confiance quel modèle choisir pour votre vrai problème de messagerie.

2. Qu’est-ce que Redis Pub/Sub

Pub/Sub (Publish/Subscribe) fonctionne selon un modèle « fire-and-forget ». Un publisher envoie un message sur un canal, et Redis le transmet immédiatement à tous les subscribers écoutant ce canal. S’il n’y a aucun subscriber en ligne, le message disparaît définitivement.

2.1. Dépendance Maven

<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 envoie un message sur le canal order-events. S’il n’y a encore personne d’abonné, cette commande réussit toujours mais le message n’est sauvegardé nulle part.

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("Évènement reçu : " + 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 gère une connexion dédiée (utilisant une connexion de blocage) pour écouter le canal order-events. Lorsqu’un publisher envoie un message, onMessage est appelé immédiatement.

Sortie lors de la publication d’un événement :

Évènement reçu : orderId=ORD-001;status=CONFIRMED

Remarque : Si OrderEventListener n’a pas fini de démarrer (le conteneur n’est pas encore lié), les messages publiés avant sont complètement perdus, il n’y a aucun moyen de les rejouer.

3. Qu’est-ce que Redis Streams

Stream est une structure de données de journal ordonnée, en ajout seul, où chaque message a un ID unique de la forme <timestamp>-<sequence>. Contrairement à Pub/Sub, Streams conserve les messages dans Redis jusqu’à ce que vous les supprimiez explicitement (XDEL) ou les tronquiez (XTRIM). Un consumer peut relire depuis le début, lire par groupe (consumer group) avec un mécanisme d’accusé de réception similaire à Kafka.

3.1. Ajouter un message à un 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 écrit un enregistrement dans le stream order-stream et retourne un RecordId — comme une clé primaire auto-générée de cet enregistrement. Cet enregistrement existe dans Redis jusqu’à ce qu’il soit tronqué ou supprimé, que quelqu’un l’ait lu ou non.

3.2. Lire un Stream avec Consumer Group

@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) {
            // Le groupe existe déjà, ignorer
        }
    }

    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("Traitement de l'enregistrement " + record.getId() + ": " + record.getValue());
            redisTemplate.opsForStream().acknowledge("order-stream", "order-processors", record.getId());
        }
    }
}

• createGroup crée un consumer group order-processors, similaire à un consumer group dans Kafka.
• Consumer.from("order-processors", "worker-1") déclare un consumer nommé worker-1 appartenant à ce groupe. Plusieurs workers du même groupe se partageront les enregistrements, sans duplication de lecture.
• ReadOffset.lastConsumed() signifie lire les nouveaux enregistrements qui n’ont jamais été distribués à aucun consumer du groupe.
• acknowledge confirme que le traitement est terminé, Redis supprime l’enregistrement de la « pending list » du groupe.

Sortie :

Traitement de l'enregistrement 1696500000000-0: {orderId=ORD-001, status=CONFIRMED}

Remarque : Si vous oubliez d’appeler acknowledge, l’enregistrement reste dans la pending list. Un autre consumer ou le même consumer (après redémarrage) peut utiliser XCLAIM/readPending pour reprendre le traitement — c’est le mécanisme « at-least-once delivery » que Pub/Sub n’a pas.

4. Comparaison directe Pub/Sub et Streams

Critère	Pub/Sub	Streams
Stockage des messages	Non	Oui, jusqu’à truncation/suppression
Rejeu des anciens messages	Non	Oui (XRANGE, offset personnalisable)
Perte lors de l’offline du consumer	Tous les messages perdus	Aucune perte, relecture possible au retour en ligne
Équilibrage de charge avec plusieurs consumers	Chaque subscriber reçoit sa propre copie (broadcast)	Consumer group répartit le travail, un seul consumer par enregistrement
Accusé de réception / retry	Aucun	Oui (XACK, XCLAIM)
Complexité	Très simple	Plus complexe, proche de Kafka
Cas d’usage approprié	Notifications en temps réel, pas de garantie	Event sourcing, job queue, besoin de garanties de traitement

4.1. Quand utiliser Pub/Sub

Utilisez Pub/Sub quand vous avez besoin d’un broadcast instantané et que vous acceptez la perte de messages si personne n’écoute. Les cas typiques : mise à jour du statut « en train d’écrire… » dans le chat, synchronisation du cache entre plusieurs instances d’une application (invalidation de cache), ou notifications en temps réel qui ne sont pas critiques si quelques-unes sont manquées.

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

Chaque instance de l’application s’abonne au canal cache-invalidation et reçoit les clés à supprimer du cache local. Si une instance redémarre et manque un événement, elle utilisera simplement l’ancien cache pendant un court moment — ce n’est pas une erreur grave.

4.2. Quand utiliser Streams

Utilisez Streams quand les messages sont des données métier critiques qui doivent être traitées complètement : commandes, transactions de paiement, journaux d’audit. Vous devez garantir que si un consumer s’arrête au milieu du traitement, l’enregistrement sera retraité après son redémarrage.

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

Chaque événement de paiement est écrit dans payment-stream. Même si PaymentWorker est arrêté pour un déploiement, cet événement attend dans Redis, et le worker le lira une fois redémarré.

5. Écrire des tests pour Stream Consumer

Comme Streams a un état persistant, vous devez bien tester le flux de publication – consommation – accusé de réception. Utilisez Testcontainers pour exécuter un vrai Redis, en évitant de mocker les comportements incorrectement.

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

Ce test publie un enregistrement, appelle pollOnce() pour le consommer et l’accuser de réception, puis vérifie que la pending list est vide — ce qui signifie que l’enregistrement a été confirmé comme traité et ne reste plus suspendu dans le groupe.

6. Erreurs courantes

• Utiliser Pub/Sub pour les données critiques: beaucoup de gens novices utilisent Pub/Sub pour envoyer des événements de commande, puis sont surpris quand un redémarrage du consumer fait perdre les événements. Pub/Sub n’est pas une message queue.
• Ne pas tronquer le Stream: un stream ne supprime pas automatiquement les anciennes données. Si vous publiez continuellement sans appeler XTRIM ou définir MAXLEN lors de l’ajout, la mémoire Redis grandir avec le temps.

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

• Oublier de créer le consumer group avant la lecture: la commande XREADGROUP échoue si le groupe n’existe pas. Appelez toujours createGroup avec un try-catch pour attraper l’erreur “BUSYGROUP” (le groupe existe déjà).
• Ne pas définir le timeout de connexion pour Pub/Sub: un subscriber utilise une connexion de blocage dédiée ; si le réseau s’interrompt et qu’il n’y a pas de mécanisme de reconnexion, vous perdez tous les messages à partir de ce moment jusqu’à ce que vous détectiez et reconnectiez.

7. Conclusion

Pub/Sub et Streams résolvent deux problèmes différents malgré le fait qu’ils vivent tous les deux dans Redis. Pub/Sub est simple et rapide, mais ne garantit rien — il convient aux notifications instantanées qui peuvent être manquées. Streams est plus complexe mais robuste, avec consumer groups et accusés de réception — il convient aux données métier qui doivent être traitées complètement, comme une file d’attente de messages légère.

Si votre système a besoin d’une fiabilité plus élevée (exactly-once, partitionnement important, plusieurs consumer groups en parallèle à grande échelle), envisagez plutôt Kafka ou RabbitMQ au lieu de forcer Redis Streams à faire trop de choses.