Pub/Sub e Redis Streams: Diferenças e Como usar com Spring Boot

1. Visão geral

O Redis oferece dois mecanismos diferentes de transmissão de mensagens: Pub/Sub e Streams. Muitos iniciantes no Redis costumam confundir esses dois conceitos como sendo um só, levando à escolha errada da ferramenta e perda de dados quando o sistema reinicia.

Este artigo explica as diferenças fundamentais entre Pub/Sub e Streams, fornece código Java que funciona com ambos, e deixa claro quando usar cada um. Após ler, você terá confiança para escolher o modelo correto para seu problema real de mensageria.

2. O que é Redis Pub/Sub

Pub/Sub (Publish/Subscribe) funciona segundo o modelo “dispara e esquece” (fire-and-forget). O Publisher envia uma mensagem para um channel, e o Redis a retransmite imediatamente para todos os subscribers que estão escutando esse channel. Se nenhum subscriber estiver online, a mensagem desaparece para sempre.

2.1. Dependência 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 envia a mensagem para o channel order-events. Se ainda não houver ninguém se subscrevendo, o comando ainda assim é bem-sucedido, mas a mensagem não fica armazenada em lugar nenhum.

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("Evento recebido: " + 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 gerencia uma conexão dedicada (usando blocking connection) para escutar o channel order-events. Quando o publisher envia uma mensagem, onMessage é chamado imediatamente.

Saída quando você publica um evento:

Evento recebido: orderId=ORD-001;status=CONFIRMED

Atenção: Se OrderEventListener não tiver inicializado (o container ainda não fez bind), a mensagem publicada desaparece completamente, não há forma de reproduzi-la novamente.

3. O que é Redis Streams

Stream é uma estrutura de dados de log ordenada, append-only, onde cada mensagem possui um ID único no formato <timestamp>-<sequence>. Diferentemente de Pub/Sub, Streams armazena a mensagem no Redis até que você a remova explicitamente (XDEL) ou a reduza (XTRIM). Consumidores podem ler do início, ler em grupos (consumer group) com mecanismo de acknowledge semelhante ao Kafka.

3.1. Colocar mensagens em um 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 escreve um registro no stream order-stream e retorna um RecordId — semelhante a uma chave primária auto-gerada do registro. Esse registro permanece no Redis até ser reduzido ou deletado, independentemente se alguém já o leu.

3.2. Ler 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) {
            // Grupo já existe, ignorar
        }
    }

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

• createGroup cria o consumer group order-processors, semelhante ao consumer group no Kafka.
• Consumer.from("order-processors", "worker-1") declara um consumer chamado worker-1 pertencente a esse grupo. Múltiplos workers no mesmo grupo compartilham os registros, não leem duplicados.
• ReadOffset.lastConsumed() significa ler registros novos que ainda não foram atribuídos a nenhum consumer no grupo.
• acknowledge confirma que terminou o processamento, o Redis remove o registro da “pending list” do grupo.

Saída:

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

Atenção: Se você esquecer de chamar acknowledge, o registro permanece na pending list. Outro consumer ou o próprio consumer (após reiniciar) pode usar XCLAIM/readPending para recuperar e processar novamente — esse é o mecanismo de “at-least-once delivery” que Pub/Sub não possui.

4. Comparação direta entre Pub/Sub e Streams

Critério	Pub/Sub	Streams
Armazenamento de mensagens	Não	Sim, até ser reduzido/deletado
Reproduzir mensagens antigas	Não	Sim (XRANGE, offset customizável)
Consumer offline perde	Todas as mensagens durante offline	Nenhuma, pode ler novamente quando online
Balanceamento de carga com múltiplos consumers	Cada subscriber recebe sua própria cópia (broadcast)	Consumer group distribui trabalho, cada registro é processado por um consumer
Acknowledge / retry	Não possui	Sim (XACK, XCLAIM)
Complexidade	Muito simples	Mais complexo, semelhante ao Kafka
Caso de uso apropriado	Notificações em tempo real, sem garantia necessária	Event sourcing, job queue, precisa garantir o processamento

4.1. Quando usar Pub/Sub

Use Pub/Sub quando você precisa de broadcast instantâneo e aceita perder mensagens se ninguém estiver escutando. Exemplos clássicos: atualizar status “digitando…” em um chat, sincronizar cache entre múltiplas instâncias da aplicação (invalidar cache), ou notificações em tempo real onde é aceitável perder alguns eventos.

@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 instância da aplicação se subscrive ao channel cache-invalidation, recebendo chaves que precisam ser removidas do cache local. Se uma instância estiver reiniciando e perder um evento, ela apenas usará o cache antigo por um tempo — não é um erro crítico.

4.2. Quando usar Streams

Use Streams quando as mensagens são dados críticos do negócio que precisam ser processadas completamente: pedidos, transações de pagamento, logs de auditoria. Você precisa garantir que um consumer que falhar no meio do caminho ainda conseguirá processar o registro novamente após 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 pagamento é registrado no payment-stream. Mesmo que PaymentWorker esteja desligado para deploy, o evento permanece armazenado no Redis, e o worker continuará processando quando reiniciar.

5. Escrevendo testes para o Stream Consumer

Como Streams possuem estado persistente, você deve testar cuidadosamente o fluxo de publicação – consumo – acknowledge. Use Testcontainers para executar um Redis real, evitando mockar incorretamente o comportamento.

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

Esse teste publica um registro, chama pollOnce() para consumir e fazer acknowledge, e depois verifica se a pending list está vazia — significando que o registro foi confirmado como processado e não está mais pendente no grupo.

6. Erros comuns

• Usar Pub/Sub para dados críticos: muitas pessoas iniciantes usam Pub/Sub para enviar eventos de pedidos e ficam surpresas quando o consumer reinicia e o evento se perde. Pub/Sub não é uma message queue.
• Não fazer trim no Stream: Streams não deletam dados antigos automaticamente. Se você publicar continuamente sem chamar XTRIM ou definir MAXLEN ao fazer add, a memória do Redis crescerá com o tempo.

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

• Esquecer de criar o consumer group antes de ler: o comando XREADGROUP falhará se o grupo não existir. Sempre chame createGroup com try-catch para capturar o erro “BUSYGROUP” (grupo já existe).
• Não configurar connection timeout para Pub/Sub: o subscriber usa uma conexão blocking dedicada; se a rede cair e não houver mecanismo de reconexão, você perde todas as mensagens daquele ponto até detectar e reconectar.

7. Conclusão

Pub/Sub e Streams resolvem dois problemas diferentes, ainda que ambos residam no Redis. Pub/Sub é simples, rápido, mas não oferece garantias — apropriado para notificações instantâneas que podem ser ignoradas. Streams é mais complexo mas robusto, com consumer groups e acknowledge — apropriado para dados críticos do negócio que precisam ser processados completamente, semelhante a uma message queue leve.

Se seu sistema precisa de ainda maior confiabilidade (exactly-once, particionamento grande, muitos consumer groups em paralelo em escala grande), considere Kafka ou RabbitMQ em vez de forçar o Redis Streams a fazer muito mais trabalho.