Redis Streams con consumer group: procesamiento de eventos confiable

1. Panorama general

Pub/Sub de Redis tiene una gran desventaja. Si el consumer está apagado cuando se publica, el mensaje se pierde. Redis Streams resuelve esto. El mensaje se guarda duraderamente en el stream, y cada consumer group registra qué mensajes ya fueron procesados completamente.

Este artículo te guiará para construir un flujo de procesamiento de eventos de órdenes con Spring Boot 3 y Spring Data Redis. El flujo consta de pasos: escribir eventos, leerlos mediante consumer group, ACK manual, recuperar mensajes pendientes y enviar mensajes fallidos a un stream dead-letter.

2. Cómo funcionan Redis Streams y consumer group

Antes de escribir Java, deberías probar los comandos nativos en 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

Significado de cada comando:

• XADD agrega una entrada al stream orders. El * hace que Redis genere automáticamente un ID en formato <millis>-<seq>.
• XGROUP CREATE ... 0 crea el group shipping-group y comienza a leer desde el inicio del stream. Si usas $, el group solo leerá mensajes nuevos.
• XREADGROUP ... > obtiene mensajes no asignados a nadie en el group. El mensaje se asigna a worker-1 y se registra en la Pending Entries List (PEL) del group.
• XPENDING muestra cuántos mensajes se han asignado pero no confirmado (ACK).
• XACK elimina el mensaje de la PEL. A partir de ese momento, el group considera el mensaje procesado.

La salida de XPENDING antes del ACK se vería así:

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

La PEL es la clave de la confiabilidad. Supongamos que el consumer falla después de recibir el mensaje pero antes del ACK. El mensaje sigue en la PEL, y otro consumer puede reclamarlo (claim) para procesarlo. Esta es la semántica at-least-once.

3. Instalación

Añade la dependencia Spring Data Redis. Este starter usa Lettuce como cliente predeterminado:

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

Configura la conexión y los nombres de stream:

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

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

El nombre del consumer se toma de HOSTNAME, así que cada instancia (pod) tiene un nombre único. Esto es importante. Si dos instancias usan el mismo nombre, Redis las considera un solo consumer, y sus PEL se mezclan.

4. Escribir eventos en el 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(...) crea un MapRecord con campos de tipo String. El comando opsForStream().add() equivale a XADD orders * ... y devuelve el RecordId generado por Redis, por ejemplo 1728640000000-0.

Usar StringRedisTemplate permite que los datos en Redis sean legibles desde redis-cli. No tienes que lidiar con bytes del serialization de JDK.

5. Crear consumer group de forma segura

El group debe existir antes de leer. La aplicación puede iniciar cuando el stream no tiene mensajes, así que necesitas el parámetro MKSTREAM. El método createGroup() de StreamOperations no activa este parámetro. Por eso necesitas bajar al nivel de conexión:

@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;
            }
            // El group ya existe: normal cuando reiniciar la aplicación
        }
    }
}

El último parámetro true activa MKSTREAM. Redis creará un stream vacío si aún no existe. Cuando reinicies la aplicación, Redis devuelve el error BUSYGROUP Consumer Group name already exists. Ignoramos este error específico y relanzamos cualquier otro.

6. Leer mensajes con StreamMessageListenerContainer

6.1. Listener con ACK manual

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

El comando ACK solo se ejecuta si createShipment tiene éxito. Si el método lanza una excepción, el mensaje permanece en la PEL para procesarlo después. Nunca hagas ACK antes de procesar. Hacerlo convertiría at-least-once en at-most-once.

ShippingService en este artículo es un servicio simple. Registra qué orderId ya tienen orden de envío creada:

@Service
public class ShippingService {

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

    public void createShipment(String orderId) {
        if (!shippedOrders.add(orderId)) {
            return; // ya procesado, ignorar (idempotent)
        }
        // llamar a API de la compañía de envío...
    }

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

6.2. Configurar el 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;
    }
}

Puntos clave en la configuración:

• ReadOffset.lastConsumed() corresponde a > en XREADGROUP, es decir, solo obtiene mensajes nuevos del group.
• pollTimeout es el tiempo de bloqueo de cada lectura. batchSize es el número máximo de mensajes en una sola lectura.
• autoAcknowledge(false) desactiva el ACK automático, porque haremos ACK manualmente en el listener.
• cancelOnError(t -> false) es la línea más importante. Por defecto, cuando el listener lanza una excepción, el container cancela la suscripción. El consumer dejará de leer silenciosamente sin avisos claros.

Nota: No uses receiveAutoAck(...) para operaciones críticas. Este modo confirma el mensaje inmediatamente al leerlo, es decir, antes de que el listener termine. Si la aplicación falla a mitad de camino, el mensaje se perderá para siempre.

7. Manejar mensajes pendientes y dead-letter

El listener solo lee mensajes nuevos (>). Un mensaje con error o de un consumer ya muerto permanecerá en la PEL para siempre si nadie lo procesa. Necesitas un job periódico para esto:

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

El flujo de procesamiento del job:

1. pending(...) equivale a XPENDING orders shipping-group - + 50. Cada PendingMessage te dice qué consumer tiene el mensaje, cuánto tiempo ha estado sin actividad y cuántas veces se ha entregado.
2. claim(...) equivale a XCLAIM. Este comando transfiere la propiedad al consumer actual. Redis solo transfiere si el mensaje ha estado inactivo al menos MIN_IDLE, así que dos instancias ejecutando el job simultáneamente no reclaman el mismo mensaje. Cada claim también incrementa el delivery count.
3. Los mensajes entregados más de MAX_DELIVERIES veces se copian al stream orders:dlq antes de confirmar. Los mensajes de error permanente (poison message) no se reintentan indefinidamente.

Si createShipment lanza una excepción, el comando ACK no se ejecuta. El mensaje permanece en la PEL e intentará procesarse en el siguiente ciclo. Recuerda activar @EnableScheduling en una clase @Configuration.

Nota: Desde Redis 6.2 existe el comando XAUTOCLAIM. Este comando combina XPENDING y XCLAIM en uno, y es más eficiente cuando la PEL es grande. Puedes llamarlo mediante redisTemplate.execute(...) si tu versión de Spring Data Redis aún no tiene una API conveniente.

8. Escribir tests con Testcontainers

Añade las dependencias para 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>

El test de abajo publica un evento. Luego verifica que la orden de envío fue creada y que la PEL está vacía:

@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") permite que Spring Boot tome automáticamente el host y puerto del container, así que no necesitas @DynamicPropertySource. await() es necesario porque el listener se ejecuta en un thread separado. La aserción getTotalPendingMessages() igual a 0 prueba que el mensaje fue confirmado, no solo leído.

9. Errores comunes y cuándo no usar

Procesamiento no idempotente. At-least-once significa que un mensaje puede llegar dos veces. Por ejemplo: el consumer procesa y falla antes del ACK. Entonces el negocio debe tolerar repeticiones. Una forma común es verificar si el orderId ya fue procesado, o usar una constraint única en la base de datos.

El stream crece indefinidamente. El ACK no elimina la entrada del stream. Necesitas limitar el tamaño al escribir:

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

Esta opción equivale a XADD orders MAXLEN ~ 100000 * .... El ~ permite que Redis corte según nodos internos, mucho más rápido que corte exacto.

Olvidar cancelOnError. Una excepción puede detener silenciosamente el consumer. El síntoma es que los mensajes se acumulan en el stream pero los logs no muestran errores claros.

Consumer “fantasma”. Cada vez que despliegas con un nuevo nombre de consumer (ej. nombre de pod), el nombre anterior sigue en el group. Deberías limpiar periódicamente con XGROUP DELCONSUMER, pero solo después de haber reclamado todos los mensajes pendientes de ese consumer.

Cuándo NO usar Redis Streams:

• Necesitas guardar eventos durante muchos días y hacer replay de grandes volúmenes. Redis mantiene todos los datos en RAM, así que el costo es muy alto.
• Necesitas garantizar que nunca se pierdan datos absolutamente. Con AOF everysec o failover de replica asíncrono, aún podrías perder algunos mensajes recién escritos.
• El throughput es muy alto, necesitas particionamiento y escalar a cientos de consumers. Kafka es más adecuado en este caso.

10. Conclusión

Redis Streams combinado con consumer group te ofrece una cola de eventos ligera y confiable, ejecutándose en el Redis que ya tienes. Necesitas recordar cuatro puntos clave:

• ACK manual después de procesar completamente.
• Establecer cancelOnError(t -> false) para que el consumer no se detenga silenciosamente.
• Tener un job que reclame mensajes pendientes, con stream dead-letter para poison messages.
• Escribir lógica de procesamiento idempotente.

Agregar MAXLEN para controlar la memoria y tests integrados con Testcontainers, tendrás un flujo de procesamiento de eventos at-least-once suficientemente confiable para la mayoría de aplicaciones Spring Boot.