RabbitMQ Attendere un messaggio con un timeout

Question

Desidero inviare un messaggio a un server RabbitMQ e quindi attendere un messaggio di risposta (in una coda "reply-to"). Certo, non voglio aspettare per sempre nel caso in cui l'applicazione che elabora questi messaggi sia inattiva - deve esserci un timeout. Sembra un compito molto semplice, ma non riesco a trovare un modo per farlo. Ora ho incontrato questo problema con l'API Java.

Answer 1

C'è una domanda simile. Anche se le sue risposte non usano java, forse puoi avere qualche suggerimento.

Wait for a single RabbitMQ message with a timeout

Answer 2

com.rabbitmq.client.QueueingConsumer ha un metodo nextDelivery(long timeout), che farà quello che vuoi. Tuttavia, questo è stato deprecato. Scrivere il proprio timeout non è così difficile, anche se potrebbe essere meglio avere un thread in corso e un elenco di identificatori in tempo, piuttosto che aggiungere e rimuovere utenti e thread di timeout associati per tutto il tempo.

Modifica per aggiungere: annotato la data su questo dopo aver risposto!

Answer 3

Mi sono avvicinato a questo problema utilizzando C# creando un oggetto per tenere traccia della risposta a un particolare messaggio. Imposta una coda di risposta univoca per un messaggio e si abbona ad essa. Se la risposta non viene ricevuta in un intervallo di tempo specificato, un timer per il conto alla rovescia annulla la sottoscrizione, che elimina la coda. Separatamente, ho metodi che possono essere sincroni dal mio thread principale (usa un semaforo) o asincroni (usa un callback) per utilizzare questa funzionalità.

In sostanza, l'applicazione si presenta così:

//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 libreria client Java ora RabbitMQ supports a timeout argument to its QueueConsumer.nextDelivery() method.

Per esempio, il tutorial RPC utilizza il seguente codice:

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

Ora, è possibile utilizzare consumer.nextDelivery(1000) per attendere al massimo un secondo. Se viene raggiunto il timeout, il metodo restituisce 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; 
    } 
}