在运行时使用 SmartLifecycle 添加队列时出现问题 [英] Problem adding queues using SmartLifecycle at runtime

问题描述

所以，这是与 这个

我创建了一个案例，其中 ListenerContainer 实际上在 @Bean 之前以阶段 0 运行，即使它具有阶段 Integer.MAX_INT.代码如下:

I have created a case where ListenerContainer is actually run before @Bean with phase 0, even though it has phase Integer.MAX_INT. Here is the code:

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) 在 Container 上，然后在 SmartLifecycle bean 内部调用 container.start() 在设置队列名称.

I guess that it is running since it is actual dependency of SmartLifecycle bean. Only workaround I can see here is to setAutostart(false) on Container, and then inside SmartLifecycle bean call container.start() after setting the queue names.

推荐答案

你说得对；生命周期处理器在启动当前 bean 之前启动任何依赖 bean...

You are correct; the lifecycle processor starts any dependent beans before starting the current bean...

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

...实际上，添加依赖会减少依赖 bean 的阶段.

...in effect, adding the dependency decreases the dependent bean's phase.

您的解决方案可能是最简单的.

Your solution is probably the simplest.

这篇关于在运行时使用 SmartLifecycle 添加队列时出现问题的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！