Implementar un mecanismo donde el receptor de mensajes en RabbitMQ notifique al emisor sobre el estado del consumo es una necesidad común en sistemas asíncronos. En este enfoque, el emisor original se suscribe a una cola de respuesta dedicada para recibir confirmaciones del receptor, permitiendo así actualizar el estado de los mensajes, por ejemplo, en una base de datos.

La arquitectura propuesta utiliza una cola de respuesta separada a la que el receptor publica un mensaje después de procesar el original. El emisor monitorea esta cola para correlacionar las respuestas con los mensajes enviados previamente mediante un identificador único. Esto difumina los roles tradicionales: el emisor actúa simultáneamente como productor y consumidor de la cola de respuesta, meintras que el receptor se convierte en un consumidor que genera un nuevo mensaje de confirmación.

Spring Boot simplifica esta implementación con anotaciones como @SendTo, que redirige el resultado de un método de escucha a una cola específica. A continuación, se detallla una configuración y código de ejemplo que logra esta funcionalidad, con cambios en nombres y estructura para ilustrar variaciones.

Configuración de la Aplicación (application.yml)

spring:
    rabbitmq:
        host: 192.168.1.100
        port: 5672
        username: user
        password: secret
        publisher-confirms: true
        publisher-returns: true
        template:
            mandatory: true
        listener:
            simple:
                prefetch: 1
                default-requeue-rejected: false
                concurrency: 1
                max-concurrency: 1
                acknowledge-mode: manual
notification:
    queue:
        name: system.response.queue
    exchange:
        name: notification.exchange
command:
    queue:
        name: system.command.queue

Clase de Configuración para RabbitMQ

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MqConfiguration {

    @Value("${command.queue.name}")
    private String commandQueueName;
    @Value("${notification.queue.name}")
    private String notificationQueueName;
    @Value("${notification.exchange.name}")
    private String notificationExchangeName;

    @Bean
    public Queue commandQueue() {
        return new Queue(commandQueueName);
    }

    @Bean
    public Queue notificationQueue() {
        return new Queue(notificationQueueName);
    }

    @Bean
    public TopicExchange notificationExchange() {
        return new TopicExchange(notificationExchangeName);
    }

    @Bean
    public Binding commandQueueBinding() {
        return BindingBuilder.bind(commandQueue()).to(notificationExchange()).with(commandQueueName + ".#");
    }
}

Controlador del Emisor de Mensajes

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.Map;
import java.util.UUID;

@RestController
public class MessageSenderController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Value("${notification.exchange.name}")
    private String exchangeName;

    @GetMapping("/enviarComando")
    public void enviarComando() {
        String correlationId = UUID.randomUUID().toString().replace("-", "");
        String mensaje = "{\"id\":\"" + correlationId + "\",\"datos\":\"Contenido de ejemplo\"}";
        rabbitTemplate.convertAndSend(exchangeName, "system.command.queue.order", mensaje, message -> {
            message.getMessageProperties().setCorrelationId(correlationId);
            return message;
        });
    }

    @RabbitListener(queues = "system.response.queue")
    @RabbitHandler
    public void procesarRespuesta(String mensaje, Channel canal, @Header(AmqpHeaders.DELIVERY_TAG) long etiqueta) throws IOException {
        try {
            // Lógica de negocio para manejar la respuesta
            System.out.println("Respuesta recibida: " + mensaje);
            canal.basicAck(etiqueta, false);
        } catch (Exception e) {
            canal.basicNack(etiqueta, false, false);
        }
    }
}

Controlador del Receptor de Mensajes

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.annotation.SendTo;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;
import java.io.IOException;

@Component
public class MessageReceiverController {

    @RabbitListener(queues = "system.command.queue")
    @RabbitHandler
    @SendTo("system.response.queue")
    public String consumirComando(String mensaje, Channel canal, @Header(AmqpHeaders.DELIVERY_TAG) long etiqueta) throws IOException {
        try {
            // Procesar el comando original
            System.out.println("Comando procesado: " + mensaje);
            canal.basicAck(etiqueta, false);
            return "{\"estado\":\"procesado\",\"detalle\":\"Completado con éxito\"}";
        } catch (Exception e) {
            canal.basicNack(etiqueta, false, false);
            return "{\"estado\":\"error\"}";
        }
    }
}