Pub/Sub y Redis Streams: Diferencias y Cómo usarlos

1. Descripción general

Redis proporciona dos mecanismos diferentes para transmitir mensajes: Pub/Sub y Streams. Muchas personas nuevas con Redis suelen confundir estos dos, lo que lleva a elegir la herramienta incorrecta y perder datos cuando el sistema se reinicia.

Este artículo explica las diferencias fundamentales entre Pub/Sub y Streams, te proporciono código Java funcional para ambos, y dejo clara cuándo usar cada uno. Después de leerlo, confiarás en elegir el modelo correcto para tu problema real de mensajería.

2. ¿Qué es Redis Pub/Sub?

Pub/Sub (Publish/Subscribe) funciona con un modelo de “disparar y olvidar” (fire-and-forget). El publicador envía un mensaje a un canal, y Redis lo retransmite inmediatamente a todos los suscriptores que escuchan ese canal. Si no hay nadie suscrito online, el mensaje desaparece para siempre.

2.1. Dependencia Maven

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

2.2. Publicador

@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 envía un mensaje al canal order-events. Si aún no hay nadie suscrito, este comando aún tiene éxito pero el mensaje no se almacena en ningún lugar.

2.3. Suscriptor

@Component
public class OrderEventListener implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String body = new String(message.getBody());
        System.out.println("Recibe evento: " + 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 gestiona una conexión privada (usa una conexión de bloqueo) para escuchar el canal order-events. Cuando el publicador envía un mensaje, onMessage se llama inmediatamente.

Salida al publicar un evento:

Recibe evento: orderId=ORD-001;status=CONFIRMED

Nota: Si OrderEventListener aún no ha terminado de iniciarse (el contenedor aún no está vinculado), el mensaje publicado antes se pierde por completo, no hay forma de reproducirlo.

3. ¿Qué es Redis Streams?

Stream es una estructura de datos de registro ordenada, solo para agregar, donde cada mensaje tiene un ID único con formato <timestamp>-<sequence>. A diferencia de Pub/Sub, Streams almacena mensajes en Redis hasta que los elimines activamente (XDEL) o los recortes (XTRIM). Los consumidores pueden leer desde el principio, leer por grupos (consumer group) con un mecanismo de reconocimiento similar al de Kafka.

3.1. Poner mensajes en 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 escribe un registro en el stream order-stream y devuelve RecordId — como una clave primaria generada automáticamente de ese registro. Este registro existe en Redis hasta que sea recortado o eliminado, incluso si nadie lo ha leído aún.

3.2. Leer Stream usando 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) {
            // El grupo ya existe, ignora
        }
    }

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

• createGroup crea un grupo de consumidores order-processors, similar a un grupo de consumidores en Kafka.
• Consumer.from("order-processors", "worker-1") declara un consumidor llamado worker-1 que pertenece a ese grupo. Varios workers del mismo grupo se dividirán los registros, sin leer duplicados.
• ReadOffset.lastConsumed() significa leer registros nuevos que aún no se han entregado a ningún consumidor en el grupo.
• acknowledge confirma que ha terminado el procesamiento, Redis elimina el registro de la “pending list” del grupo.

Salida:

Procesa registro 1696500000000-0: {orderId=ORD-001, status=CONFIRMED}

Nota: Si olvidas llamar a acknowledge, el registro sigue en la pending list. Otro consumidor o el mismo consumidor (después de reiniciar) puede usar XCLAIM/readPending para retirarlo y procesarlo nuevamente — este es el mecanismo de “at-least-once delivery” que Pub/Sub no tiene.

4. Comparación directa entre Pub/Sub y Streams

Criterio	Pub/Sub	Streams
Almacenamiento de mensajes	No	Sí, hasta que se recorten/eliminen
Reproducir mensajes antiguos	No	Sí (XRANGE, offset personalizable)
Consumer sin conexión pierde qué	Todos los mensajes enviados sin conexión	Nada, lee de nuevo al conectarse
Equilibrio de carga de múltiples consumidores	Cada suscriptor recibe una copia (broadcast)	El grupo de consumidores comparte el trabajo, cada registro es procesado por 1 consumidor
Reconocimiento / reintentos	No	Sí (XACK, XCLAIM)
Complejidad	Muy simple	Más complejo, similar a Kafka
Caso de uso adecuado	Notificaciones en tiempo real, no requiere garantía	Event sourcing, job queue, requiere garantía de procesamiento

4.1. Cuándo usar Pub/Sub

Usa Pub/Sub cuando necesitas un broadcast inmediato y aceptas perder mensajes si nadie está escuchando. Ejemplos típicos: actualizar estado de “escribiendo…” en chat, sincronizar caché entre múltiples instancias de aplicaciones, o notificaciones en tiempo real que no importa si se pierden algunas.

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

Cada instancia de aplicación se suscribe al canal cache-invalidation, recibe la clave a eliminar de su caché local. Si una instancia está reiniciando y pierde el evento, solo usa la caché anterior durante un corto tiempo — no es un error crítico.

4.2. Cuándo usar Streams

Usa Streams cuando el mensaje es información comercial importante que requiere procesamiento completo: órdenes, transacciones de pago, registros de auditoría. Necesitas garantizar que aunque el consumidor falle en medio del procesamiento, pueda procesar ese registro de nuevo después de reiniciar.

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

Cada evento de pago se escribe en payment-stream. Incluso si PaymentWorker está apagado para desplegar, este evento espera en Redis, el worker reiniciado continuará leyendo.

5. Escribir pruebas para Stream Consumer

Debido a que Streams tiene estado persistente, debes probar cuidadosamente el flujo de publicación – consumo – reconocimiento. Usa Testcontainers para ejecutar Redis real, evita simular un comportamiento incorrecto.

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

Esta prueba publica un registro, llama a pollOnce() para consumir y reconocer, luego verifica que la lista de pendientes esté vacía — significa que el registro ha sido confirmado como procesado, no está más pendiente en el grupo.

6. Errores comunes

• Usar Pub/Sub para datos importantes: muchas personas nuevas usan Pub/Sub para enviar eventos de órdenes luego se sorprenden cuando el consumidor se reinicia hace que se pierdan eventos. Pub/Sub no es una cola de mensajes.
• No recortar Stream: Stream no se elimina automáticamente. Si publicas continuamente y no llamas a XTRIM o estableces MAXLEN cuando agregas, la memoria de Redis crecerá con el tiempo.

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

• Olvidar crear un grupo de consumidores antes de leer: el comando XREADGROUP falla si el grupo aún no existe. Siempre llama a createGroup con try-catch para capturar el error “BUSYGROUP” (el grupo ya existe).
• No establecer timeout de conexión para Pub/Sub: el suscriptor usa una conexión de bloqueo privada; si la red se corta y no hay mecanismo de reconexión, pierdes todos los mensajes desde ese momento hasta que detectes y te reconectes.

7. Conclusión

Pub/Sub y Streams resuelven dos problemas diferentes aunque ambos están en Redis. Pub/Sub es simple, rápido, pero no garantiza nada — adecuado para notificaciones inmediatas que se pueden perder. Streams es más complejo pero es resistente, tiene grupos de consumidores y reconocimiento — adecuado para información comercial que requiere procesamiento completo, como una cola de mensajes ligera.

Si tu sistema necesita aún mayor confiabilidad (exactly-once, particiones grandes, múltiples grupos de consumidores en paralelo a gran escala), considera Kafka o RabbitMQ en lugar de forzar Redis Streams a hacer demasiadas cosas.