java - Springboot RabbitMQ 未接收匿名队列

Question

我正在尝试创建一个扇出交换，每个人都会收到来自发布者的消息。我的问题是监听器无法接收队列中发布的消息。设置的队列都是随应用程序实例消亡的匿名队列。发布者和订阅者位于同一应用程序中。非常感谢任何帮助。

队列配置:

@Value("${apcp.rabbitmq.refresh-exchange}")
private String fanoutExchangeName;

@Autowired
Queue anonQueue; 
@Bean("amqp-admin")
@PostConstruct
public AmqpAdmin AMQPAdmin(){
    log.info(connectionFactory.toString());
    AmqpAdmin amqpAdmin = new RabbitAdmin(connectionFactory);
    return amqpAdmin;
}
@Bean
@PostConstruct
public String initRefreshAmqp(){
    setupFanOutExchange();
    return "";
}
public void setupFanOutExchange(){
    AmqpAdmin amqpAdmin =  new RabbitAdmin(connectionFactory);
    FanoutExchange exchange = new FanoutExchange(fanoutExchangeName);
    amqpAdmin.declareExchange(exchange);
    Queue queue = new Queue(anonQueue, false, true, true);
    amqpAdmin.declareQueue(queue);
    amqpAdmin.declareBinding(BindingBuilder.bind(queue).to(exchange));
}

出版商

@RequestMapping(value = "/publish")
public String publish(String message){
    rabbitTemplate.convertAndSend(exchangeName, message);
    return "";
}

订阅者配置

@Bean
@PostConstruct
public SimpleRabbitListenerContainerFactory listenerFactory() {
    log.info("CONNECTIONS:"+connectionFactory.toString());
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setMessageConverter(jsonMessageConverter());
    return factory;
}

订阅监听器

@RabbitListener(queues = "#{anonQueue.name}", containerFactory = "listenerFactory")
public void receiverQueue(String message){
    log.info(message);
}

Answer 1

1)没有这样的方法:

rabbitTemplate.convertAndSend(exchangeName, message);

两个参数的方法是

public void convertAndSend(String routingKey, final Object object) throws AmqpException {

所以经纪人正在删除您的消息。

2) 您不得在 bean 定义中调用管理方法(或执行任何涉及代理的操作)

3)您的配置比需要的要复杂得多。

这很好用...

@SpringBootApplication
public class So49854747Application {

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

    @Bean
    public ApplicationRunner runner(RabbitTemplate template, FanoutExchange exchange) {
        return args -> {
            template.convertAndSend(exchange.getName(), "", "foo");
            Thread.sleep(10_000);
        };
    }

    @Bean
    public Queue anonQueue() {
        return new AnonymousQueue();
    }

    @Bean
    public FanoutExchange exchange() {
        return new FanoutExchange("so49854747");
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(anonQueue()).to(exchange());
    }

    @RabbitListener(queues = "#{anonQueue.name}")
    public void listen(String in) {
        System.out.println(in);
    }

}

.

2018-04-16 09:01:54.620  INFO 50389 --- [           main] com.example.So49854747Application        : Started So49854747Application in 1.407 seconds (JVM running for 1.909)
foo