Проблема добавления очередей с помощью SmartLifecycle во время выполнения

Question

Итак, это новый вопрос, связанный с этим

Я создал случай, когда ListenerContainer фактически выполняется до @Bean с фазой 0, даже если он имеет фазуInteger.MAX_INT. Вот код:

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.SmartLifecycle;
import org.springframework.context.annotation.Bean;
import org.springframework.context.support.GenericApplicationContext;

@SpringBootApplication
public class RuntimeRegisterApp {

  public static void main(String[] args) {
    SpringApplication.run(RuntimeRegisterApp.class, args);
  }

  @Bean
  public CachingConnectionFactory cachingConnectionFactory(){
    CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
    cachingConnectionFactory.setAddresses("10.10.121.199:35682");
    cachingConnectionFactory.setUsername("guest");
    cachingConnectionFactory.setPassword("guest");
    return cachingConnectionFactory;
  }

  @Bean
  public DirectRabbitListenerContainerFactory directFactory(ConnectionFactory cachingConnectionFactory) {
    DirectRabbitListenerContainerFactory factory = new DirectRabbitListenerContainerFactory();
    factory.setConnectionFactory(cachingConnectionFactory);
    return factory;
  }

  @RabbitListener(bindings = {
      @QueueBinding(value = @Queue(value = "SomeQueue", durable = "false", autoDelete = "true"),
          exchange = @Exchange(value = "SomeEX", ignoreDeclarationExceptions = "true"),
          key = "SomeKey")
  },
      containerFactory = "directFactory"
  )
  public void onMessage(String msg){
    System.out.println("Received " + msg);
  }

  public void onMessage2(String msg){
    System.out.println("Received 2 " + msg);
  }

  @Bean
  public org.springframework.amqp.core.Queue someQueue(){
    return QueueBuilder.nonDurable("some1").build();
  }

  @Bean
  public DirectMessageListenerContainer container(DirectRabbitListenerContainerFactory directFactory){
    DirectMessageListenerContainer container = directFactory.createListenerContainer();
    container.setConsumersPerQueue(2);
    container.setMessageListener(
        message -> System.out.println("BEAN CONTAINER " + message.toString()));
    container.setQueues(someQueue());
    return container;
  }

  @Bean
  public RabbitAdmin rabbitAdmin(){
    return new RabbitAdmin(cachingConnectionFactory());
  }

  @Bean
  SmartLifecycle dynamicRegister(GenericApplicationContext applicationContext,
      DirectMessageListenerContainer container,
      DirectRabbitListenerContainerFactory directFactory,
      RabbitAdmin rabbitAdmin){

    return new SmartLifecycle() {

      private boolean running;

      private void dynamicallySetQueues(){
        org.springframework.amqp.core.Queue q1 = QueueBuilder
            .nonDurable("mySomeQueue")
            .build();

        rabbitAdmin.declareQueue(q1);

        applicationContext.registerBean(org.springframework.amqp.core.Queue.class, () -> q1);

        List<String> queues = new ArrayList<>(Arrays.asList(container.getQueueNames()));
        queues.add(q1.getName());


        //THIS ONE WORKS SINCE WE USE FACTORY AND SET QUEUES BEFORE START
        DirectMessageListenerContainer container1 = directFactory.createListenerContainer();
        container1.setQueueNames(queues.toArray(new String[0]));
        container1.setMessageListener(message -> System.out.println("INNER CONTAINER" + message.toString()));
        container1.start();

        //THIS ONE WORKS SINCE WE ONLY ADD QUEUES
        container.addQueueNames(q1.getName());

        //SETTING QUEUES HERE FAILS, SINCE CONTAINER ALREADY RUNNING
        //BUT IT SHOULD NOT RUN, SINCE THIS IS PHASE 0 ?
        //I GUESS SINCE IT IS NEEDED HERE IT RUNS ANYWAY ?
        container.setQueueNames(queues.toArray(new String[0]));
      }

      @Override
      public void start() {
        dynamicallySetQueues();
        running = true;
      }

      @Override
      public void stop() {
        running = false;
      }

      @Override
      public int getPhase() {
        return 0; //return 0 so we add queues before ListenerContainer starts
      }

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

Я полагаю, что он работает, поскольку это фактическая зависимость SmartLifecycle bean. Единственный обходной путь, который я вижу здесь, - это setAutostart(false) в контейнере, а затем внутри SmartLifecycle вызов компонента container.start() после задания имен очереди.

Answer 1

Вы правы;процессор жизненного цикла запускает все зависимые bean-компоненты перед запуском текущего bean-компонента ...

        for (String dependency : dependenciesForBean) {
            doStart(lifecycleBeans, dependency, autoStartupOnly);
        }

... в действительности добавление зависимости уменьшает фазу зависимого bean-компонента.

Ваше решение, вероятно,Простейшие.

Answer 2

ОК, так что эта версия кажется мне лучшей реализацией:

public class QueueBindingRegistry implements SmartLifecycle, InitializingBean {

    private DefaultListableBeanFactory beanFactory;
    private AmqpAdmin amqpAdmin;

    private AbstractMessageListenerContainer container;
    private Supplier<Map<Queue, Set<Binding>>> queueBindingSupplier = Collections::emptyMap;

    private Map<Queue, Set<Binding>> queueBindings = new HashMap<>();
    private boolean running;

    public QueueBindingRegistry(AbstractMessageListenerContainer container) {
      //so we can set queues before container is started
      container.setAutoStartup(false);
      this.container = container;
    }

    @Autowired
    public void setBeanFactory(
        DefaultListableBeanFactory beanFactory) {
      this.beanFactory = beanFactory;
    }

    @Autowired
    public void setAmqpAdmin(AmqpAdmin amqpAdmin) {
      this.amqpAdmin = amqpAdmin;
    }

    public void setQueueBindingSupplier(
        Supplier<Map<Queue, Set<Binding>>> queueBindingSupplier) {
      this.queueBindingSupplier = queueBindingSupplier;
    }

    public void removeQueueNames(String... queueNames) {
      container.removeQueueNames(queueNames);
      Set<String> qNamesForRemoval = Set.of(queueNames);
      Set<Queue> qForRemoval = queueBindings.keySet().stream()
          .filter(q -> qNamesForRemoval.contains(q.getName()))
          .collect(Collectors.toSet());
      qForRemoval.forEach(queue ->
          queueBindings.computeIfPresent(queue, (q, qb) -> {
            qb.forEach(this::destroyBinding);
            destroyQueue(q);
            return null; //remove from registry
          }));
    }

    public void addQueuesAndBindings(Map<Queue, Set<Binding>> queuesAndBindings) {
      queuesAndBindings.entrySet().stream()
          //remove queues that already exist
          .filter(e -> queueBindings.keySet().stream()
              .noneMatch(exq -> exq.getName().equals(e.getKey().getName())))
          .forEach((e) -> {
            Queue q = e.getKey();
            Set<Binding> b = e.getValue();
            registerQueue(q);
            b.forEach(this::registerBinding);
            container.addQueues(q);
            queueBindings.put(q, b);
          });
    }

    private void destroyQueue(Queue queue) {
      beanFactory.destroySingleton(queue.getName());
      amqpAdmin.deleteQueue(queue.getName());
    }

    private void destroyBinding(Binding binding) {
      beanFactory.destroySingleton(getBindingBeanName(binding));
      amqpAdmin.removeBinding(binding);
    }

    private void initQueueBindings() {
      queueBindings.putAll(queueBindingSupplier.get());
      queueBindings.forEach((q, bs) -> {
        registerQueue(q);
        bs.forEach(this::registerBinding);
      });
      Queue[] queues = getAllQueues(queueBindings);
      container.setQueues(queues);
    }

    private Queue[] getAllQueues(Map<Queue, Set<Binding>> queueBindings) {
      return queueBindings.keySet().toArray(new Queue[0]);
    }

    private String getBindingBeanName(Binding binding) {
      return binding.getExchange() + "-" + binding.getRoutingKey() + "-" + binding.getDestination();
    }

    private void registerBinding(Binding binding) {
      amqpAdmin.declareBinding(binding);
      beanFactory.registerSingleton(getBindingBeanName(binding), binding);
    }

    private void registerQueue(Queue queue) {
      amqpAdmin.declareQueue(queue);
      beanFactory.registerSingleton(queue.getName(), queue);
    }

    @Override
    public void start() {
      initQueueBindings();
      //run after queues are initiated
      container.start();
      running = true;
    }

    @Override
    public void stop() {
      running = false;
    }

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

    @Override
    public void afterPropertiesSet() {
      Assert.isTrue(amqpAdmin != null, "AmqpAdmin bean must be set.");
      Assert.isTrue(beanFactory != null, "BeanRegistry bean must be set.");
      Assert.isTrue(queueBindingSupplier != null, "QueueBindingSupplier cannot be null.");
    }
  }

и регистрация для динамической регистрации проста:

@Bean
  public QueueBindingRegistry queueBindingRegistry(DirectMessageListenerContainer directContainer) {
    QueueBindingRegistry queueBindingRegistry = new QueueBindingRegistry(directContainer);

    Queue q1 = QueueBuilder
        .nonDurable("a1")
        .build();
    Queue q2 = QueueBuilder
        .nonDurable("q2")
        .build();

    queueBindingRegistry.setQueueBindingSupplier(() -> Map.of(
        q1,
        Set.of(
            BindingBuilder.bind(q1).to(new TopicExchange("someEX1")).with("k1"),
            BindingBuilder.bind(q1).to(new TopicExchange("someEX2")).with("k2")
        ),
        q2,
        Set.of(
            BindingBuilder.bind(q2).to(new TopicExchange("someEX1")).with("z1"),
            BindingBuilder.bind(q2).to(new TopicExchange("someEX2")).with("z2")
        )
    ));
    return queueBindingRegistry;
  }

, и если вы хотите удалитьочереди и привязки во время выполнения:

public void testRemoval() {
    System.out.println("Removing queue from registry");
    queueBindingRegistry.removeQueueNames("q2");

    Queue q2 = QueueBuilder
        .nonDurable("q2")
        .build();
    var s = Map.of(
        q2,
        Set.of(
            BindingBuilder.bind(q2).to(new TopicExchange("someEX1")).with("xxxx"),
            BindingBuilder.bind(q2).to(new TopicExchange("someEX2")).with("xxx2")
        ));

    queueBindingRegistry.addQueuesAndBindings(s);
  }