RabbitMQ Espere un mensaje con un tiempo de espera

Question

Me gustaría enviar un mensaje a un servidor RabbitMQ y luego esperar un mensaje de respuesta (en una cola de "respuesta a"). Por supuesto, no quiero esperar para siempre en caso de que la aplicación que está procesando estos mensajes no funcione, debe haber un tiempo de espera. Parece una tarea muy básica, pero no puedo encontrar la forma de hacerlo. Ahora me he encontrado con este problema con la API de Java.

Answer 1

Hay una pregunta similar. Aunque sus respuestas no usan Java, tal vez puedas obtener algunas pistas.

Wait for a single RabbitMQ message with a timeout

Answer 2

com.rabbitmq.client.QueueingConsumer tiene un método nextDelivery(long timeout), que va a hacer lo que quiere. Sin embargo, esto ha sido desaprobado. Escribir su propio tiempo de espera no es tan difícil, aunque puede ser mejor tener un hilo continuo y una lista de identificadores en tiempo, en lugar de agregar y quitar los consumidores y los hilos de tiempo de espera asociados todo el tiempo.

Editar para añadir: ¡Ha notado la fecha después de responder!

Answer 3

Abordé este problema usando C# creando un objeto para realizar un seguimiento de la respuesta a un mensaje en particular. Configura una cola de respuesta única para un mensaje y se suscribe a ella. Si la respuesta no se recibe en un marco de tiempo específico, un temporizador de cuenta regresiva cancela la suscripción, que elimina la cola. Por separado, tengo métodos que pueden ser sincrónicos desde mi hilo principal (usa un semáforo) o asincrónico (utiliza una devolución de llamada) para utilizar esta funcionalidad.

Básicamente, la puesta en práctica se parece a esto:

//Synchronous case: 
//Throws TimeoutException if timeout happens 
var msg = messageClient.SendAndWait(theMessage); 

//Asynchronous case 
//myCallback receives an exception message if there is a timeout 
messageClient.SendAndCallback(theMessage, myCallback); 

Answer 4

La biblioteca cliente Java ahora RabbitMQ supports a timeout argument to its QueueConsumer.nextDelivery() method.

Por ejemplo, el tutorial RPC utiliza el siguiente código:

channel.basicPublish("", requestQueueName, props, message.getBytes()); 

while (true) { 
    QueueingConsumer.Delivery delivery = consumer.nextDelivery(); 
    if (delivery.getProperties().getCorrelationId().equals(corrId)) { 
     response = new String(delivery.getBody()); 
     break; 
    } 
} 

Ahora, puede utilizar consumer.nextDelivery(1000) que esperar para obtener el máximo de un segundo. Si se alcanza el tiempo de espera, el método devuelve null.

channel.basicPublish("", requestQueueName, props, message.getBytes()); 

while (true) { 
    // Use a timeout of 1000 milliseconds 
    QueueingConsumer.Delivery delivery = consumer.nextDelivery(1000); 

    // Test if delivery is null, meaning the timeout was reached. 
    if (delivery != null && 
     delivery.getProperties().getCorrelationId().equals(corrId)) { 
     response = new String(delivery.getBody()); 
     break; 
    } 
}