Redis Streams com consumer group: processamento confiável

1. Visão geral

Pub/Sub do Redis tem uma grande desvantagem. Se o consumer está desligado quando a mensagem é publicada, ela é perdida. Redis Streams resolve esse problema. As mensagens são armazenadas permanentemente no stream, e cada consumer group se lembra de quais mensagens já foram processadas.

Este artigo o guia na construção de um pipeline de processamento de eventos de pedidos com Spring Boot 3 e Spring Data Redis. O pipeline compreende os seguintes passos: escrever o evento, ler via consumer group, ACK manual, recuperar mensagens travadas e transferir mensagens com erro para o stream dead-letter.

2. Como funcionam Redis Streams e consumer groups

Antes de escrever Java, você deve experimentar os comandos nativos em 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

O significado de cada comando:

• XADD adiciona uma entrada ao stream orders. O * diz ao Redis para gerar automaticamente um ID na forma <millis>-<seq>.
• XGROUP CREATE ... 0 cria o grupo shipping-group e começa a ler desde o início do stream. Se usar $, o grupo só lerá mensagens novas.
• XREADGROUP ... > obtém as mensagens que ainda não foram entregues a ninguém no grupo. A mensagem é entregue a worker-1 e, ao mesmo tempo, é registrada na Pending Entries List (PEL) do grupo.
• XPENDING mostra quantas mensagens foram entregues mas ainda não foram ACKed.
• XACK remove a mensagem da PEL. A partir desse momento, o grupo considera a mensagem como processada.

A saída de XPENDING antes do ACK se parece com:

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

A PEL é a chave da confiabilidade. Suponha que o consumer trave após receber a mensagem mas antes do ACK. A mensagem permanece na PEL e outro consumer pode reivindicá-la (claim) para continuar processando. Esta é a semântica at-least-once.

3. Instalação

Adicione a dependência do Spring Data Redis. Este starter usa Lettuce como cliente padrão:

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

Configure a conexão e o nome do stream:

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

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

O nome do consumer é obtido de HOSTNAME, portanto cada instância (pod) tem um nome único. Isso é importante. Se duas instâncias usarem o mesmo nome, Redis as tratará como um único consumer e suas PELs se misturarão.

4. Escrever eventos no 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(...) cria um MapRecord com campos string. O comando opsForStream().add() é equivalente a XADD orders * ... e retorna um RecordId gerado pelo Redis, por exemplo 1728640000000-0.

Usar StringRedisTemplate torna os dados no Redis legíveis com redis-cli. Você não precisa lidar com serialização de bytes do JDK.

5. Criar um consumer group com segurança

O grupo deve existir antes de ler. A aplicação pode ser inicializada quando o stream não tiver mensagens, portanto você precisa do parâmetro MKSTREAM. O método createGroup() de StreamOperations não habilita esse parâmetro. Portanto, chamamos na camada de conexão:

@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;
            }
            // Grupo já existe: normal ao reiniciar a aplicação
        }
    }
}

O último parâmetro true habilita MKSTREAM. Redis criará um stream vazio se o stream ainda não existir. Quando a aplicação reinicia, Redis retorna o erro BUSYGROUP Consumer Group name already exists. Ignoramos esse erro específico e relançamos outros erros.

6. Ler mensagens com StreamMessageListenerContainer

6.1. Listener com 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());
    }
}

O comando ACK é executado apenas quando createShipment é bem-sucedido. Se o método lançar uma exceção, a mensagem permanece na PEL para ser processada novamente mais tarde. Não faça ACK antes de processar. Fazer isso mudaria at-least-once para at-most-once.

O ShippingService neste artigo é um serviço simples. Ele registra os orderIds para os quais os envios foram criados:

@Service
public class ShippingService {

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

    public void createShipment(String orderId) {
        if (!shippedOrders.add(orderId)) {
            return; // já processado, pula (idempotente)
        }
        // chama a API da transportadora...
    }

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

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

Os pontos principais na configuração:

• ReadOffset.lastConsumed() corresponde a > em XREADGROUP, ou seja, obtém apenas as mensagens novas do grupo.
• pollTimeout é o tempo de bloqueio de cada leitura. batchSize é o número máximo de mensagens em uma leitura.
• autoAcknowledge(false) desabilita o ACK automático, já que fazemos ACK manualmente no listener.
• cancelOnError(t -> false) é a linha mais importante. Por padrão, quando o listener lança uma exceção, o container cancela a subscription. O consumer silenciosamente parará de ler sem nenhum aviso claro.

Nota: Não use receiveAutoAck(...) para operações críticas. Esse modo faz ACK imediatamente após a leitura, ou seja, antes de o listener terminar. Se a aplicação travar no meio, a mensagem será perdida permanentemente.

7. Processar mensagens travadas e dead-letter

O listener só lê mensagens novas (>). Uma mensagem com erro ou pertencente a um consumer morto permanecerá na PEL indefinidamente se ninguém a processar. Precisamos de um job periódico para isso:

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

O fluxo de processamento do job:

1. pending(...) é equivalente a XPENDING orders shipping-group - + 50. Cada PendingMessage informa qual consumer está segurando a mensagem, o tempo ocioso e quantas vezes ela foi entregue.
2. claim(...) é equivalente a XCLAIM. Este comando transfere a propriedade para o consumer atual. Redis só faz a transferência se a mensagem ainda está ociosa por pelo menos MIN_IDLE, portanto duas instâncias executando o job ao mesmo tempo não reivindicarão a mesma mensagem. Cada reivindicação também incrementa o delivery count.
3. As mensagens entregues mais de MAX_DELIVERIES vezes são copiadas para o stream orders:dlq e depois feitas ACK. Mensagens de erro permanente (poison messages) portanto não são repetidas infinitamente.

Se createShipment lançar uma exceção, o comando ACK não é executado. A mensagem permanece na PEL e será repetida no próximo ciclo. Lembre-se de ativar @EnableScheduling em uma classe @Configuration.

Nota: A partir do Redis 6.2 há o comando XAUTOCLAIM. Este comando combina os passos XPENDING e XCLAIM em um e é mais eficiente quando a PEL é grande. Você pode chamá-lo via redisTemplate.execute(...) se a versão do Spring Data Redis que está usando ainda não tem uma API conveniente.

8. Escrever testes com Testcontainers

Adicione as dependências para teste:

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

O teste abaixo publica um evento. Então verifica se o envio foi criado e a PEL está vazia:

@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 o Spring Boot obtenha automaticamente o host e a porta do container, para que você não precise de @DynamicPropertySource. await() é necessário porque o listener é executado em uma thread separada. A asserção getTotalPendingMessages() igual a 0 prova que a mensagem foi ACKed, não apenas lida.

9. Erros comuns e quando não usar

Processamento não-idempotente. At-least-once significa que uma mensagem pode chegar duas vezes. Por exemplo: o consumer processa mas trava antes do ACK. Portanto, a lógica de negócio deve ser resiliente a repetições. Uma abordagem comum é verificar se o orderId já foi processado ou usar uma restrição unique no banco de dados.

Stream crescendo infinitamente. ACK não remove a entrada do stream. Você precisa limitar o tamanho ao escrever:

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

Esta opção é equivalente a XADD orders MAXLEN ~ 100000 * .... O ~ permite que o Redis corte por nó interno, muito mais rápido do que cortar com precisão.

Esquecer cancelOnError. A primeira exceção silenciosamente parará o consumer. O sintoma é que as mensagens se acumulam no stream sem erros claros nos logs.

Consumer “fantasma”. Cada deploy com um novo nome de consumer (por exemplo, nome do pod) deixa o nome antigo no grupo. Limpe periodicamente com XGROUP DELCONSUMER, mas apenas depois de ter reivindicado todas as mensagens pendentes desse consumer.

Quando NÃO usar Redis Streams:

• Você precisa armazenar eventos por muitos dias e fazer replay de grandes volumes. Redis mantém todos os dados na RAM, portanto o custo é muito alto.
• Você precisa garantir zero perda de dados. Com AOF everysec ou failover de replica assíncrono, ainda é possível perder algumas mensagens recém-escritas.
• Throughput muito alto, precisando particionar e escalar para centenas de consumers. Kafka é mais adequado neste caso.

10. Conclusão

Redis Streams combinado com consumer groups oferece uma fila de eventos leve e confiável rodando no Redis que você já tem. Você precisa lembrar de quatro pontos principais:

• ACK manual após terminar o processamento.
• Configure cancelOnError(t -> false) para que o consumer não pare silenciosamente.
• Tenha um job que reclame as mensagens travadas, com um stream dead-letter para poison messages.
• Escreva lógica de processamento idempotente.

Adicione MAXLEN para controlar a memória e teste integrado com Testcontainers, e você terá um pipeline de processamento de eventos at-least-once suficientemente confiável para a maioria das aplicações Spring Boot.