Spring AMQP - 在启动时将队列注册到容器的正确方法

Question

我遇到一种情况，我需要在运行时间内将队列动态注册到 SimpleMessageListenerContainer。我遇到的问题是因为这个而发生的死锁：

Thread: [52] Thread1 wants the lock java.lang.Object@5537e0df
org.springframework.amqp.rabbit.connection.CachingConnectionFactory.getDeferredCloseExecutor(CachingConnectionFactory.java:907)
org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.restart(SimpleMessageListenerContainer.java:739)


Thread: [183] Thread2 wants the lock java.lang.Object@556fa9d6
org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.queuesChanged(SimpleMessageListenerContainer.java:689)
org.springframework.amqp.rabbit.connection.CachingConnectionFactory.createConnection(CachingConnectionFactory.java:634)
org.springframework.amqp.rabbit.connection.CachingConnectionFactory.createBareChannel(CachingConnectionFactory.java:578)

这是有问题的代码 - 这里我尝试在 connectionListener.

的 onCreate 回调中设置客户端队列

connectionFactory
        .addConnectionListener(
            new ConnectionListener() {
              @Override
              public void onCreate(Connection connection) {
                setupClientQueues(); ----> will call container.setQueueNames which will result in calling queuesChanged
              }

              @Override
              public void onClose(Connection connection) {
                // nothing to do
              }
            });

是否有一些标准（正确）的方法可以轻松地动态注册和创建队列而不会导致这种死锁？

更新

根据 Garry 的建议，我现在是这样处理的：

  @Bean
  public SmartLifecycle containerQueueSetter(){
    return new SmartLifecycle(){
      private boolean running;

      @Override
      public int getPhase() {
        return 0;
      }

      @Override
      public void start() {
        //CREATE QUEUES HERE - since I create and register them as beans,
        //it will work even when rabbit is reconnected
        //REGISTER QUEUES TO SIMPLE_MESSAGE_LISTENER_CONTAINER
        running = true;
      }

      @Override
      public void stop() {
        log.info("Stopping dynamic queue registerer.");
        running = false;
      }

      @Override
      public boolean isRunning() {
        return running;
      }

      @Override
      public boolean isAutoStartup() {
        return true;
      }

      @Override
      public void stop(Runnable callback) {
        stop();
        callback.run();
      }
    };
  }

Answer 1

最好实施 SmartLifecycle 并在 start() 中进行设置。

侦听器容器默认 phase 是 Integer.MAX_VALUE 因此容器是应用程序上下文最后启动的容器之一。

将您的 SmartLifecycle 置于较早的阶段（例如 0），以便在开始之前设置容器。

Spring AMQP - 在启动时将队列注册到容器的正确方法

Spring AMQP - Proper way to register queues to container on startup

java

spring

amqp

spring-amqp

spring-rabbit