Fluxo de Confirmação no RabbitMQ

Em um sistema RabbitMQ, o produtor envia mensagens ao broker, e o consumidor as retira. O mecanismo de confirmação garante que o produtor saiba se a mensagem chegou ao broker, e que o broker saiba se o consumidor processou a mensagem. Sem isso, problemas como consumo duplicado podem surgir devido a falhas de rede.

A cnofirmação é dividida em dois lados: confirmação do produtor e confirmação do consumidor. No produtor, usa-se ConfirmCallback e ReturnCallback. No consumidor, utiliza-se ACK (confirmação positiva) e NACK (confirmação negativa).

Confirmação do Produtor

Quando uma mensagem chega a um exchange, o RabbitMQ envia um ConfirmCallback ao produtor, indicando sucesso ou falha. Se a mensagem chega ao exchange mas falha ao ser roteada para uma fila, um ReturnCallback é retornado somente em caso de falha.

Confirmação do Consumidor

Após receber uma mensagem, o consumidor deve enviar uma confirmação ACK ao RabbitMQ. Com base nisso, o broker pode remover a mensagem da fila ou reenviá-la.

Implementação em Código

Confirmação do Produtor

Para implementar, sobrescreva os métodos de callback no template do RabbitMQ. Primeiro, configure as opções de confirmação no arquivo de propriedades.

spring:
  rabbitmq:
    host: localhost
    port: 5672
    virtual-host: /
    username: admin
    password: senha_segura
    # Habilitar modos de confirmação do produtor
    publisher-confirm-type: correlated
    publisher-returns: true

Em seguida, declare um exchange, uma fila e os vincule usando uma classe de configuração.

@Configuration
public class ConfiguracaoRabbitMQ {

    private static final String NOME_EXCHANGE = "exchange_topico_logs";
    private static final String NOME_FILA = "fila_logs_topico";

    @Bean("exchangeLogs")
    public Exchange criarExchange() {
        return ExchangeBuilder.topicExchange(NOME_EXCHANGE).durable(true).build();
    }

    @Bean
    public Queue criarFila() {
        return QueueBuilder.durable(NOME_FILA).build();
    }

    @Bean
    public Binding vincularFilaExchange(@Qualifier("criarFila") Queue fila, @Qualifier("exchangeLogs") Exchange exchange) {
        return BindingBuilder.bind(fila).to(exchange).with("modulo.#").noargs();
    }
}

Crie um consumidor simples para receber mensagens.

@Component
@RabbitListener(queues = "fila_logs_topico")
public class ConsumidorBasico {

    @RabbitHandler
    public void processarMensagem(String conteudo) {
        System.out.println("Mensagem recebida: " + conteudo);
    }
}

No teste do produtor, defina os callbacks para simular cenários normais e de erro.

@SpringBootTest
class TesteProdutor {

    @Autowired
    private RabbitTemplate templateRabbit;

    @Test
    void testarConfirmacaoSucesso() {
        templateRabbit.setConfirmCallback((dadosCorrelacao, sucesso, mensagemErro) -> {
            if (sucesso) {
                System.out.println("Entrega confirmada com sucesso");
            } else {
                System.out.println("Falha na entrega: " + mensagemErro);
            }
        });
        templateRabbit.convertAndSend("exchange_topico_logs", "modulo.info", "Dados do log: teste");
    }

    @Test
    void testarConfirmacaoFalha() {
        templateRabbit.setConfirmCallback((dadosCorrelacao, sucesso, mensagemErro) -> {
            if (sucesso) {
                System.out.println("Entrega confirmada");
            } else {
                System.out.println("Erro na confirmação");
            }
        });
        // Enviar para um exchange inexistente
        templateRabbit.convertAndSend("exchange_invalido", "modulo.info", "Dados do log: teste");
    }

    @Test
    void testarRetornoErro() {
        templateRabbit.setReturnCallback((mensagem, codigoResposta, textoResposta, exchange, chaveRota) -> {
            System.out.println("Mensagem retornada: " + mensagem.toString());
            System.out.println("Chave de rota inválida: " + chaveRota);
        });
        templateRabbit.setMandatory(true);
        // Enviar com chave de rota inválida
        templateRabbit.convertAndSend("exchange_topico_logs", "rota.invalida", "Dados do log: teste");
    }
}

Confirmação do Consumidor

Para confirmação manual, altere o modo de reconhecimento no consumidor. Configure como manual nas propriedades.

spring:
  rabbitmq:
    host: localhost
    port: 5672
    virtual-host: /
    username: admin
    password: senha_segura
    listener:
      direct:
        acknowledge-mode: manual
      simple:
        acknowledge-mode: manual

Implemente o consumidor com confirmação manual usando o canal de mensagens.

@Component
@RabbitListener(queues = "fila_logs_topico")
public class ConsumidorManual {

    @RabbitHandler
    public void processarComConfirmacao(String conteudo, Channel canal, Message mensagem) throws IOException {
        try {
            System.out.println("Processando: " + conteudo);
            // Simular um erro para teste
            if (conteudo.contains("erro")) {
                throw new RuntimeException("Erro no processamento");
            }
            // ACK manual - confirmar recebimento
            canal.basicAck(mensagem.getMessageProperties().getDeliveryTag(), false);
            System.out.println("Mensagem confirmada e removida da fila");
        } catch (Exception e) {
            System.out.println("Falha no processamento, solicitando reenvio");
            // NACK - rejeitar e solicitar reenvio
            canal.basicNack(mensagem.getMessageProperties().getDeliveryTag(), false, true);
        }
    }
}

No teste, envie uma mensagem para o consumidro manual.

@SpringBootTest
class TesteConsumidor {

    @Autowired
    private RabbitTemplate templateRabbit;

    @Test
    void testarConfirmacaoManual() {
        templateRabbit.convertAndSend("exchange_topico_logs", "modulo.info", "Dados do log: normal");
    }
}