使用RabbitMQ(Java客户端),有没有办法确定消费期间网络连接是否关闭？

Question

我正在使用Java客户端在RHEL 5.3上使用RabbitMQ.我有2个节点(机器).Node1使用Java帮助程序类QueueingConsumer消耗来自Node2上队列的消息.

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

如果接口在Node1或Node2上关闭(例如ifconfig eth1 down),则客户端(上面)永远不会知道网络不再存在.RabbitMQ是否在Java客户端上提供某种类型的配置,可用于确定连接是否已消失.关闭Node2上的RabbitMQ服务器将触发ShutdownSignalException,可以捕获该应用程序并且应用程序可以进入重新连接循环.但是,关闭接口不会导致任何类型的异常发生,因此代码将永远等待consumer.nextDelivery().

我也尝试过使用此调用的超时版本.例如

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

但似乎这总是返回true(即使接口已关闭).我假设在连接上注册ShutdownListener会产生相同的结果,但还没有尝试过.

有没有办法配置某种心跳,或者你只需​​要编写自定义租约逻辑(例如"我现在在这里")才能使其正常工作？

Answer 1

一般来说，您最好在rabbitmq-discuss邮件列表上发布有关rabbitmq的问题。我们不倾向于跟踪在此之外提出的问题。

您可以配置心跳，但默认情况下它是关闭的。您还可以打开 TCP Keep Alive。setRequestedHeartbeat在创建新连接之前调用ConnectionFactory，或者子类化ConnectionFactory，重写该configureSocket方法，然后调用socket.setKeepAlive(true)。两者都应该导致连接在网络中断时发出通知。