Usando RabbitMQ (cliente Java), ¿hay alguna forma de determinar si la conexión de red está cerrada durante el consumo?

Question

Estoy usando RabbitMQ en RHEL 5.3 usando el cliente de Java. Tengo 2 nodos (máquinas). Node1 está consumiendo mensajes de una cola en Node2 utilizando la clase de ayuda de Java QueueingConsumer.

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

Si la interfaz es llevado hacia abajo en el nodo 1 o Nodo2 (por ejemplo, ifconfig eth1 hacia abajo), el cliente (arriba) nunca sabe la red ya no está allí. ¿RabbitMQ proporciona algún tipo de configuración en el cliente Java que se puede usar para determinar si la conexión se ha eliminado? Si se cierra el servidor RabbitMQ en el Nodo2, se activará ShutdownSignalException, que se puede capturar y la aplicación puede entrar en un circuito de reconexión. Pero cerrar la interfaz no causa ningún tipo de excepción, por lo que el código estará esperando por siempre en consumer.nextDelivery().

También intenté utilizar la versión de tiempo de espera de esta llamada. p.ej.

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() 
    } 
} 

pero parece que esto siempre devuelve verdadero (aunque la interfaz esté desactivada). Supongo que registrarse para ShutdownListener en la conexión arrojará los mismos resultados, pero aún no lo he intentado.

¿Hay alguna forma de configurar algún tipo de latido, o simplemente tiene que escribir la lógica de arrendamiento personalizada (por ejemplo, "Estoy aquí ahora") para que esto funcione?

Answer 1

En general, es mucho mejor publicar preguntas sobre rabbitmq en la lista de correo de rabbitmq-discuss. No tendemos a rastrear preguntas que se hacen fuera de esto.

Hay un latido que puede configurar, aunque está desactivado por defecto. También puede activar TCP Keep Alive. Llame al setRequestedHeartbeat en el ConnectionFactory antes de crear una nueva conexión o, la subclase ConnectionFactory, anule el método configureSocket y llame al socket.setKeepAlive(true). Ambos deberían provocar que la conexión se note cuando la red muere.

Answer 2

Con respecto al método isOpen, que está bien descrita en la documentación: http://www.rabbitmq.com/api-guide.html#shutdown-atomicity

En cuanto al cierre: con el cierre de nodo 1 o 2 que quiere decir el derecho de aplicación, no el propio servidor RabbitMQ? ¿Por qué le gustaría saber en cualquier aplicación si otra aplicación se desconecta del intermediario de mensajes? Ese no es el punto de los mensajes.

Lo único que puede hacer es enviar mensajes con un parámetro 'obligatorio'. Eso le dice al servidor RabbitMQ que espera al menos 1 oyente para el mensaje que ha enviado (ya sea una cola directa o alguna cola en un intercambio de tema/fanout). Si el mensaje no se puede entregar a ninguna cola, el mensaje volverá a su canal y se reenviará a ReturnListener.