Search code examples
javaspringamqpspring-amqpspring-rabbit

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

I have a situation where I need to register queues dynamically in a run-time to a SimpleMessageListenerContainer. The problem I am experiencing is a deadlock that happens because of this:

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)

enter image description here

and this is the code that is problematic - here I try to setup client queues in onCreate callback in connectionListener.

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

Is there some standard (proper) way to easily register and create queues dynamically without causing this deadlock?

UPDATE

This is how I handle it now, after Garry suggestion:

  @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();
      }
    };
  }
Solution

  • It's better to implement SmartLifecycle and do the setup in start().

    The listener container default phase is Integer.MAX_VALUE so the containers are amongst the last things started by the application context.

    Put your SmartLifecycle in an earlier phase (e.g. 0) so the containers will be set up before starting.