删除 RabbitMQ 消费者并在浏览器的 RabbitMQ 控制台中查看答案

【问题标题】：Remove RabbitMQ consumers and see it in Browser's RabbitMQ Console删除 RabbitMQ 消费者并在浏览器的 RabbitMQ 控制台中查看
【发布时间】：2021-02-12 15:38:21
【问题描述】：

免责声明：RabbitMq 和/或 Spring Integration 和/或 Spring Cloud Stream 中的 Noobie。

我有以下课程：

@Component
public class RabbitMQChannelBindingFactory {

...

  private org.springframework.cloud.stream.binder.rabbitRabbitMessageChannelBinder binder;
  private org.springframework.cloud.stream.config.BindingServiceProperties bindingServiceProperties;
  private org.springframework.cloud.stream.binding.BindingService bindingService;
  private org.springframework.beans.factory.config.ConfigurableListableBeanFactory beanFactory;
  private org.springframework.cloud.stream.binding.SubscribableChannelBindingTargetFactory bindingTargetFactory;
  private org.springframework.amqp.rabbit.connection.ConnectionFactory rabbitConnectionFactory;
...
}

需要什么？

我有一个创建 Exchange+Queue+Consumer 的机制，并且我有一个销毁这些的机制。交换和队列的自动删除设置为 true。

有什么问题？

销毁所有这 3 个元素的继承机制不起作用。

它只删除 Exchange。队列没有被删除，因为它仍然有一个消费者，我也可以在我的应用程序中看到它。

尝试过什么？

我尝试使用 JVisualVM 获取客户标签的字符串实例，然后我沿着层次结构移除消费者。

我已经更改了我的应用程序中的 org.springframework.amqp.rabbit.listener.BlockingQueueConsumer，以便类加载器首先加载它。

我在里面添加了这样的内容，以便跟踪在我的应用程序中创建的所有消费者：

public class BlockingQueueConsumer {
  ...
  public static List<BlockingQueueConsumer> all = new ArrayList<>();
  public BlockingQueueConsumer(...) {
     ...
     all.add(this);
     ...
  }
  ...
}

完成上一步后，我在 RabbitMQChannelBindingFactory 中添加了另一个方法类为所有消费者调用取消方法，如下所示：

class RabbitMQChannelBindingFactory {

   public void disconnect(...) {
      BlockingQueueConsumer lastBlockingQueueConsumer = 
          BlockingQueueConsumer.all.get(BlockingQueueConsumer.all.size() - 1);
      lastBlockingQueueConsumer.getConsumerTags()
          .forEach(consumerTag -> basicCancel(lastBlockingQueueConsumer, consumerTag));
   }
}

此时在加载了 RabbitMQ 控制台的浏览器上，我们可以看到队列被删除（除了 Exchange 和 Consumer）。

有什么问题？

我找不到将 BlockingQueueConsumer 连接到自动装配属性的方法。

比如我试过

  public void deleteRabbitMQConsumer() {
    RabbitTemplate rabbitTemplate = new RabbitTemplate(rabbitConnectionFactory);
    rabbitTemplate.execute(channel -> {
      if (channel instanceof ChannelN) {
        ChannelN channelN = (ChannelN) channel;
        return true;
      }
      return false;
    });
  }

但 ChannelN 内似乎没有消费者。

你能给我一个方向吗？首先需要了解什么？

或者有什么可以帮助我的资源？

或者任何尝试过使用此自动装配属性取消消费者的操作的人？

或者我需要添加其他自动装配的属性吗？

我已经尝试过https://stackoverflow.com/a/27633771/13622666 解决方案。

解决方案

@Component
public class RabbitMQChannelBindingFactory {

...

  private org.springframework.cloud.stream.binder.rabbitRabbitMessageChannelBinder binder;

  private void connectAndDisconnectConsumer(...) {
      ...
      Binding<MessageChannel> messageChannelBinding = 
         binder.bindConsumer(exchangeName, "", channel, consumerProperties);
      ... // receive messages
      messageChannelBinding.stop(); 
     ...
   }  

}

还有堆栈跟踪：

messageChannelBinding.stop();
 DefaultBinding#stop
  AbstractEndpoint#stop()
   AmqpInboundChannelAdapter#doStop
    AbstractMessageListenerContainer#stop()
     AbstractMessageListenerContainer#doStop
      AbstractMessageListenerContainer#shutdown
      SimpleMessageListenerContainer#doShutdown
       BlockingQueueConsumer#basicCancel(boolean)

【问题讨论】：

太棒了！你到底在做什么？您不应该像那样直接与BlockingQueueConsumer 交互，它是由绑定创建的默认SimpleMessageListenerContainer 使用的类。您只需要stop() 绑定即可。
感谢两位解决方案 stop() messageChannelBinding。如果您能推荐一些阅读 Spring Cloud Stream 生命周期的地方，我将不胜感激。

标签： java rabbitmq spring-integration spring-cloud-stream spring-rabbit

【解决方案1】：

投票结束。你不能像那样滥用 Java 类系统，而是更专注于学习你使用的库。可能有人已经询问过您正在寻找的解决方案。正如 Gary 所说：Spring Cloud Stream 绑定中只有 stop()，这将停止 MessageListenerContainer，而 MessageListenerContainer 反过来将取消它在队列中的所有消费者。您的自动删除队列将从 RabbitMQ 中删除。没有理由破坏交易所。虽然你可以通过AmqpAdmin.deleteExchange() 做到这一点。

【讨论】：