带RabbitMQ的Spring Cloud Spring服务连接器

Question

我使用Spring云Spring服务连接器连接CloudFoundry上的Rabbitmq服务。

public class CloudConfig extends AbstractCloudConfig {

    @Bean
    public ConnectionFactory rabbitFactory()
    {
         return connectionFactory().rabbitConnectionFactory();
    }
}

但我需要声明一个CachingConnectionFactory并将其PublisherConfirms设置为true。因为我们需要使用PublisherConfig在向队列发送消息时检查确认。我不知道如何注入从cloud spring服务连接器获得的connectionFactory。或者我们如何处理这种情况。

Answer 1

这是RabbitTemplate

@Bean
public RabbitTemplate rabbitTemplate() {
    RabbitTemplate template = new RabbitTemplate(connectionFactory);
    template.setMandatory(true);
    template.setMessageConverter(new Jackson2JsonMessageConverter());
    template.setConfirmCallback((correlationData, ack, cause) -> {
        if (!ack) {
            System.out.println("send message failed: " + cause + correlationData.toString());
        } else {
            System.out.println("Publisher Confirm" + correlationData.toString());
        }
    });
    return template;
}

这是spring云配置：

@Bean
public ConnectionFactory rabbitConnectionFactory() {
    Map<String, Object> properties = new HashMap<String, Object>();
    properties.put("publisherConfirms", true);
    RabbitConnectionFactoryConfig rabbitConfig = new RabbitConnectionFactoryConfig(properties);
    return connectionFactory().rabbitConnectionFactory(rabbitConfig);
}

当我使用此发件人发送消息时。结果是不期望的。

@Component
public class TestSender {

@Autowired
private RabbitTemplate rabbitTemplate;

@Scheduled(cron = "0/5 * *  * * ? ")
public void send() {
System.out.println("===============================================================");
    this.rabbitTemplate.convertAndSend(EXCHANGE, "routingkey", "hello world",
            (Message m) -> {
                m.getMessageProperties().setHeader("tenant", "aaaaa");
                return m;
            }, new CorrelationData(UUID.randomUUID().toString()));
    Date date = new Date();
    System.out.println("Sender Msg Successfully - " + date);
}

}

Answer 2

您可以按如下方式重新配置连接器创建的CCF：

@Bean
public SmartInitializingSingleton factoryConfigurer() {
    return new SmartInitializingSingleton() {

        @Autowired
        private CachingConnectionFactory connectionFactory;

        @Override
        public void afterSingletonsInstantiated() {
            this.connectionFactory.setPublisherConfirms(true);
        }
    };
}

在应用程序上下文完全初始化之前，必须确保不执行任何RabbitMQ操作（无论如何，这是最佳实践）。

Answer 3

该文档包括自定义连接器提供的连接细节的示例。

在你的情况下，你应该能够做这样的事情：

@Bean
public RabbitConnectionFactory rabbitFactory() {
    Map<String, Object> properties = new HashMap<String, Object>();
    properties.put("publisherConfirms", true);

    RabbitConnectionFactoryConfig rabbitConfig = new RabbitConnectionFactoryConfig(properties);
    return connectionFactory().rabbitConnectionFactory(rabbitConfig);
}