【问题标题】:Spring AMQP - Publisher confirms returns ack-ed when publish to non-existent queueSpring AMQP - 发布者在发布到不存在的队列时确认返回确认
【发布时间】:2019-11-28 21:40:56
【问题描述】:

如果我在 RabbitMQ 上使用发布者确认并返回回调,我永远不会收到返回的消息。 来自 Spring AMQP 文档:

- Publish to an exchange but there is no matching destination queue.
- Publish to a non-existent exchange.
The first case is covered by publisher returns, as described in Publisher Confirms and Returns.

所以我想如果我发布到存在的交换,但不存在的队列,我会收到返回消息。但是返回回调从未调用过。 我需要设置其他东西吗?
我正在使用 RabbitMQ 3.8.0 和 Spring Boot 2.2.1

application.yml

spring:
  rabbitmq:
    publisher-confirms: true
    publisher-returns: true
    template:
      mandatory: true

制片人

@Service
public class PublisherConfirmProducer {

    private static final Logger log = LoggerFactory.getLogger(PublisherConfirmProducer.class);

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct
    private void postConstruct() {
        this.rabbitTemplate.setConfirmCallback((correlation, ack, reason) -> {
            if (correlation != null) {
                log.info("Received " + (ack ? " ack " : " nack ") + "for correlation: " + correlation);
            }
        });

        this.rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            log.info("Returned: " + message + "\nreplyCode: " + replyCode + "\nreplyText: " + replyText
                    + "\nexchange/rk: " + exchange + "/" + routingKey);
        });
    }

    // Careful : will be silently dropped, since the exchange is exists, but no
    // route to queue, but ack-ed. How to know that I publish to non-existing queue?
    public void sendMessage_ValidExchange_InvalidQueue(DummyMessage message) {
        CorrelationData correlationData = new CorrelationData("Correlation for message " + message.getContent());
        this.rabbitTemplate.convertAndSend("x.test", "not-valid-routing-key", message, correlationData);
    }

}

主应用

@SpringBootApplication
public class RabbitmqProducerTwoApplication implements CommandLineRunner {

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

    @Autowired
    private PublisherConfirmProducer producer;

    @Override
    public void run(String... args) throws Exception {
        var dummyMessage_2 = new DummyMessage("Message 2", 2);
        producer.sendMessage_ValidExchange_InvalidQueue(dummyMessage_2);
    }

}

日志结果

2019-11-29 04:45:23.796  INFO 8352 --- [           main] c.c.r.RabbitmqProducerTwoApplication     : Starting RabbitmqProducerTwoApplication on timpamungkas with PID 8352 (D:\workspace\eclipse\my-courses\rabbitmq-1.2\rabbitmq-producer-two\bin\main started by USER in D:\workspace\eclipse\my-courses\rabbitmq-1.2\rabbitmq-producer-two)
2019-11-29 04:45:23.800  INFO 8352 --- [           main] c.c.r.RabbitmqProducerTwoApplication     : No active profile set, falling back to default profiles: default
2019-11-29 04:45:24.952  INFO 8352 --- [           main] c.c.r.RabbitmqProducerTwoApplication     : Started RabbitmqProducerTwoApplication in 1.696 seconds (JVM running for 3.539)
2019-11-29 04:45:24.990  INFO 8352 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [localhost:5672]
2019-11-29 04:45:25.024  INFO 8352 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#599f571f:0/SimpleConnection@86733 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 50688]
2019-11-29 04:45:25.058  INFO 8352 --- [nectionFactory1] c.c.r.producer.PublisherConfirmProducer  : Received  ack for correlation: CorrelationData [id=Correlation for message Message 2]

为加里编辑

RabbitMqConfig.java

@Configuration
public class RabbitmqConfig {

    @Bean
    public Jackson2JsonMessageConverter converter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, Jackson2JsonMessageConverter converter) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMessageConverter(converter);
        return template;
    }

}

【问题讨论】:

    标签: spring-boot spring-amqp spring-rabbit


    【解决方案1】:

    返回消息后,您将收到肯定的确认。

    你所拥有的对我来说是正确的。它与this sample 非常相似,因此应该以相同的方式工作。

    x.test是什么类型的交换?它绑定了哪些队列,以及使用哪些路由键?

    如果您在查看该示例后仍然无法使其工作,请将项目发布到某个地方,我会看看。

    【讨论】:

    • x.test是什么类型的交换?它绑定了哪些队列,以及使用哪些路由键?
    • 嗨,加里。 “x.test”是直接交换,用路由键“test”绑定到“q.test”。经过一番尝试,我可以让它工作。我有一个返回 RabbitTemplate 的 RabbitMQConfig.java(参见上面的代码)。为了使返回回调起作用,我必须在我的配置中注释返回 RabbitTemplate 的 Bean。有趣的是,如果我不评论 Bean,则会触发确认回调,但不会触发返回回调。这是预期的行为吗?顺便说一句,我真的很佩服你为 spring-amqp 所做的一切!它真的对我帮助很大!谢谢 3000 @gary-russell
    • 如果你不介意看一下,我已经输入了代码here。非常感谢
    • spring.rabbitmq.template.* 属性由 Boot 在自动配置 RabbitTemplate 时使用。由于您有自己的 bean,因此禁用了自动配置,因此不应用 mandatory 属性;当您定义自己的模板 bean 时,您需要自己设置 mandatory 属性。此外,Boot 会检测您的消息转换器并自动将其连接到模板中,因此您真的不需要自己的 bean。
    【解决方案2】:

    实现看起来是正确的,只需实现自己的Private回调类并扩展ConfirmCallback来注册回调,在初始化RabbitTemplate时设置ConfirmCallback。

     private static class PublisherCallback implements RabbitTemplate.ConfirmCallback {
        
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            System.out.println("Confirm message returned " + correlationData.toString() + " Ack " + ack + " cause " + cause);
        }
    
    }
    
    rabbitTemplate.setConfirmCallback(new PublisherCallback());
    

    【讨论】:

      【解决方案3】:

      我使用 Springboot 2.5.2 和 RabbitMQ 3.8.18,当我使用无效路由键将队列绑定到主题交换时,我的 ReturnsCallback 被调用

      @Bean(name = "binding")
      
      Binding binding(@Qualifier("queue") Queue queue, @Qualifier("exchange") TopicExchange exchange) {
      
      return BindingBuilder.bind(queue).to(exchange).with("aaa"+ Constants.VALID_ROUTING_KEY);
      }
      
      //configuration
      rabbitTemplate.setConfirmCallback(confirmCallbackService);
      rabbitTemplate.setReturnsCallback(returnCallbackService);
      rabbitTemplate.setMandatory(true); //seems this config is mandatory
      
      
      //logs
      2021-07-08 15:08:05,078 ERROR [connectionFactory1] com..publisher.config.ReturnCallbackService: returnedMessage =>
      
      

      Rabbitmq-Batch-Rabbitmq-Publish-Subscribe

      【讨论】:

        猜你喜欢
        • 1970-01-01
        • 1970-01-01
        • 1970-01-01
        • 2020-02-01
        • 1970-01-01
        • 1970-01-01
        • 1970-01-01
        • 2020-03-13
        • 2013-11-16
        相关资源
        最近更新 更多