Utilizzando RabbitMQ (client Java), esiste un modo per determinare se la connessione di rete viene chiusa durante il consumo?

Question

Sto usando RabbitMQ su RHEL 5.3 utilizzando il client Java. Ho 2 nodi (macchine). Il nodo 1 sta consumando messaggi da una coda su Nodo2 utilizzando la classe helper Java QueueingConsumer.

QueueingConsumer consumer = new QueueingConsumer(channel); 
channel.basicConsume("MyQueueOnNode2", noAck, consumer); 
while (true) 
{ 
    QueueingConsumer.Delivery delivery = consumer.nextDelivery(); 
    ... Process message - delivery.getBody() 
} 

Se l'interfaccia è portato giù sul Nodo 1 o Nodo2 (ad esempio ifconfig eth1 verso il basso), il cliente (sopra) non si sa mai la rete non è più lì. RabbitMQ fornisce un tipo di configurazione sul client Java che può essere utilizzata per determinare se la connessione è andata via. La chiusura del server RabbitMQ su Nodo2 attiverà un'eccezione ShutdownSignalException, che può essere intercettata e l'app può entrare in un ciclo di riconnessione. Ma l'interruzione dell'interfaccia non causa alcun tipo di eccezione, quindi il codice sarà in attesa per sempre su consumer.nextDelivery().

Ho anche provato a utilizzare la versione di timeout di questa chiamata. per esempio.

QueueingConsumer consumer = new QueueingConsumer(channel); 
channel.basicConsume("MyQueueOnNode2", noAck, consumer); 
int timeout_ms = 30000; 
while (true) 
{ 
    QueueingConsumer.Delivery delivery = consumer.nextDelivery(timeout_ms); 
    if (delivery == null) 
    { 
     if (channel.isOpen() == false)    // Seems to always return true 
     { throw new ShutdownSignalException(); } 
    } 
    else 
    { 
    ... Process message - delivery.getBody() 
    } 
} 

ma sembra che questo restituisca sempre true (anche se l'interfaccia è inattiva). Presumo che la registrazione per ShutdownListener sulla connessione produrrà gli stessi risultati, ma non l'ho ancora provato.

C'è un modo per configurare una sorta di heartbeat, o devi solo scrivere la logica del lease personalizzato (ad esempio "Sono qui ora") per farlo funzionare?

Answer 1

In generale, è molto meglio postare domande relative a rabbitmq sulla mailing list di rabbitmq-discuss. Non tendiamo a tenere traccia delle domande poste al di fuori di questo.

C'è un heartbeat che è possibile configurare, anche se è disattivato per impostazione predefinita. È anche possibile attivare TCP Keep Alive. Chiamare setRequestedHeartbeat su ConnectionFactory prima di creare una nuova connessione, o sottoclasse ConnectionFactory, ignorare il metodo configureSocket e chiamare socket.setKeepAlive(true). Entrambi dovrebbero comportare la connessione notare quando la rete muore.

Answer 2

Per quanto riguarda il metodo di isOpen, che è ben descritto nella documentazione: http://www.rabbitmq.com/api-guide.html#shutdown-atomicity

Per quanto riguarda la chiusura di: con la chiusura node1 o 2 si intende la giusta applicazione, non il server RabbitMQ stessa? Perché vorresti sapere su qualsiasi applicazione se un'altra applicazione si disconnette dal broker dei messaggi? Non è questo il punto di messaggistica.

L'unica cosa che puoi fare è inviare messaggi con un parametro 'obbligatorio'. Questo dice al server RabbitMQ che ti aspetti almeno 1 listener per il messaggio che hai inviato (sia che si tratti di una coda diretta o di qualche coda in uno scambio topic/fanout). Se il messaggio non può essere consegnato a nessuna coda, il messaggio tornerà al tuo canale e verrà inoltrato a ReturnListener.