java - @RabbitListener 以 "stopped"模式开始绑定(bind)到扇出交换

Question

我想在应用程序开始收集消息时将匿名队列绑定(bind)到扇出交换，但消息的实际处理应该稍后完成(在其他地方进行一些初始化之后)。我尝试过:

@RabbitListener(autoStartup="false",
                    bindings = @QueueBinding(value = @Queue,
                    exchange = @Exchange(name="myexchange",
                                         type=ExchangeTypes.FANOUT)))
public void processMessage(String message) {
}

但是autoStartup="false"不会将(匿名)队列绑定(bind)到交换器。

换句话说，我需要的是应用程序启动后立即绑定(bind)到交换器的匿名队列，并仅在稍后开始读取消息。@RabbitListener 可以吗？

更新:尝试声明队列和交换，但队列不会添加到rabbit，除非我还为其声明RabbitListener:

@Configuration
public class AmqpConfig {
    @Bean
    RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }
    @Bean
    public FanoutExchange fanout() {
        return new FanoutExchange("myexchange");
    }
    private static class ReceiverConfig {
        @Bean
        public Queue myQueue() {
            return new AnonymousQueue();
        }
        @Bean
        public Binding binding(FanoutExchange fanout, Queue myQueue) {
            return BindingBuilder.bind(myQueue).to(fanout);
        }
}

除非我还添加@RabbitListener，否则它不会创建队列:

@Component
public class AmqpReceiver {
    @RabbitListener(queues = "#{myQueue.name}")
    public void receive(String in) throws InterruptedException {
    }
}

Answer 1

由于您没有启动监听器，因此它不会打开连接。

只要您在应用程序上下文中将队列、绑定(bind)和 RabbitAdmin 定义为 beans，您所需要做的就是强制打开连接(管理员监听新连接)并执行声明)。

只需在 CachingConnectionFactory 上调用 createConnection() 即可。

编辑

@SpringBootApplication
public class So49401150Application {

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

    @Bean
    ApplicationRunner runner(ConnectionFactory cf, RabbitTemplate template,
            RabbitListenerEndpointRegistry registry) {
        return args -> {
            cf.createConnection().close(); // Admin does declarations here
            template.convertAndSend("myexchange", "", "foo");
            Thread.sleep(30_000);
            registry.start();
        };
    }

    @RabbitListener(autoStartup="false",
            bindings = @QueueBinding(value = @Queue,
            exchange = @Exchange(name="myexchange",
                                 type=ExchangeTypes.FANOUT)))
    public void processMessage(String message) {
        System.out.println(message);
    }

}