java - RabbitMQ 测试消费者处于 Activity 状态

Question

我有一个JAVA应用程序，它创建监听rabbitmq的消费者。我需要知道启动的消费者仍然工作正常，如果不是那么我需要重新启动消费者。

他们有什么办法我可以做到这一点吗？目前，我的主应用程序创建一个执行程序线程池，并在创建新连接时传递该执行程序。

ExecutorService executor = Executors.newFixedThreadPool(30);
Connection connection = factory.newConnection(executor);

然后，main方法通过以新 channel 为参数调用构造函数来创建30个consumerApp对象，并调用listen()方法

for(int i=0;i<=30;i++) {
 ConsumerApp consumer = new ConsumerApp(i,connection.createChanell());
 consumer.listen() }

consumerApp中的listen方法监听队列并启动一个DefaultConsumer对象，它只是打印接收到的消息

listen() { 
  try {
  channel.queueDeclare("test-queue-name", false, false, false, null);
  }
catch {
  System.out.println("Exception on creating Queue")
  }
Consumer consumer = new DefaultConsumer(this.channel) {
@Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                byte[] body) throws IOException {
            String message = new String(body, "UTF-8");
            System.out.println(" [x] Received Message in consumer '"+consumerId+" "+  message + "'");
        }
    };
 //Now starting the consumer
  try {
    channel.basicConsume(QUEUE_NAME, true, consumer);   
   }
    catch (ShutdownSignalException | IOException ex) {
        ex.printStackTrace();
  }
 }

我想知道他们有什么方法可以检查消费者是否活跃。我的想法是捕获关闭信号异常并重新创建消费者对象并调用监听方法。这是必要的，因为rabbitmq自动恢复并连接回来。 ？但我怎样才能确保这一点呢？

使用传递给rabbitmq连接器的线程池可以通过任何方式实现这一点吗？

我使用的是最新版本的rabbitmq客户端5.3.0

Answer 1

消费者有 different methods这可以帮助您跟踪消费者的状态。您可能对 handleConsumeOk 和 handleCancel 感兴趣。 Automatic connection recovery确实会在连接失败后重新注册消费者，但这并不妨碍您手动跟踪它们的状态，例如公开一些关于 JMX 的信息。